[pypy-svn] r40853 - in pypy/dist: demo pypy/lib pypy/lib/test2
hpk at codespeak.net
hpk at codespeak.net
Tue Mar 20 18:49:50 CET 2007
Author: hpk
Date: Tue Mar 20 18:49:49 2007
New Revision: 40853
Modified:
pypy/dist/demo/tp-persistence.py
pypy/dist/pypy/lib/test2/test_tputil.py
pypy/dist/pypy/lib/tputil.py
Log:
simplified tputil support to provide a "make_proxy" helper
that will pass a somewhat nice Invocation object to the provided
callback. Thus got rid of having to subclass etc.
Adapated demo and added another test.
Modified: pypy/dist/demo/tp-persistence.py
==============================================================================
--- pypy/dist/demo/tp-persistence.py (original)
+++ pypy/dist/demo/tp-persistence.py Tue Mar 20 18:49:49 2007
@@ -5,32 +5,24 @@
"""
from pypymagic import transparent_proxy, get_transparent_controller
-from tputil import BaseDispatcher
+from tputil import make_proxy
-class PListDispatcher(BaseDispatcher):
- _changeops = ('__iadd__ __imul__ __delitem__ __setitem__ '
+list_changeops = ('__iadd__ __imul__ __delitem__ __setitem__ '
'__delslice__ __setslice__ '
'append extend insert pop remove reverse sort').split()
- def __init__(self, realobj, storage):
- parent = super(PListDispatcher, self)
- parent.__init__(realobj, list)
- self._storage = storage
-
- def op_default(self, realmethod, *args, **kwargs):
- res = realmethod(*args, **kwargs)
- if realmethod.__name__ in self._changeops:
- self.persist()
- return res
-
- def persist(self):
- self._storage.dump(self.realobj)
-
- @classmethod
- def load(cls, storage):
- obj = storage.load()
- return cls(obj, storage)
-
+def make_plist(instance, storage):
+ def perform(invocation):
+ res = invocation.perform()
+ if invocation.opname in list_changeops:
+ storage.dump(instance)
+ return res
+ return make_proxy(instance, perform, typ=list)
+
+def get_plist(storage):
+ obj = storage.load()
+ return make_plist(obj, storage)
+
def work_with_list(mylist):
assert isinstance(mylist, list)
assert mylist.__class__ is list
@@ -41,21 +33,19 @@
import py
storage = py.path.local("/tmp/mystorage")
- somelist = [1,2,3]
- newlist = PListDispatcher(somelist, storage).proxyobj
-
+ plist = make_plist([1,2,3], storage)
# here we may call into application code which can
- # not detect easily that it is dealing with a persistent
- # object
- work_with_list(newlist)
- del somelist, newlist
+ # not detect easily that it is dealing with a
+ # transparently persistent list
+ work_with_list(plist)
+ del plist
- restoredlist = PListDispatcher.load(storage).proxyobj
+ restoredlist = get_plist(storage)
print "restored list", restoredlist
assert restoredlist == [1,2,3,4,5,6,7]
restoredlist *= 2
del restoredlist
- restoredlist = PListDispatcher.load(storage).proxyobj
+ restoredlist = get_plist(storage)
print "restored list 2", restoredlist
assert restoredlist == [1,2,3,4,5,6,7] * 2
Modified: pypy/dist/pypy/lib/test2/test_tputil.py
==============================================================================
--- pypy/dist/pypy/lib/test2/test_tputil.py (original)
+++ pypy/dist/pypy/lib/test2/test_tputil.py Tue Mar 20 18:49:49 2007
@@ -3,34 +3,39 @@
class AppTestTPListproxy:
def setup_class(cls):
cls.space = gettestobjspace(**{"objspace.std.withtproxy": True})
- cls.w_BaseDispatcher = cls.space.appexec([],"""
- ():
- from tputil import BaseDispatcher
- return BaseDispatcher
- """)
def test_listproxy_basic(self):
- x = []
- wrapper = self.BaseDispatcher(x)
- assert wrapper.realobj is x
- l = wrapper.proxyobj
- assert type(l) is list
+ from tputil import make_proxy
+ record = []
+ def func(invocation):
+ record.append(invocation)
+ return invocation.perform()
+ l = make_proxy([], func)
l.append(1)
- l.extend([2,3])
- assert l == [1,2,3]
- assert x == l
-
- def test_listproxy_getattribute(self):
- disp = self.BaseDispatcher([])
- meth = disp.proxyobj.append
- assert meth.im_self == disp.proxyobj
- meth = disp.proxyobj.__getattribute__
- assert meth.im_self == disp.proxyobj
+ assert len(record) == 2
+ i1, i2 = record
+ assert i1.opname == '__getattribute__'
+ assert i2.opname == 'append'
- def test_listproxy_hook(self):
- class MyBaseDispatcher(self.BaseDispatcher):
- def op___getitem__(self, *args, **kwargs):
- return 42
- l = MyBaseDispatcher([]).proxyobj
- assert l[0] == 42
- assert l[-1] == 42
+ def test_proxy_double(self):
+ from tputil import make_proxy
+ r1 = []
+ r2 = []
+ def func1(invocation):
+ r1.append(invocation)
+ return invocation.perform()
+ def func2(invocation):
+ r2.append(invocation)
+ return invocation.perform()
+
+ l = make_proxy([], func1)
+ l2 = make_proxy(l, func2)
+ assert not r1 and not r2
+ l2.append
+ assert len(r2) == 1
+ assert r2[0].opname == '__getattribute__'
+ assert len(r1) == 2
+ assert r1[0].opname == '__getattribute__'
+ assert r1[0].args[0] == '__getattribute__'
+ assert r1[1].opname == '__getattribute__'
+ assert r1[1].args[0] == 'append'
Modified: pypy/dist/pypy/lib/tputil.py
==============================================================================
--- pypy/dist/pypy/lib/tputil.py (original)
+++ pypy/dist/pypy/lib/tputil.py Tue Mar 20 18:49:49 2007
@@ -10,30 +10,32 @@
from pypymagic import transparent_proxy
from types import MethodType
-class BaseDispatcher(object):
- def __init__(self, realobj, typ=None):
- self.realobj = realobj
- if typ is None:
- typ = type(realobj)
- self.proxyobj = transparent_proxy(typ, self.invoke)
+def make_proxy(instance, invokefunc=None, typ=None):
+ if typ is None:
+ typ = type(instance)
+ def perform(opname, *args, **kwargs):
+ invocation = Invocation(tp, instance, opname, args, kwargs)
+ return invokefunc(invocation)
+ tp = transparent_proxy(typ, perform)
+ return tp
- def invoke(self, operation, *args, **kwargs):
- """ return result from dispatching to proxied operation. """
- realmethod = getattr(self.realobj, operation)
- print "operation", operation
- dispmethod = getattr(self, "op_" + operation, None)
- if dispmethod is None:
- dispmethod = self.op_default
- res = dispmethod(realmethod, *args, **kwargs)
- return res
+class Invocation(object):
+ def __init__(self, proxyobj, realobj, opname, args, kwargs):
+ self.proxyobj = proxyobj
+ self.realobj = realobj
+ self.opname = opname
+ self.args = args
+ self.kwargs = kwargs
+ self.realmethod = getattr(realobj, opname)
- def op___getattribute__(self, realmethod, *args, **kwargs):
- res = realmethod(*args, **kwargs)
- if (isinstance(res, MethodType) and
- res.im_self is self.realobj):
- res= MethodType(res.im_func, self.proxyobj, res.im_class)
- return res
-
- def op_default(self, realmethod, *args, **kwargs):
- return realmethod(*args, **kwargs)
+ def perform(self):
+ res = self.realmethod(*self.args, **self.kwargs)
+ if self.opname == "__getattribute__":
+ if (isinstance(res, MethodType) and
+ res.im_self is self.realobj):
+ res = MethodType(res.im_func, self.proxyobj, res.im_class)
+ return res
+ def __repr__(self):
+ return "<Invocation %s(*%r, **%r)" %(self.realmethod,
+ self.args, self.kwargs)
More information about the Pypy-commit
mailing list