Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@fbase.interface_attrs(
transform_args=fbase.transform_kwargs_only, return_annotation=bool)
def has(self, wire, **kwargs):
key = self.key(wire, **kwargs)
return wire.storage.has(key)
@fbase.interface_attrs(
transform_args=transform_cache_page_args)
def execute(self, wire, request, *args, **kwargs):
middleware = self.middleware
view_func = wire.__func__
try:
response = view_func(request, *args, **kwargs)
except Exception as e:
if hasattr(middleware, 'process_exception'):
result = middleware.process_exception(request, e)
if result is not None:
return result
raise
if hasattr(response, 'render') and callable(response.render):
if hasattr(middleware, 'process_template_response'):
response = middleware.process_template_response(
request, response)
@fbase.interface_attrs(
transform_args=transform_cache_page_args,
return_annotation=Tuple[str, str])
def key(self, wire, request, *args, **kwargs):
middleware = self.middleware
key_get = get_cache_key(
request, middleware.key_prefix, 'GET', cache=middleware.cache)
key_head = get_cache_key(
request, middleware.key_prefix, 'HEAD', cache=middleware.cache)
return key_get, key_head
@fbase.interface_attrs(
transform_args=(fbase.transform_kwargs_only, {'prefix_count': 1}),
return_annotation=None)
def set(self, wire, _value, **kwargs):
key = self.key(wire, **kwargs)
wire.storage.set(key, _value)
@fbase.interface_attrs(
transform_args=fbase.transform_kwargs_only,
return_annotation=lambda a: Optional[a.get('return', Any)])
@asyncio.coroutine
def get(self, wire, **kwargs):
key = self.key(wire, **kwargs)
try:
result = yield from wire.storage.get(key)
except fbase.NotFound:
result = wire._rope.config.miss_value
return result
@fbase.interface_attrs(
return_annotation=lambda a: List[a.get('return', Any)])
def get_or_update_many(self, wire, *args_list):
keys = self.key_many(wire, *args_list)
miss_value = object()
results = wire.storage.get_many(keys, miss_value=miss_value)
miss_indices = []
for i, akr in enumerate(zip(args_list, keys, results)):
args, key, result = akr
if result is not miss_value:
continue
miss_indices.append(i)
new_results = [
fbase.execute_bulk_item(wire, args_list[i]) for i in miss_indices]
new_keys = [keys[i] for i in miss_indices]
@fbase.interface_attrs(
return_annotation=lambda a: List[a.get('return', Any)])
def execute_many(self, wire, *args_list):
values = [fbase.execute_bulk_item(wire, args) for args in args_list]
return values
@fbase.interface_attrs(transform_args=fbase.transform_kwargs_only)
@asyncio.coroutine
def get_or_update(self, wire, **kwargs):
key = self.key(wire, **kwargs)
try:
result = yield from wire.storage.get(key)
except fbase.NotFound:
result = yield from self.execute(wire, **kwargs)
yield from wire.storage.set(key, result)
return result
@fbase.interface_attrs(
transform_args=fbase.transform_kwargs_only, return_annotation=bool)
def has(self, wire, **kwargs):
key = self.key(wire, **kwargs)
return wire.storage.has(key)
@fbase.interface_attrs(
transform_args=fbase.transform_kwargs_only,
return_annotation=lambda a: Optional[a.get('return', Any)])
def get(self, wire, **kwargs):
key = self.key(wire, **kwargs)
try:
result = wire.storage.get(key)
except fbase.NotFound:
result = wire._rope.config.miss_value
return result