mirror of
https://github.com/GSA/notifications-admin.git
synced 2026-02-06 03:13:42 -05:00
f6367f2278
when clients are defined in app/__init__.py, it increases the chance of cyclical imports. By moving module level client singletons out to a separate extensions file, we stop cyclical imports, but keep the same code flow - the clients are still initialised in `create_app` in `__init__.py`. The redis client in particular is no longer separate - previously redis was set up on the `NotifyAdminAPIClient` base class, but now there's one singleton in `app.extensions`. This was done so that we can access redis from outside of the existing clients.
71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
import json
|
||
from contextlib import suppress
|
||
from datetime import timedelta
|
||
from functools import wraps
|
||
from inspect import signature
|
||
|
||
from app.extensions import redis_client
|
||
|
||
TTL = int(timedelta(days=7).total_seconds())
|
||
|
||
|
||
def _get_argument(argument_name, client_method, args, kwargs):
|
||
|
||
with suppress(KeyError):
|
||
return kwargs[argument_name]
|
||
|
||
with suppress(ValueError, IndexError):
|
||
argument_index = list(signature(client_method).parameters).index(argument_name)
|
||
return args[argument_index - 1] # -1 because `args` doesn’t include `self`
|
||
|
||
with suppress(KeyError):
|
||
return signature(client_method).parameters[argument_name].default
|
||
|
||
raise TypeError("{}() takes no argument called '{}'".format(
|
||
client_method.__name__, argument_name
|
||
))
|
||
|
||
|
||
def _make_key(key_format, client_method, args, kwargs):
|
||
return key_format.format(**{
|
||
argument_name: _get_argument(argument_name, client_method, args, kwargs)
|
||
for argument_name in list(signature(client_method).parameters)
|
||
})
|
||
|
||
|
||
def set(key_format):
|
||
|
||
def _set(client_method):
|
||
|
||
@wraps(client_method)
|
||
def new_client_method(client_instance, *args, **kwargs):
|
||
redis_key = _make_key(key_format, client_method, args, kwargs)
|
||
cached = redis_client.get(redis_key)
|
||
if cached:
|
||
return json.loads(cached.decode('utf-8'))
|
||
api_response = client_method(client_instance, *args, **kwargs)
|
||
redis_client.set(
|
||
redis_key,
|
||
json.dumps(api_response),
|
||
ex=TTL,
|
||
)
|
||
return api_response
|
||
|
||
return new_client_method
|
||
return _set
|
||
|
||
|
||
def delete(key_format):
|
||
|
||
def _delete(client_method):
|
||
|
||
@wraps(client_method)
|
||
def new_client_method(client_instance, *args, **kwargs):
|
||
redis_key = _make_key(key_format, client_method, args, kwargs)
|
||
redis_client.delete(redis_key)
|
||
return client_method(client_instance, *args, **kwargs)
|
||
|
||
return new_client_method
|
||
|
||
return _delete
|