Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class Config(object):
coder = attr.ib()
user_interface = attr.ib()
storage_backend = attr.ib()
storage_class = attr.ib()
miss_value = attr.ib()
default_action = attr.ib()
key_encoding = attr.ib()
expire_default = attr.ib()
key_refactor = attr.ib()
key_prefix = attr.ib()
ignorable_keys = attr.ib()
# wire_class = attr.ib()
class RingWire(Wire):
__slots__ = ()
def __init__(self, rope, *args, **kwargs):
super(RingWire, self).__init__(rope, *args, **kwargs)
@property
def storage(self):
return self._rope.storage
def encode(self, v):
return self._rope.encode(v)
def decode(self, v):
return self._rope.decode(v)
__all__ = 'lru_cache',
if hasattr(functools, 'lru_cache'):
_functools_lru_cache = functools.lru_cache
else:
try:
import functools32
except ImportError:
# raise AttributeError about fallback failure
functools.lru_cache # install `functools32` to run on py2
else:
_functools_lru_cache = functools32.lru_cache
class _LruCacheWire(Wire):
def __init__(self, rope, *args, **kwargs):
super(_LruCacheWire, self).__init__(rope, *args, **kwargs)
lru_args, lru_kwargs = rope._args
wrapper = _functools_lru_cache(
*lru_args, **lru_kwargs)(self.__func__)
self.__call__ = wrapper
self.cache_clear = wrapper.cache_clear
self.cache_info = wrapper.cache_info
def __call__(self, *args, **kwargs):
# descriptor detection support - never called
return self.__call__(*args, **kwargs)
def _on_property(self):
return self.__call__()
# wire_slots = ()
# func = self.__func__ if type(self.__func__) is types.FunctionType else Callable(self.__func__).wrapped_callable # noqa
# interface_keys = tuple(k for k in dir(user_interface) if k[0] != '_')
class _RingWire(RingWire):
# if wire_slots is not False:
# assert isinstance(wire_slots, tuple)
# __slots__ = interface_keys + wire_slots
if allows_default_action:
# @functools.wraps(func)
def __call__(self, *args, **kwargs):
return self.run(self._rope.config.default_action, *args, **kwargs)
self.wire_rope = WireRope(_RingWire, RingRope)
self.wire_rope._ring_object = self
def lru_cache(*args, **kwargs):
return WireRope(_LruCacheWire, wraps=True, rope_args=(args, kwargs))
merged[p.name] = var_kws
elif var_kws:
raise TypeError(
"{}() got multiple values for arguments '{}'".format(
self.code.co_name, list(var_kws.keys())))
return merged
def _identifier(self):
return '.'.join((
self.wrapped_callable.__module__, qualname(self.wrapped_callable)))
Callable.code = cached_property(_code)
Callable.annotations = property(_annotations)
Callable.kwargify = _kwargify
Callable.identifier = cached_property(_identifier)
elif var_kws:
raise TypeError(
"{}() got multiple values for arguments '{}'".format(
self.code.co_name, list(var_kws.keys())))
return merged
def _identifier(self):
return '.'.join((
self.wrapped_callable.__module__, qualname(self.wrapped_callable)))
Callable.code = cached_property(_code)
Callable.annotations = property(_annotations)
Callable.kwargify = _kwargify
Callable.identifier = cached_property(_identifier)
class PublicRing(object):
def __init__(self, rope):
self._rope = rope
def key(self, func):
self._rope.compose_key = func
def encode(self, func):
self._rope._encode = func
def decode(self, func):
self._rope._decode = func
class RingRope(RopeCore):
def __init__(self, *args, **kwargs):
super(RingRope, self).__init__(*args, **kwargs)
self._encode = None
self._decode = None
self.ring = PublicRing(self)
def compose_key(self, *bound_args, **kwargs):
config = self.config
_ignorable_keys = suggest_ignorable_keys(
self.callable, self.config.ignorable_keys)
_key_prefix = suggest_key_prefix(self.callable, config.key_prefix)
c = self.callable
merged[p.name] = var_kws
elif var_kws:
raise TypeError(
"{}() got multiple values for arguments '{}'".format(
self.code.co_name, list(var_kws.keys())))
return merged
def _identifier(self):
return '.'.join((
self.wrapped_callable.__module__, qualname(self.wrapped_callable)))
Callable.code = cached_property(_code)
Callable.annotations = property(_annotations)
Callable.kwargify = _kwargify
Callable.identifier = cached_property(_identifier)
raise TypeError(
"{}() got multiple values for arguments '{}'".format(
self.code.co_name, list(var_kws.keys())))
return merged
def _identifier(self):
return '.'.join((
self.wrapped_callable.__module__, qualname(self.wrapped_callable)))
Callable.code = cached_property(_code)
Callable.annotations = property(_annotations)
Callable.kwargify = _kwargify
Callable.identifier = cached_property(_identifier)