Add docstrings in Revolut

This commit is contained in:
Mark Veidemanis 2021-12-27 21:41:35 +00:00
parent 987e8545f5
commit 2cdc43e1cd
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
1 changed files with 40 additions and 6 deletions

View File

@ -19,6 +19,10 @@ class Revolut(object):
""" """
def __init__(self): def __init__(self):
"""
Initialise the Revolut object.
Set the logger and token.
"""
self.log = Logger("revolut") self.log = Logger("revolut")
self.token = None self.token = None
@ -26,10 +30,16 @@ class Revolut(object):
self.irc = irc self.irc = irc
def setup_auth(self): def setup_auth(self):
"""
Function to create a new Java Web Token and use it to get a refresh/access token.
"""
self.create_new_jwt() self.create_new_jwt()
self.get_access_token() self.get_access_token()
def create_new_jwt(self): def create_new_jwt(self):
"""
Create a new Java Web Token.
"""
payload = { payload = {
"iss": settings.Revolut.Domain, "iss": settings.Revolut.Domain,
"sub": settings.Revolut.ClientID, "sub": settings.Revolut.ClientID,
@ -46,6 +56,11 @@ class Revolut(object):
settings.write() settings.write()
def get_access_token(self): def get_access_token(self):
"""
Get an access token with our JWT.
:return: True or False
:rtype: bool
"""
headers = {"Content-Type": "application/x-www-form-urlencoded"} headers = {"Content-Type": "application/x-www-form-urlencoded"}
data = { data = {
"grant_type": "authorization_code", "grant_type": "authorization_code",
@ -76,6 +91,13 @@ class Revolut(object):
return False return False
def get_new_token(self, fail=False): def get_new_token(self, fail=False):
"""
Get a new access token with the refresh token.
:param fail: whether to exit() if this fails
:type fail: bool
:return: True or False
:rtype: bool
"""
headers = {"Content-Type": "application/x-www-form-urlencoded"} headers = {"Content-Type": "application/x-www-form-urlencoded"}
data = { data = {
"grant_type": "refresh_token", "grant_type": "refresh_token",
@ -108,10 +130,16 @@ class Revolut(object):
return False return False
def setup_webhook(self): def setup_webhook(self):
webhooks = self.get_webhooks() """
if "url" in webhooks.keys(): Check the webhook we have set up in Revolut.
if webhooks["url"] == settings.Revolut.WebhookURL: Set up configured webhook if not already set up.
self.log.info("Webhook exists - skipping setup: {url}", url=webhooks["url"]) :return: True or False
:rtype: bool
"""
webhook = self.get_webhook()
if "url" in webhook.keys():
if webhook["url"] == settings.Revolut.WebhookURL:
self.log.info("Webhook exists - skipping setup: {url}", url=webhook["url"])
return True # Webhook already exists return True # Webhook already exists
self.log.info("Setting up webhook: {url}", url=settings.Revolut.WebhookURL) self.log.info("Setting up webhook: {url}", url=settings.Revolut.WebhookURL)
headers = {"Authorization": f"Bearer {self.token}"} headers = {"Authorization": f"Bearer {self.token}"}
@ -123,8 +151,14 @@ class Revolut(object):
else: else:
parsed = r.json() parsed = r.json()
self.log.info("Failed setting up webhook: {parsed}", parsed=parsed) self.log.info("Failed setting up webhook: {parsed}", parsed=parsed)
return False
def get_webhooks(self): def get_webhook(self):
"""
Get the webhook address active in Revolut.
:return: dict of hook with key url or False
:rtype: dict or bool
"""
headers = {"Authorization": f"Bearer {self.token}"} headers = {"Authorization": f"Bearer {self.token}"}
r = requests.get(f"{settings.Revolut.Base}/webhook", headers=headers) r = requests.get(f"{settings.Revolut.Base}/webhook", headers=headers)
if r.status_code == 200: if r.status_code == 200:
@ -133,5 +167,5 @@ class Revolut(object):
elif r.status_code == 404: elif r.status_code == 404:
return {} return {}
else: else:
self.log.error("Cannot get webhooks: {content}", r.content) self.log.error("Cannot get webhook: {content}", r.content)
return False return False