Personal finance aggregator: The first prototype: a CLI script that links a bank - Part 3

~10 min read

By this point I had sandbox credentials, a registered redirect URI, and a scope string I'd thought through (see the previous post). What I didn't have was any actual code, and I've learned not to trust an integration design until I've built the ugliest possible version of it and watched it work. So the first real artifact in this project wasn't an API, or a database, or a frontend - it was a single ~180-line script, truelayer_sync.py, that did exactly two things: link <bank> and fetch. No web server, no persistence layer beyond a JSON file, no error handling beyond what sys.exit gives you for free. The point wasn't elegance, it was proving the OAuth mechanics actually worked the way the docs said they would.

A note on names, for anyone following along in the git history: the script's own docstring calls it truelayer_sync.py, but the first commit that contains it filed it as true_layer/check_banks_truelayer.py - it started life as the TrueLayer half of the provider comparison in the previous posts, and grew link and fetch commands from there. It was renamed into the package as src/finance_system/truelayer.py in the initial backend commit, which is where the backend skeleton first appears.

The redirect problem, solved the cheap way

Authorization code flow needs a redirect URI that can receive the authorization code after the user approves access at their bank. In a real web app that's a route on your server. In a CLI script running on my laptop, there is no server - so I made one, just for the duration of the flow:

def wait_for_code():
    """Run a tiny local server that catches the redirect and returns the auth code."""
    result = {}

    class Handler(BaseHTTPRequestHandler):
        def do_GET(self):
            query = parse_qs(urlparse(self.path).query)
            result["code"] = query.get("code", [None])[0]
            result["error"] = query.get("error", [None])[0]
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b"Done - you can close this tab.")

        def log_message(self, *args):
            pass

    HTTPServer(("localhost", 3000), Handler).handle_request()
    if result.get("error"):
        sys.exit(f"Bank linking failed: {result['error']}")
    return result["code"]

HTTPServer(...).handle_request() - not serve_forever() - is the detail that makes this work as a script rather than a service: it starts a server, blocks until exactly one request comes in, handles it, and returns. Combined with the redirect URI I'd registered (http://localhost:3000/callback), the flow became: open a browser tab to TrueLayer's hosted consent page, block on wait_for_code(), and the moment the browser gets redirected back with a ?code=... query param, the script wakes up with it in hand. No polling, no webhook, no server that needs to keep running after the script exits.

def link(label):
    auth_url = f"{AUTH_HOST}/?" + urlencode({
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": SCOPES,
        "providers": PROVIDERS,
    })
    print("Opening browser to link your bank...")
    webbrowser.open(auth_url)

    code = wait_for_code()
    tokens = token_request({
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": REDIRECT_URI,
    })

    all_tokens = load_tokens()
    all_tokens[label] = {"refresh_token": tokens["refresh_token"]}
    save_tokens(all_tokens)
    print(f"Linked '{label}'.")

label here is just whatever name I pass on the command line - monzo, nationwide-cc, whatever I want to call it locally. TrueLayer doesn't know or care what I call it; it's purely a key in my own tokens.json so I can tell my accounts apart later. Note what gets stored: not the access token (useless after about an hour) but the refresh token, which is what offline_access scope buys you, and what lets fetch run unattended without a browser in sight.

Storage that's really just a dict and a JSON file

load_tokens and save_tokens are about as unglamorous as code gets, and that was entirely the point at this stage:

TOKENS_FILE = Path("tokens.json")  # sensitive - never commit this file

def load_tokens():
    return json.loads(TOKENS_FILE.read_text()) if TOKENS_FILE.exists() else {}

def save_tokens(tokens):
    TOKENS_FILE.write_text(json.dumps(tokens, indent=2))

No database, no schema, no migrations - a dict keyed by label, dumped as JSON, read back in on the next run. I could open tokens.json in a text editor at any point and see exactly what state the script thought it was in, which turned out to be a genuinely useful debugging property in its own right: when the second run's refresh failed (more on that below), the first thing I did was just look at the file, and the stale refresh token was immediately obvious sitting there in plain text. A "real" database at this stage would have added ceremony without adding anything I actually needed yet - table definitions, a connection to manage, a query to write just to answer "what got saved." That trade only flips once you need things a flat file genuinely can't do well, like concurrent access or querying by a field other than the key - which is exactly what justified SQLite a few commits later, not before.

Fetching, the boring but important part

fetch() is where the actual value lives, and it's almost entirely plumbing: for every bank I'd linked, swap the stored refresh token for a fresh access token, list its accounts and cards, then pull transactions for each one.

def fetch(days=90):
    all_tokens = load_tokens()
    if not all_tokens:
        sys.exit("No banks linked yet - run: python truelayer_sync.py link <name>")

    date_range = {
        "from": (date.today() - timedelta(days=days)).isoformat(),
        "to": date.today().isoformat(),
    }
    output = []

    for label, stored in all_tokens.items():
        # Access tokens last about an hour, so swap the refresh token for a fresh one.
        tokens = token_request({
            "grant_type": "refresh_token",
            "refresh_token": stored["refresh_token"],
        })
        all_tokens[label]["refresh_token"] = tokens["refresh_token"]  # it rotates
        access = tokens["access_token"]

        sources = [("accounts", a["account_id"], a["display_name"]) for a in get(access, "/accounts")]
        sources += [("cards", c["account_id"], c["display_name"]) for c in get(access, "/cards")]

        for kind, account_id, name in sources:
            txns = get(access, f"/{kind}/{account_id}/transactions", date_range)
            ...

A few things in here turned out to matter a lot more later than they looked like they would at the time:

Refresh tokens rotate. all_tokens[label]["refresh_token"] = tokens["refresh_token"] isn't defensive copying, it's load-bearing. TrueLayer issues a new refresh token every time you use one, and if you don't save the new one, the next refresh fails with the old one already invalidated. I found this the hard way on the second run of the script, when fetch worked once and then errored on every subsequent call until I noticed the token file wasn't being rewritten after a successful refresh.

One get() helper handles a provider-specific quirk. Not every bank supports every endpoint - my Monzo connection has no /cards, so TrueLayer returns a 501 Not Implemented for that call rather than an empty list. Treating that as "no cards for this bank" instead of an error, right in the shared helper, meant every call site downstream didn't need to know or care which banks have cards:

def get(access_token, path, params=None):
    resp = requests.get(
        f"{API_HOST}/data/v1{path}",
        headers={"Authorization": f"Bearer {access_token}"},
        params=params,
    )
    if resp.status_code == 501:  # endpoint not supported by this bank (e.g. no cards)
        return []
    resp.raise_for_status()
    return resp.json()["results"]

kind ("accounts" or "cards") travels with each source all the way through the loop, because the transactions endpoint path differs - /accounts/{id}/transactions vs /cards/{id}/transactions. It's a small thing, but structuring sources as a flat list of (kind, id, name) tuples up front, rather than handling accounts and cards as two separate code paths, is a pattern that survived essentially unchanged all the way to the current codebase.

The rest is shaping TrueLayer's transaction object into something flatter and easier to work with downstream - bank, account, id, date, description, merchant, amount, currency, bank_category - and dumping the whole list to transactions.json at the end.

What this version got right

For a script I expected to throw away, a surprising amount of it didn't get thrown away. The link / fetch split - one interactive step that needs a browser, one unattended step that doesn't - is still exactly the shape of the real system today, just with link's browser-opening replaced by a proper web redirect and fetch running on a nightly timer instead of by hand. The refresh-token rotation handling, the 501-as-empty-list treatment, the (kind, account_id, name) tuple shape: all still there, close to verbatim, several rewrites later.

Why a CLI first, and not a minimal web app

It would have been entirely possible to skip straight to a FastAPI app with a /link and /callback route instead of a local HTTP server that lives for one request. I deliberately didn't, and it's worth explaining why rather than treating it as an obvious choice, because it wasn't obvious to me until partway through.

A web app, even a minimal one, means a process that stays running, a port that has to be free, and - if I'm testing the deployed shape of things - a place to actually put it. None of that helps answer the one question this stage of the project needed answered: does the OAuth exchange work the way the docs describe, against a real (or sandbox) TrueLayer environment, with my actual credentials? A CLI script that runs, does one thing, and exits is the fastest possible loop for answering that - I could change one line, re-run python truelayer_sync.py link monzo, and know within seconds whether that change helped, with no server to restart, no port conflicts from a previous run I forgot to kill, nothing else in the way. Collapsing "validate the integration" and "build the real service" into one step would have meant every OAuth debugging cycle also paying the cost of a web server's startup and routing, for no benefit at this stage. Keeping them separate meant the slow, careful part (getting a web service's shape right) only had to happen once, after the fast, disposable part (does this API actually work the way I think) had already answered the questions that mattered.

What it obviously couldn't do

It also made very clear, very quickly, what it couldn't do, which is really the point of building the throwaway version first:

It only worked on my laptop. webbrowser.open() and a localhost:3000 callback server assume there's a browser and a human sitting right there. No deploying this anywhere.
No de-duplication. Every fetch overwrote transactions.json wholesale. Fine for a script I ran by hand a few times; not fine for something meant to run nightly and accumulate history.
tokens.json sitting on disk in plaintext was acceptable for an evening's experiment and not acceptable for anything I'd trust with real account access longer-term.
No real error handling. A sys.exit() on missing tokens is fine for a CLI; it's not something you can build a service around.

None of that mattered yet. The only question this version of the code needed to answer was "does the OAuth flow work, end to end, against my real banks" - and by the time I had a transactions.json file sitting on disk with real Monzo transactions in it, formatted the way I wanted, the answer was yes. Everything after this post is about turning that yes into something that could run unattended, store data properly, and be looked at from more than a terminal.

Previous: ← Developer access and app setup · Next: From script to service →