AkahuSync/AkahuClient/akahuclient.py

import requests
import os
from typing import Iterable, Optional, Any

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