nikdoof/leggen
1
1
Fork 0
mirror of https://github.com/elisiariocouto/leggen.git synced 2025-12-25 09:49:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: Implement code review suggestions and format code.

Browse Source
This commit is contained in:
Elisiário Couto
2025-09-03 21:11:19 +01:00
committed by Elisiário Couto
parent 47164e8546
commit de3da84dff
42 changed files with 1144 additions and 966 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
leggen/api_client.py
View File

@@ -8,19 +8,20 @@ from leggen.utils.text import error
class LeggendAPIClient:
"""Client for communicating with the leggend FastAPI service"""
def __init__(self, base_url: Optional[str] = None):
self.base_url = base_url or os.environ.get("LEGGEND_API_URL", "http://localhost:8000")
self.base_url = base_url or os.environ.get(
"LEGGEND_API_URL", "http://localhost:8000"
)
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
self.session.headers.update(
{"Content-Type": "application/json", "Accept": "application/json"}
)
def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request to the API"""
url = urljoin(self.base_url, endpoint)
try:
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
@@ -53,15 +54,19 @@ class LeggendAPIClient:
# Bank endpoints
def get_institutions(self, country: str = "PT") -> List[Dict[str, Any]]:
"""Get bank institutions for a country"""
response = self._make_request("GET", "/api/v1/banks/institutions", params={"country": country})
response = self._make_request(
"GET", "/api/v1/banks/institutions", params={"country": country}
)
return response.get("data", [])
def connect_to_bank(self, institution_id: str, redirect_url: str = "http://localhost:8000/") -> Dict[str, Any]:
def connect_to_bank(
self, institution_id: str, redirect_url: str = "http://localhost:8000/"
) -> Dict[str, Any]:
"""Connect to a bank"""
response = self._make_request(
"POST",
"POST",
"/api/v1/banks/connect",
json={"institution_id": institution_id, "redirect_url": redirect_url}
json={"institution_id": institution_id, "redirect_url": redirect_url},
)
return response.get("data", {})
@@ -91,31 +96,39 @@ class LeggendAPIClient:
response = self._make_request("GET", f"/api/v1/accounts/{account_id}/balances")
return response.get("data", [])
def get_account_transactions(self, account_id: str, limit: int = 100, summary_only: bool = False) -> List[Dict[str, Any]]:
def get_account_transactions(
self, account_id: str, limit: int = 100, summary_only: bool = False
) -> List[Dict[str, Any]]:
"""Get account transactions"""
response = self._make_request(
"GET",
"GET",
f"/api/v1/accounts/{account_id}/transactions",
params={"limit": limit, "summary_only": summary_only}
params={"limit": limit, "summary_only": summary_only},
)
return response.get("data", [])
# Transaction endpoints
def get_all_transactions(self, limit: int = 100, summary_only: bool = True, **filters) -> List[Dict[str, Any]]:
# Transaction endpoints
def get_all_transactions(
self, limit: int = 100, summary_only: bool = True, **filters
) -> List[Dict[str, Any]]:
"""Get all transactions with optional filters"""
params = {"limit": limit, "summary_only": summary_only}
params.update(filters)
response = self._make_request("GET", "/api/v1/transactions", params=params)
return response.get("data", [])
def get_transaction_stats(self, days: int = 30, account_id: Optional[str] = None) -> Dict[str, Any]:
def get_transaction_stats(
self, days: int = 30, account_id: Optional[str] = None
) -> Dict[str, Any]:
"""Get transaction statistics"""
params = {"days": days}
if account_id:
params["account_id"] = account_id
response = self._make_request("GET", "/api/v1/transactions/stats", params=params)
response = self._make_request(
"GET", "/api/v1/transactions/stats", params=params
)
return response.get("data", {})
# Sync endpoints
@@ -124,21 +137,25 @@ class LeggendAPIClient:
response = self._make_request("GET", "/api/v1/sync/status")
return response.get("data", {})
def trigger_sync(self, account_ids: Optional[List[str]] = None, force: bool = False) -> Dict[str, Any]:
def trigger_sync(
self, account_ids: Optional[List[str]] = None, force: bool = False
) -> Dict[str, Any]:
"""Trigger a sync"""
data = {"force": force}
if account_ids:
data["account_ids"] = account_ids
response = self._make_request("POST", "/api/v1/sync", json=data)
return response.get("data", {})
def sync_now(self, account_ids: Optional[List[str]] = None, force: bool = False) -> Dict[str, Any]:
def sync_now(
self, account_ids: Optional[List[str]] = None, force: bool = False
) -> Dict[str, Any]:
"""Run sync synchronously"""
data = {"force": force}
if account_ids:
data["account_ids"] = account_ids
response = self._make_request("POST", "/api/v1/sync/now", json=data)
return response.get("data", {})
@@ -147,11 +164,17 @@ class LeggendAPIClient:
response = self._make_request("GET", "/api/v1/sync/scheduler")
return response.get("data", {})
def update_scheduler_config(self, enabled: bool = True, hour: int = 3, minute: int = 0, cron: Optional[str] = None) -> Dict[str, Any]:
def update_scheduler_config(
self,
enabled: bool = True,
hour: int = 3,
minute: int = 0,
cron: Optional[str] = None,
) -> Dict[str, Any]:
"""Update scheduler configuration"""
data = {"enabled": enabled, "hour": hour, "minute": minute}
if cron:
data["cron"] = cron
response = self._make_request("PUT", "/api/v1/sync/scheduler", json=data)
return response.get("data", {})
return response.get("data", {})

12
leggen/commands/balances.py
View File

@@ -12,10 +12,12 @@ def balances(ctx: click.Context):
List balances of all connected accounts
"""
api_client = LeggendAPIClient(ctx.obj.get("api_url"))
# Check if leggend service is available
if not api_client.health_check():
click.echo("Error: Cannot connect to leggend service. Please ensure it's running.")
click.echo(
"Error: Cannot connect to leggend service. Please ensure it's running."
)
return
accounts = api_client.get_accounts()
@@ -24,11 +26,7 @@ def balances(ctx: click.Context):
for account in accounts:
for balance in account.get("balances", []):
amount = round(float(balance["amount"]), 2)
symbol = (
"€"
if balance["currency"] == "EUR"
else f" {balance['currency']}"
)
symbol = "€" if balance["currency"] == "EUR" else f" {balance['currency']}"
amount_str = f"{amount}{symbol}"
date = (
datefmt(balance.get("last_change_date"))

36
leggen/commands/bank/__init__.py
View File

@@ -1,36 +0,0 @@
import os
import click
from leggen.main import cli
cmd_folder = os.path.abspath(os.path.dirname(__file__))
class BankGroup(click.Group):
def list_commands(self, ctx):
rv = []
for filename in os.listdir(cmd_folder):
if filename.endswith(".py") and not filename.startswith("__init__"):
if filename == "list_banks.py":
rv.append("list")
else:
rv.append(filename[:-3])
rv.sort()
return rv
def get_command(self, ctx, name):
try:
if name == "list":
name = "list_banks"
mod = __import__(f"leggen.commands.bank.{name}", None, None, [name])
except ImportError:
return
return getattr(mod, name)
@cli.group(cls=BankGroup)
@click.pass_context
def bank(ctx):
"""Manage banks connections"""
return

30
leggen/commands/bank/add.py
View File

@@ -13,30 +13,32 @@ def add(ctx):
Connect to a bank
"""
api_client = LeggendAPIClient(ctx.obj.get("api_url"))
# Check if leggend service is available
if not api_client.health_check():
click.echo("Error: Cannot connect to leggend service. Please ensure it's running.")
click.echo(
"Error: Cannot connect to leggend service. Please ensure it's running."
)
return
try:
# Get supported countries
countries = api_client.get_supported_countries()
country_codes = [c["code"] for c in countries]
country = click.prompt(
"Bank Country",
type=click.Choice(country_codes, case_sensitive=True),
default="PT",
)
info(f"Getting bank list for country: {country}")
banks = api_client.get_institutions(country)
if not banks:
warning(f"No banks available for country {country}")
return
filtered_banks = [
{
"id": bank["id"],
@@ -46,14 +48,14 @@ def add(ctx):
for bank in banks
]
print_table(filtered_banks)
allowed_ids = [str(bank["id"]) for bank in banks]
bank_id = click.prompt("Bank ID", type=click.Choice(allowed_ids))
# Show bank details
selected_bank = next(bank for bank in banks if bank["id"] == bank_id)
info(f"Selected bank: {selected_bank['name']}")
click.confirm("Do you agree to connect to this bank?", abort=True)
info(f"Connecting to bank with ID: {bank_id}")
@@ -65,11 +67,15 @@ def add(ctx):
save_file(f"req_{result['id']}.json", result)
success("Bank connection request created successfully!")
warning(f"Please open the following URL in your browser to complete the authorization:")
warning(
f"Please open the following URL in your browser to complete the authorization:"
)
click.echo(f"\n{result['link']}\n")
info(f"Requisition ID: {result['id']}")
info("After completing the authorization, you can check the connection status with 'leggen status'")
info(
"After completing the authorization, you can check the connection status with 'leggen status'"
)
except Exception as e:
click.echo(f"Error: Failed to connect to bank: {str(e)}")

6
leggen/commands/status.py
View File

@@ -12,10 +12,12 @@ def status(ctx: click.Context):
List all connected banks and their status
"""
api_client = LeggendAPIClient(ctx.obj.get("api_url"))
# Check if leggend service is available
if not api_client.health_check():
click.echo("Error: Cannot connect to leggend service. Please ensure it's running.")
click.echo(
"Error: Cannot connect to leggend service. Please ensure it's running."
)
return
# Get bank connection status

28
leggen/commands/sync.py
View File

@@ -6,15 +6,15 @@ from leggen.utils.text import error, info, success
@cli.command()
@click.option('--wait', is_flag=True, help='Wait for sync to complete (synchronous)')
@click.option('--force', is_flag=True, help='Force sync even if already running')
@click.option("--wait", is_flag=True, help="Wait for sync to complete (synchronous)")
@click.option("--force", is_flag=True, help="Force sync even if already running")
@click.pass_context
def sync(ctx: click.Context, wait: bool, force: bool):
"""
Sync all transactions with database
"""
api_client = LeggendAPIClient(ctx.obj.get("api_url"))
# Check if leggend service is available
if not api_client.health_check():
error("Cannot connect to leggend service. Please ensure it's running.")
@@ -25,35 +25,37 @@ def sync(ctx: click.Context, wait: bool, force: bool):
# Run sync synchronously and wait for completion
info("Starting synchronous sync...")
result = api_client.sync_now(force=force)
if result.get("success"):
success(f"Sync completed successfully!")
info(f"Accounts processed: {result.get('accounts_processed', 0)}")
info(f"Transactions added: {result.get('transactions_added', 0)}")
info(f"Balances updated: {result.get('balances_updated', 0)}")
if result.get('duration_seconds'):
if result.get("duration_seconds"):
info(f"Duration: {result['duration_seconds']:.2f} seconds")
if result.get('errors'):
if result.get("errors"):
error(f"Errors encountered: {len(result['errors'])}")
for err in result['errors']:
for err in result["errors"]:
error(f" - {err}")
else:
error("Sync failed")
if result.get('errors'):
for err in result['errors']:
if result.get("errors"):
for err in result["errors"]:
error(f" - {err}")
else:
# Trigger async sync
info("Starting background sync...")
result = api_client.trigger_sync(force=force)
if result.get("sync_started"):
success("Sync started successfully in the background")
info("Use 'leggen sync --wait' to run synchronously or check status with API")
info(
"Use 'leggen sync --wait' to run synchronously or check status with API"
)
else:
error("Failed to start sync")
except Exception as e:
error(f"Sync failed: {str(e)}")
return

54
leggen/commands/transactions.py
View File

@@ -7,7 +7,9 @@ from leggen.utils.text import datefmt, info, print_table
@cli.command()
@click.option("-a", "--account", type=str, help="Account ID")
@click.option("-l", "--limit", type=int, default=50, help="Number of transactions to show")
@click.option(
"-l", "--limit", type=int, default=50, help="Number of transactions to show"
)
@click.option("--full", is_flag=True, help="Show full transaction details")
@click.pass_context
def transactions(ctx: click.Context, account: str, limit: int, full: bool):
@@ -19,10 +21,12 @@ def transactions(ctx: click.Context, account: str, limit: int, full: bool):
If the --account option is used, it will only list transactions for that account.
"""
api_client = LeggendAPIClient(ctx.obj.get("api_url"))
# Check if leggend service is available
if not api_client.health_check():
click.echo("Error: Cannot connect to leggend service. Please ensure it's running.")
click.echo(
"Error: Cannot connect to leggend service. Please ensure it's running."
)
return
try:
@@ -32,16 +36,14 @@ def transactions(ctx: click.Context, account: str, limit: int, full: bool):
transactions_data = api_client.get_account_transactions(
account, limit=limit, summary_only=not full
)
info(f"Bank: {account_details['institution_id']}")
info(f"IBAN: {account_details.get('iban', 'N/A')}")
else:
# Get all transactions
transactions_data = api_client.get_all_transactions(
limit=limit,
summary_only=not full,
account_id=account
limit=limit, summary_only=not full, account_id=account
)
# Format transactions for display
@@ -49,24 +51,32 @@ def transactions(ctx: click.Context, account: str, limit: int, full: bool):
# Full transaction details
formatted_transactions = []
for txn in transactions_data:
formatted_transactions.append({
"ID": txn["internal_transaction_id"][:12] + "...",
"Date": datefmt(txn["transaction_date"]),
"Description": txn["description"][:50] + "..." if len(txn["description"]) > 50 else txn["description"],
"Amount": f"{txn['transaction_value']:.2f} {txn['transaction_currency']}",
"Status": txn["transaction_status"].upper(),
"Account": txn["account_id"][:8] + "...",
})
formatted_transactions.append(
{
"ID": txn["internal_transaction_id"][:12] + "...",
"Date": datefmt(txn["transaction_date"]),
"Description": txn["description"][:50] + "..."
if len(txn["description"]) > 50
else txn["description"],
"Amount": f"{txn['transaction_value']:.2f} {txn['transaction_currency']}",
"Status": txn["transaction_status"].upper(),
"Account": txn["account_id"][:8] + "...",
}
)
else:
# Summary view
formatted_transactions = []
for txn in transactions_data:
formatted_transactions.append({
"Date": datefmt(txn["date"]),
"Description": txn["description"][:60] + "..." if len(txn["description"]) > 60 else txn["description"],
"Amount": f"{txn['amount']:.2f} {txn['currency']}",
"Status": txn["status"].upper(),
})
formatted_transactions.append(
{
"Date": datefmt(txn["date"]),
"Description": txn["description"][:60] + "..."
if len(txn["description"]) > 60
else txn["description"],
"Amount": f"{txn['amount']:.2f} {txn['currency']}",
"Status": txn["status"].upper(),
}
)
if formatted_transactions:
print_table(formatted_transactions)

6
leggen/main.py
View File

@@ -90,10 +90,10 @@ class Group(click.Group):
@click.option(
"--api-url",
type=str,
default=None,
default="http://localhost:8000",
envvar="LEGGEND_API_URL",
show_envvar=True,
help="URL of the leggend API service (default: http://localhost:8000)",
help="URL of the leggend API service",
)
@click.group(
cls=Group,
@@ -113,7 +113,7 @@ def cli(ctx: click.Context, api_url: str):
# Store API URL in context for commands to use
if api_url:
ctx.obj["api_url"] = api_url
# For backwards compatibility, still support direct GoCardless calls
# This will be used as fallback if leggend service is not available
try: