Refactor initialisation and add docstrings to util module

This commit is contained in:
Mark Veidemanis 2022-02-28 20:18:42 +00:00
parent e7283b304f
commit 90da53ce99
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
2 changed files with 44 additions and 15 deletions

View File

@ -2,7 +2,6 @@
# Twisted/Klein imports # Twisted/Klein imports
from twisted.logger import Logger from twisted.logger import Logger
from twisted.internet import reactor from twisted.internet import reactor
from twisted.internet.task import LoopingCall, deferLater
from klein import Klein from klein import Klein
# Other library imports # Other library imports
@ -76,18 +75,6 @@ class WebApp(object):
return dumps(True) return dumps(True)
def setup_call_loops(token_setting, function_init, function_continuous, delay, function_post_start=None):
if token_setting == "1":
deferLater(reactor, 1, function_init)
else:
deferLater(reactor, 1, function_continuous, True)
if function_post_start:
deferLater(reactor, 4, function_post_start)
lc = LoopingCall(function_continuous)
lc.start(delay)
if __name__ == "__main__": if __name__ == "__main__":
init_map = { init_map = {
"notify": Notify(), "notify": Notify(),
@ -103,16 +90,18 @@ if __name__ == "__main__":
"webapp": WebApp(), "webapp": WebApp(),
"money": Money(), "money": Money(),
} }
# Merge all classes into each other
util.xmerge_attrs(init_map) util.xmerge_attrs(init_map)
setup_call_loops( # Setup the authcode -> refresh token and refresh_token -> auth_token stuff
util.setup_call_loops(
token_setting=settings.Revolut.SetupToken, token_setting=settings.Revolut.SetupToken,
function_init=init_map["revolut"].setup_auth, function_init=init_map["revolut"].setup_auth,
function_continuous=init_map["revolut"].get_new_token, function_continuous=init_map["revolut"].get_new_token,
delay=int(settings.Revolut.RefreshSec), delay=int(settings.Revolut.RefreshSec),
function_post_start=init_map["revolut"].setup_webhook, function_post_start=init_map["revolut"].setup_webhook,
) )
setup_call_loops( util.setup_call_loops(
token_setting=settings.TrueLayer.SetupToken, token_setting=settings.TrueLayer.SetupToken,
function_init=init_map["truelayer"].setup_auth, function_init=init_map["truelayer"].setup_auth,
function_continuous=init_map["truelayer"].get_new_token, function_continuous=init_map["truelayer"].get_new_token,

View File

@ -1,5 +1,7 @@
# Twisted/Klein imports # Twisted/Klein imports
from twisted.logger import Logger from twisted.logger import Logger
from twisted.internet import reactor
from twisted.internet.task import LoopingCall, deferLater
# Other library imports # Other library imports
from httpx import ReadTimeout, ReadError, RemoteProtocolError from httpx import ReadTimeout, ReadError, RemoteProtocolError
@ -9,6 +11,12 @@ log = Logger("util.global")
def xmerge_attrs(init_map): def xmerge_attrs(init_map):
"""
Given a dictionary of strings and classes, set all corresponding class.<string> attributes
on each class, to every other class.
"a": A(), "b": B() -> A.b = B_instance, B.a = A_instance
:param init_map: dict of class names to classes
"""
for classname, object_instance in init_map.items(): for classname, object_instance in init_map.items():
# notify, Notify # notify, Notify
for classname_inside, object_instance_inside in init_map.items(): for classname_inside, object_instance_inside in init_map.items():
@ -17,6 +25,26 @@ def xmerge_attrs(init_map):
setattr(object_instance, classname_inside, object_instance_inside) setattr(object_instance, classname_inside, object_instance_inside)
def setup_call_loops(token_setting, function_init, function_continuous, delay, function_post_start=None):
"""
Setup the loops for dealing with access, refresh and auth tokens for various providers.
:param token_setting: the setting for whether to do the initial authentication
:param function_init: the initial authentication function
:param function_continuous: the ongoing authentication function (refresh_token -> access_token)
:param delay: time in seconds to wait between calls to function_continuous
:param function_post_start: an optional function to run after the access token is obtained
"""
if token_setting == "1":
deferLater(reactor, 1, function_init)
else:
deferLater(reactor, 1, function_continuous, True)
if function_post_start:
deferLater(reactor, 4, function_post_start)
lc = LoopingCall(function_continuous)
lc.start(delay)
def convert(data): def convert(data):
""" """
Recursively convert a dictionary. Recursively convert a dictionary.
@ -48,7 +76,19 @@ def last_online_recent(date):
def handle_exceptions(func): def handle_exceptions(func):
"""
Wrapper helper to handle Agora API errors.
:param func: function to wrap
:rtype: func
:return: the wrapped function
"""
def inner_function(*args, **kwargs): def inner_function(*args, **kwargs):
"""
Inner wrapper helper.
:rtype: any or bool
:return: False or the normal return
"""
try: try:
rtrn = func(*args, **kwargs) rtrn = func(*args, **kwargs)
except (ReadTimeout, ReadError, RemoteProtocolError): except (ReadTimeout, ReadError, RemoteProtocolError):