Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def netobtain(obj):
"""
"""
if isinstance(obj, rpyc.core.netref.BaseNetref):
return rpyc.utils.classic.obtain(obj)
else:
return obj
def _box(self, obj):
"""store a local object in such a way that it could be recreated on
the remote party either by-value or by-reference"""
if brine.dumpable(obj):
return consts.LABEL_VALUE, obj
if type(obj) is tuple:
return consts.LABEL_TUPLE, tuple(self._box(item) for item in obj)
elif isinstance(obj, netref.BaseNetref) and obj.____conn__() is self:
return consts.LABEL_LOCAL_REF, obj.____oid__
else:
self._local_objects.add(obj)
try:
cls = obj.__class__
except Exception:
# see issue #16
cls = type(obj)
if not isinstance(cls, type):
cls = type(obj)
return consts.LABEL_REMOTE_REF, (id(obj), cls.__name__, cls.__module__)
def _netref_factory(self, oid, clsname, modname):
typeinfo = (clsname, modname)
if typeinfo in self._netref_classes_cache:
cls = self._netref_classes_cache[typeinfo]
elif typeinfo in netref.builtin_classes_cache:
cls = netref.builtin_classes_cache[typeinfo]
else:
info = self.sync_request(consts.HANDLE_INSPECT, oid)
cls = netref.class_factory(clsname, modname, info)
self._netref_classes_cache[typeinfo] = cls
return cls(weakref.ref(self), oid)
def _handle_instancecheck(self, obj, other_id_pack):
# TODOs:
# + refactor cache instancecheck/inspect/class_factory
# + improve cache docs
if hasattr(obj, '____conn__'): # keep unwrapping!
# When RPyC is chained (RPyC over RPyC), id_pack is cached in local objects as a netref
# since __mro__ is not a safe attribute the request is forwarded using the proxy connection
# relates to issue #346 or tests.test_netref_hierachy.Test_Netref_Hierarchy.test_StandardError
conn = obj.____conn__
return conn.sync_request(consts.HANDLE_INSPECT, id_pack)
# Create a name pack which would be familiar here and see if there is a hit
other_id_pack2 = (other_id_pack[0], other_id_pack[1], 0)
if other_id_pack[0] in netref.builtin_classes_cache:
cls = netref.builtin_classes_cache[other_id_pack[0]]
other = cls(self, other_id_pack)
elif other_id_pack2 in self._netref_classes_cache:
cls = self._netref_classes_cache[other_id_pack2]
other = cls(self, other_id_pack2)
else: # might just have missed cache, FIX ME
return False
return isinstance(other, obj)
def _box(self, obj):
"""store a local object in such a way that it could be recreated on
the remote party either by-value or by-reference"""
if brine.dumpable(obj):
return consts.LABEL_VALUE, obj
if type(obj) is tuple:
return consts.LABEL_TUPLE, tuple(self._box(item) for item in obj)
elif isinstance(obj, netref.BaseNetref) and obj.____conn__() is self:
return consts.LABEL_LOCAL_REF, obj.____oid__
else:
self._local_objects.add(obj)
try:
cls = obj.__class__
except Exception:
# see issue #16
cls = type(obj)
if not isinstance(cls, type):
cls = type(obj)
return consts.LABEL_REMOTE_REF, (id(obj), cls.__name__, cls.__module__)
def _handle_inspect(self, id_pack): # request handler
obj = self._local_objects[id_pack]
if hasattr(obj, "____conn__") and not isinstance(
obj, ghpythonlib.components.__namedtuple
):
# When RPyC is chained (RPyC over RPyC), id_pack is cached in local objects as a netref
# since __mro__ is not a safe attribute the request is forwarded using the proxy connection
# see issue #346 or tests.test_rpyc_over_rpyc.Test_rpyc_over_rpyc
conn = self._local_objects[id_pack].____conn__
return conn.sync_request(rpyc.core.consts.HANDLE_INSPECT, id_pack)
else:
return tuple(
rpyc.lib.get_methods(
rpyc.core.netref.LOCAL_ATTRS, self._local_objects[id_pack]
)
def _handle_inspect(self, oid):
return tuple(netref.inspect_methods(self._local_objects[oid]))
def _handle_getattr(self, oid, name):