YNAB Assistant

"""Transaction management tools for YNAB API.""" from typing import Annotated, Any import ynab from auth import get_api_client, get_api_configuration def get_transaction_by_id( transaction_id: Annotated[ str, "The ID of the transaction to retrieve. Transaction IDs are returned " "from other transaction queries.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", ) -> dict[str, Any]: """Get a single transaction by its ID. Retrieves detailed information for a specific transaction including date, amount, payee, category, memo, cleared status, and other transaction metadata. Args: transaction_id: The ID of the transaction to retrieve. Transaction IDs are returned from other transaction queries. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. Returns: dict[str, Any]: Complete transaction data including all fields and metadata. Raises: Exception: If the API call fails or transaction not found. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transaction_by_id(budget_id, transaction_id) transaction = api_response.data.transaction # type: ignore[attr-defined] return transaction.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching transaction: {e!s}" raise Exception(msg) from e def get_transactions( budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", since_date: Annotated[ str | None, "Only include transactions on or after this date in ISO format " "YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions.", ] = None, ) -> dict[str, Any]: """Get all transactions for a budget. Retrieves all transactions for the specified budget, optionally filtered by date. This returns transactions across all accounts and categories. For more targeted queries, use the account, category, month, or payee-specific transaction tools. Args: budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. since_date: Only include transactions on or after this date in ISO format YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions. Returns: dict[str, Any]: List of all transactions with server knowledge for efficient delta synchronization. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transactions( budget_id, since_date=since_date # type: ignore[arg-type] ) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching transactions: {e!s}" raise Exception(msg) from e def get_transactions_by_account( account_id: Annotated[ str, "The ID of the account to query. Use get_accounts to discover available " "account IDs.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", since_date: Annotated[ str | None, "Only include transactions on or after this date in ISO format " "YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions.", ] = None, ) -> dict[str, Any]: """Get all transactions for a specific account. Retrieves transactions filtered by a specific account, optionally limited by date. If you don't know the account ID, use get_accounts first to list all accounts. Args: account_id: The ID of the account to query. Use get_accounts to discover available account IDs. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. since_date: Only include transactions on or after this date in ISO format YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions. Returns: dict[str, Any]: List of transactions for the specified account. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transactions_by_account( budget_id, account_id, since_date=since_date # type: ignore[arg-type] ) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching account transactions: {e!s}" raise Exception(msg) from e def get_transactions_by_category( category_id: Annotated[ str, "The ID of the category to query. Use get_categories to discover " "available category IDs.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", since_date: Annotated[ str | None, "Only include transactions on or after this date in ISO format " "YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions.", ] = None, ) -> dict[str, Any]: """Get all transactions for a specific category. Retrieves transactions filtered by a specific category, optionally limited by date. If you don't know the category ID, use get_categories first to list all categories. Args: category_id: The ID of the category to query. Use get_categories to discover available category IDs. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. since_date: Only include transactions on or after this date in ISO format YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions. Returns: dict[str, Any]: List of transactions for the specified category. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transactions_by_category( budget_id, category_id, since_date=since_date # type: ignore[arg-type] ) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching category transactions: {e!s}" raise Exception(msg) from e def get_transactions_by_month( month: Annotated[ str, "The budget month in ISO format YYYY-MM-DD (e.g., '2024-12-01') or " "'current' for the current month.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", ) -> dict[str, Any]: """Get all transactions for a specific month. Retrieves all transactions that occurred within the specified budget month, across all accounts and categories. This is useful for monthly financial analysis and reporting. Args: month: The budget month in ISO format YYYY-MM-DD (e.g., '2024-12-01') or 'current' for the current month. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. Returns: dict[str, Any]: List of all transactions for the specified month. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transactions_by_month(budget_id, month) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching month transactions: {e!s}" raise Exception(msg) from e def get_transactions_by_payee( payee_id: Annotated[ str, "The ID of the payee to query. Payee IDs can be discovered from " "transaction data or payee-related queries.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", since_date: Annotated[ str | None, "Only include transactions on or after this date in ISO format " "YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions.", ] = None, ) -> dict[str, Any]: """Get all transactions for a specific payee. Retrieves transactions filtered by a specific payee, optionally limited by date. This is useful for analyzing spending patterns with particular vendors or merchants. Args: payee_id: The ID of the payee to query. Payee IDs can be discovered from transaction data or payee-related queries. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. since_date: Only include transactions on or after this date in ISO format YYYY-MM-DD (e.g., '2024-01-01'). Use None to retrieve all transactions. Returns: dict[str, Any]: List of transactions for the specified payee. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.get_transactions_by_payee( budget_id, payee_id, since_date=since_date # type: ignore[arg-type] ) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error fetching payee transactions: {e!s}" raise Exception(msg) from e def create_transaction( account_id: Annotated[ str, "The ID of the account for this transaction. Use get_accounts to " "discover available account IDs.", ], date: Annotated[ str, "Transaction date in ISO format YYYY-MM-DD (e.g., '2024-12-01').", ], amount: Annotated[ int, "Transaction amount in milliunits. Negative for outflows (spending), " "positive for inflows (income).", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", payee_id: Annotated[ str | None, "The ID of the payee for this transaction. Leave as None if not applicable.", ] = None, category_id: Annotated[ str | None, "The ID of the category for this transaction. Use get_categories to " "discover category IDs. Leave as None if uncategorized.", ] = None, memo: Annotated[ str | None, "Optional memo/note for this transaction. Leave as None for no memo.", ] = None, cleared: Annotated[ str, "Cleared status: 'cleared', 'uncleared', or 'reconciled'. Defaults to " "'uncleared'.", ] = "uncleared", approved: Annotated[ bool, "Whether the transaction is approved. Unapproved transactions may appear " "differently in YNAB.", ] = False, ) -> dict[str, Any]: """Create a new transaction in YNAB. Amounts in YNAB are expressed in milliunits: 1000 milliunits = $1.00. For example, to record spending $50.00, use -50000 (negative for outflows). To record income of $125.00, use 125000 (positive for inflows). Args: account_id: The ID of the account for this transaction. Use get_accounts to discover available account IDs. date: Transaction date in ISO format YYYY-MM-DD (e.g., '2024-12-01'). amount: Transaction amount in milliunits. Negative for outflows (spending), positive for inflows (income). budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. payee_id: The ID of the payee for this transaction. Leave as None if not applicable. category_id: The ID of the category for this transaction. Use get_categories to discover category IDs. Leave as None if uncategorized. memo: Optional memo/note for this transaction. Leave as None for no memo. cleared: Cleared status: 'cleared', 'uncleared', or 'reconciled'. Defaults to 'uncleared'. approved: Whether the transaction is approved. Unapproved transactions may appear differently in YNAB. Returns: dict[str, Any]: Created transaction data with assigned transaction ID and all field values. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: # Build transaction data transaction_data = { "account_id": account_id, "date": date, "amount": amount, "cleared": cleared, "approved": approved, } # Add optional fields if provided if payee_id: transaction_data["payee_id"] = payee_id if category_id: transaction_data["category_id"] = category_id if memo: transaction_data["memo"] = memo # Create wrapper with transaction wrapper = ynab.PostTransactionsWrapper(transaction=transaction_data) # type: ignore[arg-type] api_response = api_instance.create_transaction(budget_id, wrapper) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error creating transaction: {e!s}" raise Exception(msg) from e def create_transactions( transactions: Annotated[ list[dict[str, Any]], "List of transaction dictionaries. Each must have account_id, date (ISO " "format YYYY-MM-DD), and amount (in milliunits). Optional fields: " "payee_id, payee_name, category_id, memo, cleared, approved, flag_color.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", ) -> dict[str, Any]: """Create multiple transactions in a single API call (bulk import). Amounts in YNAB are expressed in milliunits: 1000 milliunits = $1.00. For example, to record spending $50.00, use -50000 (negative for outflows). To record income of $125.00, use 125000 (positive for inflows). This is more efficient than creating transactions one at a time when you need to import multiple transactions. Args: transactions: List of transaction dictionaries. Each must have account_id, date (ISO format YYYY-MM-DD), and amount (in milliunits). Optional fields: payee_id, payee_name, category_id, memo, cleared, approved, flag_color. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. Returns: dict[str, Any]: Created transactions data including the list of created transactions, any duplicate_import_ids encountered, and server_knowledge for synchronization. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: # Create wrapper with transactions list wrapper = ynab.PostTransactionsWrapper(transactions=transactions) # type: ignore[arg-type] api_response = api_instance.create_transaction(budget_id, wrapper) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error creating transactions: {e!s}" raise Exception(msg) from e def update_transaction( transaction_id: Annotated[ str, "The ID of the transaction to update. Transaction IDs are returned from " "other transaction queries.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", account_id: Annotated[ str | None, "The ID of the account to move this transaction to. Leave as None to " "keep unchanged.", ] = None, date: Annotated[ str | None, "Transaction date in ISO format YYYY-MM-DD (e.g., '2024-12-01'). Leave " "as None to keep unchanged.", ] = None, amount: Annotated[ int | None, "Transaction amount in milliunits. Leave as None to keep unchanged.", ] = None, payee_id: Annotated[ str | None, "The ID of the payee. Leave as None to keep unchanged.", ] = None, category_id: Annotated[ str | None, "The ID of the category. Leave as None to keep unchanged.", ] = None, memo: Annotated[ str | None, "Transaction memo/note. Leave as None to keep unchanged.", ] = None, cleared: Annotated[ str | None, "Cleared status: 'cleared', 'uncleared', or 'reconciled'. Leave as None " "to keep unchanged.", ] = None, approved: Annotated[ bool | None, "Whether the transaction is approved. Leave as None to keep unchanged.", ] = None, ) -> dict[str, Any]: """Update an existing transaction. Amounts in YNAB are expressed in milliunits: 1000 milliunits = $1.00. Only provide the fields you want to update; fields left as None will remain unchanged. For example, to only update the amount, provide transaction_id and amount, leaving all other fields as None. Args: transaction_id: The ID of the transaction to update. Transaction IDs are returned from other transaction queries. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. account_id: The ID of the account to move this transaction to. Leave as None to keep unchanged. date: Transaction date in ISO format YYYY-MM-DD (e.g., '2024-12-01'). Leave as None to keep unchanged. amount: Transaction amount in milliunits. Leave as None to keep unchanged. payee_id: The ID of the payee. Leave as None to keep unchanged. category_id: The ID of the category. Leave as None to keep unchanged. memo: Transaction memo/note. Leave as None to keep unchanged. cleared: Cleared status: 'cleared', 'uncleared', or 'reconciled'. Leave as None to keep unchanged. approved: Whether the transaction is approved. Leave as None to keep unchanged. Returns: dict[str, Any]: Updated transaction data with all current field values. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: # Build transaction data with only provided fields transaction_data: dict[str, Any] = {} if account_id is not None: transaction_data["account_id"] = account_id if date is not None: transaction_data["date"] = date if amount is not None: transaction_data["amount"] = amount if payee_id is not None: transaction_data["payee_id"] = payee_id if category_id is not None: transaction_data["category_id"] = category_id if memo is not None: transaction_data["memo"] = memo if cleared is not None: transaction_data["cleared"] = cleared if approved is not None: transaction_data["approved"] = approved # Create wrapper wrapper = ynab.PutTransactionWrapper(transaction=transaction_data) # type: ignore[arg-type] api_response = api_instance.update_transaction( budget_id, transaction_id, wrapper ) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error updating transaction: {e!s}" raise Exception(msg) from e def update_transactions( transactions: Annotated[ list[dict[str, Any]], "List of transaction dictionaries. Each must have 'id' and the fields to " "update (e.g., amount, date, category_id, memo, cleared, approved). Only " "include fields you want to change.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", ) -> dict[str, Any]: """Update multiple transactions in a single API call (bulk update). Amounts in YNAB are expressed in milliunits: 1000 milliunits = $1.00. This is more efficient than updating transactions one at a time when you need to modify multiple transactions. Args: transactions: List of transaction dictionaries. Each must have 'id' and the fields to update (e.g., amount, date, category_id, memo, cleared, approved). Only include fields you want to change. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. Returns: dict[str, Any]: Updated transactions data including all modified transactions with their current values. Raises: Exception: If the API call fails. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: # Create wrapper with transactions list wrapper = ynab.PatchTransactionsWrapper(transactions=transactions) # type: ignore[arg-type] api_response = api_instance.update_transactions(budget_id, wrapper) # Check if response is None (defensive check despite type annotation) if api_response is None: # type: ignore[reportUnnecessaryComparison] msg = "API returned None response" raise Exception(msg) # Check if response has data attribute if not hasattr(api_response, "data"): msg = ( f"API response missing data attribute. " f"Response type: {type(api_response)}, " f"Response: {api_response}" ) raise Exception(msg) return api_response.data.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error updating transactions: {e!s}" raise Exception(msg) from e def delete_transaction( transaction_id: Annotated[ str, "The ID of the transaction to delete. Transaction IDs are returned from " "other transaction queries.", ], budget_id: Annotated[ str, "The ID of the budget to query. Use 'last-used' for the most recently " "accessed budget, or provide a specific budget ID.", ] = "last-used", ) -> dict[str, Any]: """Delete a transaction (destructive operation). This permanently removes a transaction from YNAB. This operation cannot be undone via the API. Use with caution. Args: transaction_id: The ID of the transaction to delete. Transaction IDs are returned from other transaction queries. budget_id: The ID of the budget to query. Use 'last-used' for the most recently accessed budget, or provide a specific budget ID. Returns: dict[str, Any]: Deleted transaction data showing the transaction that was removed. Raises: Exception: If the API call fails or transaction not found. """ configuration = get_api_configuration() with get_api_client(configuration) as api_client: api_instance = ynab.TransactionsApi(api_client) try: api_response = api_instance.delete_transaction(budget_id, transaction_id) transaction = api_response.data.transaction # type: ignore[attr-defined] return transaction.model_dump() # type: ignore[attr-defined] except Exception as e: msg = f"Error deleting transaction: {e!s}" raise Exception(msg) from e