pluto/handler/transactions.py

# Twisted/Klein imports
from twisted.logger import Logger

# Other library imports
from json import dumps

# Project imports
from db import r


class Transactions(object):
    """
    Handler class for incoming Revolut transactions.
    """

    def __init__(self):
        self.log = Logger("transactions")

    def transaction(self, data):
        """
        Store details of transaction.
        """
        event = data["event"]
        ts = data["timestamp"]

        inside = data["data"]
        txid = inside["id"]
        txtype = inside["type"]
        state = inside["state"]
        reference = inside["reference"]

        leg = inside["legs"][0]

        account_type = leg["counterparty"]["account_type"]
        account_id = leg["counterparty"]["account_id"]

        amount = leg["amount"]
        currency = leg["currency"]
        description = leg["description"]

        to_store = {
            "event": event,
            "ts": ts,
            "txid": txid,
            "txtype": txtype,
            "state": state,
            "reference": reference,
            "account_type": account_type,
            "account_id": account_id,
            "amount": amount,
            "currency": currency,
            "description": description,
        }

        self.log.info("Transaction processed: {formatted}", formatted=dumps(to_store, indent=2))
        r.hmset(f"tx.{txid}", to_store)

    def find_tx(self, reference, amount):
        all_transactions = r.scan(0, match="tx.*")
        print(f"ALL {all_transactions}")
        for tx_iter in all_transactions[1]:
            tx_obj = r.hgetall(tx_iter)
            print(f"ITER {tx_iter}")
            print(f"OBJ {tx_obj}")
            if tx_obj[b"reference"] == str.encode(reference):
                if tx_obj[b"amount"] == str.encode(amount):
                    return tx_obj
                else:
                    return "AMOUNT_INVALID"
        return False