XF
/
pluto
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Implement serde for JSON deserialisation

Browse Source
master
Mark Veidemanis 3 years ago
parent a69a585c71
commit b3a39401b5
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
2 changed files with 73 additions and 46 deletions
Show Stats Download Patch File Download Diff File

92
handler/sinks/nordigen.py
Unescape Escape View File

@ -5,6 +5,8 @@ from twisted.internet.task import LoopingCall
import requests
from simplejson.errors import JSONDecodeError
from json import dumps, loads
from lib.serde.nordigen import TXRoot, AccessToken, Institutions, Agreement, Requisitions, AccountDetails
from serde import ValidationError
# Project imports
from settings import settings
@ -57,15 +59,13 @@ class Nordigen(util.Base):
path = f"{settings.Nordigen.Base}/token/new/"
r = requests.post(path, headers=headers, data=dumps(data))
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing access token response: {r.content}")
return False
if "access" in parsed:
obj = AccessToken.from_json(r.content)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()
self.token = parsed["access"]
self.log.info("Refreshed access token")
else:
self.log.error(f"Access token not in response: {parsed}")
if not self.authed:
self.__authed__()
self.authed = True
@ -84,10 +84,17 @@ class Nordigen(util.Base):
path = f"{settings.Nordigen.Base}/institutions/?country={country}"
r = requests.get(path, headers=headers)
try:
parsed = r.json()
parsed_pre = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing institutions response: {r.content}")
return False
parsed = {"institutions": parsed_pre}
try:
obj = Institutions.from_dict(parsed)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()["institutions"]
new_list = []
if filter_name:
for i in parsed:
@ -96,21 +103,6 @@ class Nordigen(util.Base):
return new_list
return parsed
def create_agreement(self, institution_id):
"""Create an agreement to access an institution.
:param institution_id: ID of the institution
"""
headers = {"accept": "application/json", "Authorization": f"Bearer {self.token}"}
path = f"{settings.Nordigen.Base}/agreements/enduser"
data = {"institution_id": institution_id}
r = requests.post(path, headers=headers, data=dumps(data))
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing agreement response: {r.content}")
return False
return parsed
def build_link(self, institution_id):
"""Create a link to access an institution.
:param institution_id: ID of the institution
@ -120,10 +112,11 @@ class Nordigen(util.Base):
data = {"institution_id": institution_id, "redirect": settings.Nordigen.CallbackURL}
r = requests.post(path, headers=headers, data=data)
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing link response: {r.content}")
return False
obj = Agreement.from_json(r.content)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()
if "link" in parsed:
return parsed["link"]
return False
@ -150,14 +143,16 @@ class Nordigen(util.Base):
path = f"{settings.Nordigen.Base}/requisitions"
r = requests.get(path, headers=headers)
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing requisitions response: {r.content}")
return False
obj = Requisitions.from_json(r.content)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()
if "results" in parsed:
return parsed["results"]
else:
self.log.error(f"Results not in requisitions response: {parsed}")
return False
def get_accounts(self, requisition):
@ -168,10 +163,11 @@ class Nordigen(util.Base):
path = f"{settings.Nordigen.Base}/requisitions/{requisition}/"
r = requests.get(path, headers=headers)
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing accounts response: {r.content}")
return False
obj = Agreement.from_json(r.content)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()
if "accounts" in parsed:
return parsed["accounts"]
return False
@ -184,14 +180,11 @@ class Nordigen(util.Base):
path = f"{settings.Nordigen.Base}/accounts/{account_id}/details/"
r = requests.get(path, headers=headers)
try:
parsed = r.json()
except JSONDecodeError:
self.log.error(f"Error parsing account response: {r.content}")
return False
# if "accounts" in parsed:
# return parsed["accounts"]
# return False
parsed = parsed["account"]
obj = AccountDetails.from_json(r.content)
except ValidationError as err:
self.log.error(f"Validation error: {err}")
return
parsed = obj.to_dict()["account"]
if "bban" in parsed and parsed["currency"] == "GBP":
sort_code = parsed["bban"][0:6]
account_number = parsed["bban"][6:]
@ -252,3 +245,16 @@ class Nordigen(util.Base):
else:
to_return[req["institution_id"]] = [account_info]
return to_return
def get_transactions(self, account_id):
"""
Get all transactions for an account.
:param account_id: account to fetch transactions for
:return: list of transactions
:rtype: dict
"""
headers = {"accept": "application/json", "Authorization": f"Bearer {self.token}"}
path = f"{settings.Nordigen.Base}/accounts/{account_id}/transactions/"
r = requests.get(path, headers=headers)
obj = TXRoot.from_json(r.content)
return obj.to_dict()["transactions"]["booked"]

23
handler/ux/commands.py
Unescape Escape View File

@ -541,7 +541,7 @@ class IRCCommands(object):
class transactions(object):
name = "transactions"
authed = True
helptext = "Get a list of transactions. Usage: transactions <account> <account_id>"
helptext = "Get a list of transactions from TrueLayer. Usage: transactions <account> <account_id>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
@ -559,6 +559,27 @@ class IRCCommands(object):
description = transaction["description"]
msg(f"{timestamp} {txid} {ptxid} {txtype} {amount}{currency} {description}")
class ntransactions(object):
name = "ntransactions"
authed = True
helptext = "Get a list of transactions from Nordigen. Usage: ntransactions <account_id>"
@staticmethod
def run(cmd, spl, length, authed, msg, agora, tx, ux):
if length == 2:
account_id = spl[1]
transactions = tx.sinks.nordigen.get_transactions(account_id)
for transaction in transactions:
msg(dumps(transaction))
# txid = transaction["transaction_id"]
# ptxid = transaction["meta"]["provider_transaction_id"]
# txtype = transaction["transaction_type"]
# timestamp = transaction["timestamp"]
# amount = transaction["amount"]
# currency = transaction["currency"]
# description = transaction["description"]
# msg(f"{timestamp} {txid} {ptxid} {txtype} {amount}{currency} {description}")
class mapaccount(object):
name = "mapaccount"
authed = True

Write Preview
Loading…
Cancel
Save