80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Iterable
|
||
|
|
from urllib.parse import urljoin
|
||
|
|
import requests
|
||
|
|
|
||
|
|
|
||
|
|
class FireflyClient:
|
||
|
|
def __init__(self, base_url: str, api_token: str, page_limit: int = 200):
|
||
|
|
self.base_url = self._normalize_base_url(base_url)
|
||
|
|
self.page_limit = page_limit
|
||
|
|
self.session = requests.Session()
|
||
|
|
self.session.headers.update({
|
||
|
|
"Authorization": f"Bearer {api_token}",
|
||
|
|
"Accept": "application/json",
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
})
|
||
|
|
|
||
|
|
def list_accounts(self, types: list[str] | None = None) -> list[dict[str, Any]]:
|
||
|
|
params: dict[str, Any] | None = None
|
||
|
|
if types:
|
||
|
|
params = {"type": ",".join(types)}
|
||
|
|
return list(self._list_all("accounts", params=params))
|
||
|
|
|
||
|
|
def get_account(self, account_id: str) -> dict[str, Any]:
|
||
|
|
return self._request("GET", f"accounts/{account_id}")
|
||
|
|
|
||
|
|
def create_account(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("POST", "accounts", json=payload)
|
||
|
|
|
||
|
|
def update_account(self, account_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("PUT", f"accounts/{account_id}", json=payload)
|
||
|
|
|
||
|
|
def list_categories(self) -> list[dict[str, Any]]:
|
||
|
|
return list(self._list_all("categories"))
|
||
|
|
|
||
|
|
def create_category(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("POST", "categories", json=payload)
|
||
|
|
|
||
|
|
def create_transaction(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("POST", "transactions", json=payload)
|
||
|
|
|
||
|
|
def search_transactions(self, query: str) -> dict[str, Any]:
|
||
|
|
return self._request("GET", "search/transactions", params={"query": query})
|
||
|
|
|
||
|
|
def _list_all(self, path: str, params: dict[str, Any] | None = None) -> Iterable[dict[str, Any]]:
|
||
|
|
page = 1
|
||
|
|
while True:
|
||
|
|
merged = {"page": page, "limit": self.page_limit}
|
||
|
|
if params:
|
||
|
|
merged.update(params)
|
||
|
|
payload = self._request("GET", path, params=merged)
|
||
|
|
data = payload.get("data") or []
|
||
|
|
if not data:
|
||
|
|
break
|
||
|
|
for item in data:
|
||
|
|
yield item
|
||
|
|
if len(data) < self.page_limit:
|
||
|
|
break
|
||
|
|
page += 1
|
||
|
|
|
||
|
|
def _request(self, method: str, path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
|
|
url = urljoin(f"{self.base_url}/", path)
|
||
|
|
response = self.session.request(method, url, params=params, json=json, timeout=30)
|
||
|
|
if response.status_code >= 400:
|
||
|
|
detail = response.text.strip()
|
||
|
|
raise RuntimeError(f"Firefly API {method} {url} failed: {response.status_code} {detail}")
|
||
|
|
if not response.content:
|
||
|
|
return {}
|
||
|
|
return response.json()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _normalize_base_url(base_url: str) -> str:
|
||
|
|
base = base_url.rstrip("/")
|
||
|
|
if base.endswith("/api/v1"):
|
||
|
|
return base
|
||
|
|
if base.endswith("/api"):
|
||
|
|
return f"{base}/v1"
|
||
|
|
return f"{base}/api/v1"
|