Add ynab budgets and accounts commands

requirements.txt

@@ -1,3 +1,4 @@
 appeal
 pydantic
 requests
+tabulate

ynab_sync/cli.py

@@ -1,22 +1,25 @@
+import decimal
 import logging
 import os
 import sys
 from datetime import date
 from uuid import UUID
 
 import appeal
+from tabulate import tabulate
 
 app = appeal.Appeal()
 
 import logging
 
 logging.basicConfig(
-    level=logging.DEBUG,
+    level=logging.INFO,
     format="%(asctime)s %(name)s [%(levelname)s] %(message)s",
     handlers=[logging.StreamHandler(sys.stdout)],
 )
 
-from .logic import (get_gocardless_transactions, prepare_ynab_transactions,
+from .logic import (get_gocardless_transactions, get_ynab_budget,
+                    get_ynab_budgets, prepare_ynab_transactions,
                     upload_to_ynab)
 
 
@@ -78,3 +81,45 @@ def upload(
     upload_to_ynab(
         transactions=ynab_transactions, token=ynab_token, budget_id=UUID(ynab_budget_id)
     )
+
+
+@app.command()
+def ynab():
+    pass
+
+
+@app.command("ynab").command()
+def budgets(
+    *,
+    ynab_token: str = "",
+):
+    budgets = get_ynab_budgets(token=ynab_token)
+
+    table = tabulate(
+        [(budget.id, budget.name, budget.currency) for budget in budgets],
+        headers=["ID", "NAME", "CURRENCY"],
+        tablefmt="github",
+    )
+    print(table)
+
+
+@app.command("ynab").command()
+def accounts(*, ynab_token: str = "", ynab_budget_id: str = ""):
+    budget = get_ynab_budget(token=ynab_token, budget_id=UUID(ynab_budget_id))
+
+    table = tabulate(
+        [
+            (
+                account.id,
+                account.name,
+                account.type,
+                f"{budget.currency_symbol}{account.balance / decimal.Decimal(1000)}",
+                account.closed,
+                account.deleted,
+            )
+            for account in budget.accounts
+        ],
+        headers=["ID", "NAME", "TYPE", "BALANCE", "CLOSED", "DELETED"],
+        tablefmt="github",
+    )
+    print(table)

ynab_sync/logic.py

@@ -9,7 +9,8 @@
 
 from .gocardless.api import GoCardLessAPI
 from .ynab.api import YnabAPI
-from .ynab.models import YNABTransaction, YNABTransactions
+from .ynab.models import (YNABAccount, YNABBudget, YNABTransaction,
+                          YNABTransactions)
 
 
 def get_gocardless_transactions(
@@ -96,3 +97,13 @@ def upload_to_ynab(
         raise
 
     log.debug("YNAB response: %s", response)
+
+
+def get_ynab_budgets(token: str) -> list[YNABBudget]:
+    ynab_api = YnabAPI(access_token=token)
+    return ynab_api.get_budgets()
+
+
+def get_ynab_budget(token: str, budget_id: UUID) -> YNABBudget:
+    ynab_api = YnabAPI(access_token=token)
+    return ynab_api.get_budget(budget_id=budget_id)

ynab_sync/ynab/api.py

@@ -2,6 +2,7 @@
 import uuid
 
 import requests
+from ynab_sync.ynab.models import YNABAccount, YNABBudget
 
 BASE_URL = "https://api.ynab.com/v1"
 
@@ -43,3 +44,13 @@ def post_transactions(
         response.raise_for_status()
 
         return response.json()
+
+    def get_budget(self, budget_id: uuid.UUID) -> YNABBudget:
+        response = self._requests_session.get(f"{BASE_URL}/budgets/{budget_id}")
+        response.raise_for_status()
+        return YNABBudget(**response.json()["data"]["budget"])
+
+    def get_budgets(self) -> list[YNABBudget]:
+        response = self._requests_session.get(f"{BASE_URL}/budgets")
+        response.raise_for_status()
+        return [YNABBudget(**budget) for budget in response.json()["data"]["budgets"]]

ynab_sync/ynab/models.py

@@ -1,4 +1,4 @@
-from datetime import date
+from datetime import date, datetime
 from uuid import UUID
 
 from pydantic import BaseModel, Field
@@ -17,3 +17,31 @@ class YNABTransaction(BaseModel):
 
 class YNABTransactions(BaseModel):
     transactions: list[YNABTransaction]
+
+
+class YNABAccount(BaseModel):
+    id: UUID
+    name: str
+    type: str
+    balance: int
+    closed: bool
+    deleted: bool
+
+
+class YNABBudget(BaseModel):
+    id: UUID
+    name: str
+    last_modified_on: datetime
+    first_month: date
+    last_month: date
+    date_format: dict
+    currency_format: dict
+    accounts: list[YNABAccount] = Field(default_factory=list)
+
+    @property
+    def currency(self):
+        return self.currency_format["iso_code"]
+
+    @property
+    def currency_symbol(self):
+        return self.currency_format["currency_symbol"]