more work on akahuclient, hard coding in account names I care about

This commit is contained in:
jethro 2026-05-13 18:45:12 +12:00
parent 55ee7e61b5
commit 63588d35f9
2 changed files with 58 additions and 11 deletions

View file

@ -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)
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()

16
AkahuClient/main.py Normal file
View file

@ -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)