pluto/handler/app.py

#!/usr/bin/env python3
# Twisted/Klein imports
from twisted.internet import reactor
from klein import Klein

# Other library imports
from json import dumps, loads
from json.decoder import JSONDecodeError
from signal import signal, SIGINT

# Project imports
from settings import settings
import util

# Old style classes
from agora import Agora
from transactions import Transactions
from markets import Markets
from money import Money

# New style classes
import sinks
import ux

init_map = None


# TODO: extend this with more
def cleanup(sig, frame):
    if init_map:
        try:
            init_map["tx"].lc_es_checks.stop()
            init_map["agora"].lc_dash.stop()
            init_map["agora"].lc_cheat.stop()
        except:  # noqa
            pass  # noqa
    reactor.stop()


signal(SIGINT, cleanup)  # Handle Ctrl-C and run the cleanup routine


class WebApp(util.Base):
    """
    Our Klein webapp.
    """

    app = Klein()

    @app.route("/callback", methods=["POST"])
    def callback(self, request):
        content = request.content.read()
        try:
            parsed = loads(content)
        except JSONDecodeError:
            self.log.error(f"Failed to parse JSON callback: {content}")
            return dumps(False)
        self.log.info("Callback received: {parsed}", parsed=parsed["data"]["id"])
        # self.tx.transaction(parsed)
        return dumps(True)

    # set up another connection to a bank
    @app.route("/signin", methods=["GET"])
    def signin(self, request):
        auth_url = self.sinks.truelayer.create_auth_url()
        return f'Please sign in <a href="{auth_url}" target="_blank">here.</a>'

    # endpoint called after we finish setting up a connection above
    @app.route("/callback-truelayer", methods=["POST"])
    def signin_callback(self, request):
        code = request.args[b"code"]
        self.sinks.truelayer.handle_authcode_received(code)
        return dumps(True)

    @app.route("/accounts", methods=["GET"])
    def balance(self, request):
        accounts = self.sinks.truelayer.get_accounts()
        return dumps(accounts, indent=2)


if __name__ == "__main__":
    init_map = {
        "ux": ux.UX(),
        "agora": Agora(),
        "markets": Markets(),
        "sinks": sinks.Sinks(),
        "tx": Transactions(),
        "webapp": WebApp(),
        "money": Money(),
    }
    # Merge all classes into each other
    util.xmerge_attrs(init_map)

    # Let the classes know they have been merged
    for class_name, class_instance in init_map.items():
        if hasattr(class_instance, "__xmerged__"):
            class_instance.__xmerged__()

    # Set up the loops to put data in ES
    init_map["tx"].setup_loops()

    # Run the WebApp
    init_map["webapp"].app.run(settings.App.BindHost, 8080)