Add gocardless requisition commands

ynab_sync/cli.py

@@ -18,7 +18,8 @@
     handlers=[logging.StreamHandler(sys.stdout)],
 )
 
-from .logic import (get_gocardless_banks, get_gocardless_transactions,
+from .logic import (create_gocardless_requisition, get_gocardless_banks,
+                    get_gocardless_requisition, get_gocardless_transactions,
                     get_ynab_budget, get_ynab_budgets,
                     prepare_ynab_transactions, upload_to_ynab)
 
@@ -145,3 +146,47 @@ def banks(
     table = tabulate([(bank.id, bank.name) for bank in banks], headers=["ID", "NAME"])
 
     print(table)
+
+
+@app.command("gocardless").command()
+def generate_bank_auth_link(
+    institution_id: str,
+    *,
+    gocardless_secret_id: str = "",
+    gocardless_secret_key: str = "",
+):
+    requisition = create_gocardless_requisition(
+        secret_id=gocardless_secret_id,
+        secret_key=gocardless_secret_key,
+        redirect="",
+        institution_id=institution_id,
+    )
+
+    print("GOCARDLESS_REQUISITION_ID: ", requisition.id)
+    print(
+        "Open this link in your browser and proceed with authorization:",
+        requisition.link,
+    )
+    print(
+        "Run `gocardless list_requisition_accounts` afterwards to get GOCARDLESS_ACCOUNT_ID"
+    )
+
+
+@app.command("gocardless").command()
+def list_requisition_accounts(
+    requisition_id: str,
+    *,
+    gocardless_secret_id: str = "",
+    gocardless_secret_key: str = "",
+):
+    requisition = get_gocardless_requisition(
+        secret_id=gocardless_secret_id,
+        secret_key=gocardless_secret_key,
+        requisition_id=requisition_id,
+    )
+
+    table = tabulate(
+        [(requisition.id, account) for account in requisition.accounts],
+        headers=["GOCARDLESS_REQUISITION_ID", "GOCARDLESS_ACCOUNT_ID"],
+    )
+    print(table)

ynab_sync/gocardless/api.py

@@ -3,7 +3,8 @@
 
 import requests
 
-from .models import GoCardlessBankAccountData, GoCardlessInstitution
+from .models import (GoCardlessBankAccountData, GoCardlessInstitution,
+                     GoCardlessRequisition)
 
 BASE_URL = "https://bankaccountdata.gocardless.com/api/v2"
 
@@ -75,3 +76,20 @@ def get_banks(self, country: str) -> list[GoCardlessInstitution]:
         response.raise_for_status()
         json_data = response.json()
         return [GoCardlessInstitution(**institution) for institution in json_data]
+
+    def get_requisition(self, requisition_id: str) -> GoCardlessRequisition:
+        response = self._requests_session.get(
+            f"{BASE_URL}/requisitions/{requisition_id}",
+        )
+        response.raise_for_status()
+        return GoCardlessRequisition(**response.json())
+
+    def post_requisition(
+        self, redirect: str, institution_id: str
+    ) -> GoCardlessRequisition:
+        response = self._requests_session.post(
+            f"{BASE_URL}/requisitions/",
+            json={"redirect": redirect, "institution_id": institution_id},
+        )
+        response.raise_for_status()
+        return GoCardlessRequisition(**response.json())

ynab_sync/gocardless/models.py

@@ -1,4 +1,4 @@
-from datetime import date
+from datetime import date, datetime
 from decimal import Decimal
 
 from pydantic import BaseModel, Field
@@ -38,3 +38,13 @@ class GoCardlessBankAccountData(BaseModel):
 class GoCardlessInstitution(BaseModel):
     id: str
     name: str
+
+
+class GoCardlessRequisition(BaseModel):
+    id: str
+    created: datetime
+    redirect: str
+    status: str
+    institution_id: str
+    link: str
+    accounts: list[str] = Field(default_factory=list)

ynab_sync/logic.py

@@ -5,7 +5,8 @@
 
 from requests import HTTPError
 
-from ynab_sync.gocardless.models import GoCardlessBankAccountData
+from ynab_sync.gocardless.models import (GoCardlessBankAccountData,
+                                         GoCardlessRequisition)
 
 from .gocardless.api import GoCardLessAPI
 from .ynab.api import YnabAPI
@@ -112,3 +113,22 @@ def get_ynab_budget(token: str, budget_id: UUID) -> YNABBudget:
 def get_gocardless_banks(secret_id: str, secret_key: str, country: str):
     gocardless_api = GoCardLessAPI(secret_id=secret_id, secret_key=secret_key)
     return gocardless_api.get_banks(country=country)
+
+
+def create_gocardless_requisition(
+    secret_id: str,
+    secret_key: str,
+    institution_id: str,
+    redirect: str,
+) -> GoCardlessRequisition:
+    gocardless_api = GoCardLessAPI(secret_id=secret_id, secret_key=secret_key)
+    return gocardless_api.post_requisition(
+        redirect=redirect, institution_id=institution_id
+    )
+
+
+def get_gocardless_requisition(
+    secret_id: str, secret_key: str, requisition_id: str
+) -> GoCardlessRequisition:
+    gocardless_api = GoCardLessAPI(secret_id=secret_id, secret_key=secret_key)
+    return gocardless_api.get_requisition(requisition_id=requisition_id)