import yaml from datetime import datetime, timedelta from pathlib import Path from typing import TypedDict import ynab class YNABTransaction(TypedDict): """Transaction data from YNAB.""" id: str date: str payee_name: str | None category_name: str | None amount_milliCurrency: int memo: str | None account_id: str account_name: str cleared: str import_id: str | None class YNABConfig(TypedDict): """Configuration loaded from ynab.yml.""" token: str plan_id: str accounts: list[dict[str, str]] class YNABClient: """Client for fetching transactions from YNAB.""" def __init__(self, config_path: str | Path = "ynab.yml") -> None: """Initialize YNAB client from config file.""" self.config = self._load_config(config_path) self.token = self.config["token"] self.plan_id = self.config["plan_id"] self.account_configs = self.config["accounts"] # Use SDK-native access_token auth for bearer token handling. config = ynab.Configuration(access_token=self.token) self.api_client = ynab.ApiClient(config) self.transactions_api = ynab.TransactionsApi(self.api_client) self.accounts_api = ynab.AccountsApi(self.api_client) self.categories_api = ynab.CategoriesApi(self.api_client) @staticmethod def _load_config(config_path: str | Path) -> YNABConfig: """Load YNAB configuration from YAML file.""" path = Path(config_path) with path.open("r", encoding="utf-8") as f: data = yaml.safe_load(f) ynab_config = data.get("ynab", [{}])[0] return { "token": ynab_config.get("token", ""), "plan_id": ynab_config.get("plan_id", ""), "accounts": ynab_config.get("accounts", []), } @staticmethod def _normalize_cleared_status(cleared: object) -> str: """Normalize YNAB cleared status values to lowercase strings.""" if hasattr(cleared, "value"): return str(getattr(cleared, "value")).strip().lower() status = str(cleared).strip() if "." in status: status = status.split(".")[-1] return status.lower() def get_transactions_for_days( self, days: int = 30, exclude_cleared_statuses: list[str] | None = None, ) -> list[YNABTransaction]: """ Fetch transactions from configured accounts for the past x days. Args: days: Number of days to look back (default: 30) exclude_cleared_statuses: Optional list of normalized statuses to skip, for example ["reconciled"] Returns: List of transactions with transformed data """ transactions: list[YNABTransaction] = [] cutoff_date = (datetime.now() - timedelta(days=days)).date() excluded_statuses = { status.strip().lower() for status in (exclude_cleared_statuses or []) } # Fetch all accounts to build name lookup accounts_response = self.accounts_api.get_accounts(self.plan_id) account_map = { str(acc.id): acc.name for acc in accounts_response.data.accounts } # Fetch categories for name lookup categories_response = self.categories_api.get_categories(self.plan_id) category_map = {} for category_group in categories_response.data.category_groups: for cat in category_group.categories: if cat.id: category_map[str(cat.id)] = cat.name # If specific accounts are configured, use only those; otherwise use all if self.account_configs: account_ids = [acc["id"] for acc in self.account_configs] else: account_ids = list(account_map.keys()) # Fetch transactions for each account for account_id in account_ids: account_name = account_map.get(account_id, account_id) try: txn_response = self.transactions_api.get_transactions_by_account( plan_id=self.plan_id, account_id=account_id, since_date=cutoff_date, ) for txn in txn_response.data.transactions: txn_date = txn.var_date if txn_date < cutoff_date: continue cleared_status = self._normalize_cleared_status(txn.cleared) if cleared_status in excluded_statuses: continue category_name = None if txn.category_id: category_name = category_map.get(str(txn.category_id)) processed: YNABTransaction = { "id": txn.id, "date": str(txn.var_date), "payee_name": txn.payee_name, "category_name": category_name, "amount_milliCurrency": txn.amount, "memo": txn.memo, "account_id": account_id, "account_name": account_name, "cleared": cleared_status, "import_id": txn.import_id, } transactions.append(processed) except Exception as e: print(f"Error fetching transactions for account {account_name}: {e}") continue return transactions def get_account_balance(self, account_id: str) -> int: """ Get current balance for an account (in milliCurrency units). Args: account_id: YNAB account ID Returns: Account balance in milliCurrency """ accounts_response = self.accounts_api.get_accounts(self.plan_id) for account in accounts_response.data.accounts: if str(account.id) == account_id: return account.balance raise ValueError(f"Account {account_id} not found") def get_last_reconciled_date( self, account_ids: list[str] | None = None, search_days: int = 90, ) -> datetime | None: """ Find the date of the most recent reconciled transaction across accounts. Args: account_ids: Account IDs to search. If None, uses all configured accounts. search_days: How far back to search for reconciled transactions (default 90). Returns: The date of the last reconciled transaction, or None if none found. """ from datetime import date as date_type cutoff_date = (datetime.now() - timedelta(days=search_days)).date() if account_ids is None: account_ids = [acc["id"] for acc in self.account_configs] latest_date: date_type | None = None for account_id in account_ids: try: txn_response = self.transactions_api.get_transactions_by_account( plan_id=self.plan_id, account_id=account_id, since_date=cutoff_date, ) for txn in txn_response.data.transactions: status = self._normalize_cleared_status(txn.cleared) if status == "reconciled": txn_date = txn.var_date if latest_date is None or txn_date > latest_date: latest_date = txn_date except Exception as e: print(f"Error searching reconciled transactions for {account_id}: {e}") continue return latest_date def create_transactions( self, transactions: list[ynab.NewTransaction], ) -> ynab.SaveTransactionsResponseData: """ Create multiple transactions in YNAB. Args: transactions: List of NewTransaction objects to create. Returns: SaveTransactionsResponseData with created IDs and duplicate info. """ wrapper = ynab.PostTransactionsWrapper(transactions=transactions) response = self.transactions_api.create_transaction( plan_id=self.plan_id, data=wrapper, ) return response.data def update_transaction( self, transaction_id: str, cleared: str | None = None, flag_color: str | None = None, ) -> None: """ Update an existing YNAB transaction. Args: transaction_id: The YNAB transaction ID to update. cleared: New cleared status ('cleared', 'uncleared', 'reconciled'). flag_color: New flag color (e.g. 'red', 'blue'). """ fields: dict = {} if cleared is not None: fields["cleared"] = ynab.TransactionClearedStatus(cleared) if flag_color is not None: fields["flag_color"] = ynab.TransactionFlagColor(flag_color) existing = ynab.ExistingTransaction(**fields) wrapper = ynab.PutTransactionWrapper(transaction=existing) self.transactions_api.update_transaction( plan_id=self.plan_id, transaction_id=transaction_id, data=wrapper, )