From 63588d35f9538693e493119bf63868175d1f81dc Mon Sep 17 00:00:00 2001 From: jethro Date: Wed, 13 May 2026 18:45:12 +1200 Subject: [PATCH] more work on akahuclient, hard coding in account names I care about --- AkahuClient/akahuclient.py | 53 ++++++++++++++++++++++++++++++-------- AkahuClient/main.py | 16 ++++++++++++ 2 files changed, 58 insertions(+), 11 deletions(-) create mode 100644 AkahuClient/main.py diff --git a/AkahuClient/akahuclient.py b/AkahuClient/akahuclient.py index 5aad4eb..a6f88e6 100644 --- a/AkahuClient/akahuclient.py +++ b/AkahuClient/akahuclient.py @@ -1,11 +1,22 @@ import requests import os from dotenv import load_dotenv +from typing import Iterable, Optional, Any load_dotenv() class AkahuClient: + # Central place to configure which accounts you care about. + # You can override this by passing account_names=... to get_accounts(). + DEFAULT_ACCOUNT_NAMES = { + "Jethro's Investments", + "Jethro's KiwiSaver", + "Jethro's Rainy day fund", + "Income Account", + "Rent", + } + def __init__(self, token: str, app_id: str): self.token = token self.app_id = app_id @@ -13,19 +24,39 @@ class AkahuClient: self.headers = { "Authorization": f"Bearer {self.token}", "X-Akahu-Id": self.app_id, - "Accept": "application/json" + "Accept": "application/json", } - def get_accounts(self): - client = AkahuClient(self.token, self.app_id) - return client.get_accounts() + def get_accounts(self, account_names: Optional[Iterable[str]] = None) -> dict[str, Any]: + """ + Fetch accounts from Akahu and optionally filter by account name. - def get_transactions(self, account_id: str): - client = AkahuClient(self.token, self.app_id) - return client.get_transactions(account_id) + If account_names is None, DEFAULT_ACCOUNT_NAMES is used. + Returns the same shape as the API response, but with 'items' filtered. + """ + url = f"{self.api_url}/accounts" + resp = requests.get(url, headers=self.headers, timeout=30) + resp.raise_for_status() + data: dict[str, Any] = resp.json() -if __name__ == "__main__": - client = AkahuClient(TOKEN, APP_ID) - accounts = client.get_accounts() - print(accounts) \ No newline at end of file + items = data.get("items") or [] + if not isinstance(items, list): + return data + + names = set(account_names) if account_names is not None else set(self.DEFAULT_ACCOUNT_NAMES) + if not names: + return data + + data["items"] = [acct for acct in items if acct.get("name") in names] + return data + + def get_transactions(self, account_id: str, start_date: str, end_date: str) -> dict[str, Any]: + url = f"{self.api_url}/accounts/{account_id}/transactions" + params = { + "start_date": start_date, + "end_date": end_date + } + resp = requests.get(url, headers=self.headers, params=params, timeout=30) + resp.raise_for_status() + return resp.json() \ No newline at end of file diff --git a/AkahuClient/main.py b/AkahuClient/main.py new file mode 100644 index 0000000..dfcf466 --- /dev/null +++ b/AkahuClient/main.py @@ -0,0 +1,16 @@ +from akahuclient import AkahuClient +import os +from dotenv import load_dotenv + + +load_dotenv() + +TOKEN = os.getenv("AKAHU_API_TOKEN") +APP_ID = os.getenv("AKAHU_APP_ID") + +if not TOKEN or not APP_ID: + print("Please set AKAHU_API_TOKEN and AKAHU_APP_ID in your environment.") + exit(1) +client = AkahuClient(TOKEN, APP_ID) +accounts = client.get_accounts() +print(accounts) \ No newline at end of file