62 lines
2 KiB
Python
62 lines
2 KiB
Python
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
|
|
self.api_url = "https://api.akahu.io/v1"
|
|
self.headers = {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"X-Akahu-Id": self.app_id,
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
def get_accounts(self, account_names: Optional[Iterable[str]] = None) -> dict[str, Any]:
|
|
"""
|
|
Fetch accounts from Akahu and optionally filter by account name.
|
|
|
|
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()
|
|
|
|
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() |