nikdoof/leggen
1
1
Fork 0
mirror of https://github.com/elisiariocouto/leggen.git synced 2025-12-11 17:22:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add mypy to pre-commit.

Browse Source
This commit is contained in:
Elisiário Couto
2025-09-03 21:40:15 +01:00
committed by Elisiário Couto
parent de3da84dff
commit ec8ef8346a
34 changed files with 226 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.claude/settings.local.json
View File

@@ -4,9 +4,11 @@
"Bash(mkdir:*)",
"Bash(uv sync:*)",
"Bash(uv run pytest:*)",
"Bash(git commit:*)"
"Bash(git commit:*)",
"Bash(ruff check:*)",
"Bash(git add:*)"
],
"deny": [],
"ask": []
}
}
}

14
.pre-commit-config.yaml
View File

@@ -1,13 +1,23 @@
repos:
- repo: https://github.com/charliermarsh/ruff-pre-commit
rev: "v0.9.1"
rev: "v0.12.11"
hooks:
- id: ruff
- id: ruff-format
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v5.0.0
rev: v6.0.0
hooks:
- id: trailing-whitespace
exclude: ".*\\.md$"
- id: end-of-file-fixer
- id: check-added-large-files
- repo: local
hooks:
- id: mypy
name: Static type check with mypy
entry: uv run mypy leggen leggend --check-untyped-defs
files: "^leggen(d)?/.*"
language: "system"
types: ["python"]
always_run: true
pass_filenames: false

2
CLAUDE.md
View File

@@ -66,4 +66,4 @@ All operations require a valid `config.toml` file with GoCardless API credential
- `[gocardless]` - API credentials and endpoint
- `[database]` - Storage backend selection
- `[notifications]` - Discord/Telegram webhook settings
- `[filters]` - Transaction matching patterns for notifications
- `[filters]` - Transaction matching patterns for notifications

2
PROJECT.md
View File

@@ -88,4 +88,4 @@ Transform leggen from CLI-only to web application with FastAPI backend (`leggend
- **APScheduler**: For internal job scheduling (replacing Ofelia)
- **SvelteKit**: For modern, reactive frontend
- **Existing Logic**: Reuse all business logic from current CLI commands
- **Configuration**: Centralize in `leggend` service, maintain TOML compatibility
- **Configuration**: Centralize in `leggend` service, maintain TOML compatibility

0
leggen/__init__.py Normal file
View File

26
leggen/api_client.py
View File

@@ -1,6 +1,6 @@
import os
import requests
from typing import Dict, Any, Optional, List
from typing import Dict, Any, Optional, List, Union
from urllib.parse import urljoin
from leggen.utils.text import error
@@ -9,9 +9,13 @@ from leggen.utils.text import error
class LeggendAPIClient:
"""Client for communicating with the leggend FastAPI service"""
base_url: str
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")
or "http://localhost:8000"
)
self.session = requests.Session()
self.session.headers.update(
@@ -36,7 +40,7 @@ class LeggendAPIClient:
try:
error_data = response.json()
error(f"Error details: {error_data.get('detail', 'Unknown error')}")
except:
except Exception:
error(f"Response: {response.text}")
raise
except Exception as e:
@@ -48,7 +52,7 @@ class LeggendAPIClient:
try:
response = self._make_request("GET", "/health")
return response.get("status") == "healthy"
except:
except Exception:
return False
# Bank endpoints
@@ -122,7 +126,7 @@ class LeggendAPIClient:
self, days: int = 30, account_id: Optional[str] = None
) -> Dict[str, Any]:
"""Get transaction statistics"""
params = {"days": days}
params: Dict[str, Union[int, str]] = {"days": days}
if account_id:
params["account_id"] = account_id
@@ -141,7 +145,7 @@ class LeggendAPIClient:
self, account_ids: Optional[List[str]] = None, force: bool = False
) -> Dict[str, Any]:
"""Trigger a sync"""
data = {"force": force}
data: Dict[str, Union[bool, List[str]]] = {"force": force}
if account_ids:
data["account_ids"] = account_ids
@@ -152,7 +156,7 @@ class LeggendAPIClient:
self, account_ids: Optional[List[str]] = None, force: bool = False
) -> Dict[str, Any]:
"""Run sync synchronously"""
data = {"force": force}
data: Dict[str, Union[bool, List[str]]] = {"force": force}
if account_ids:
data["account_ids"] = account_ids
@@ -172,7 +176,11 @@ class LeggendAPIClient:
cron: Optional[str] = None,
) -> Dict[str, Any]:
"""Update scheduler configuration"""
data = {"enabled": enabled, "hour": hour, "minute": minute}
data: Dict[str, Union[bool, int, str]] = {
"enabled": enabled,
"hour": hour,
"minute": minute,
}
if cron:
data["cron"] = cron

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

@@ -68,7 +68,7 @@ def add(ctx):
success("Bank connection request created successfully!")
warning(
f"Please open the following URL in your browser to complete the authorization:"
"Please open the following URL in your browser to complete the authorization:"
)
click.echo(f"\n{result['link']}\n")

10
leggen/commands/bank/delete.py
View File

@@ -1,7 +1,6 @@
import click
from leggen.main import cli
from leggen.utils.network import delete as http_delete
from leggen.utils.text import info, success
@@ -16,11 +15,12 @@ def delete(ctx, requisition_id: str):
Check `leggen status` to get the REQUISITION_ID
"""
import requests
info(f"Deleting Bank Requisition: {requisition_id}")
_ = http_delete(
ctx,
f"/requisitions/{requisition_id}",
)
api_url = ctx.obj.get("api_url", "http://localhost:8000")
res = requests.delete(f"{api_url}/requisitions/{requisition_id}")
res.raise_for_status()
success(f"Bank Requisition {requisition_id} deleted")

2
leggen/commands/sync.py
View File

@@ -27,7 +27,7 @@ def sync(ctx: click.Context, wait: bool, force: bool):
result = api_client.sync_now(force=force)
if result.get("success"):
success(f"Sync completed successfully!")
success("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)}")

13
leggen/main.py
View File

@@ -5,7 +5,6 @@ from pathlib import Path
import click
from leggen.utils.auth import get_token
from leggen.utils.config import load_config
from leggen.utils.text import error
@@ -111,14 +110,4 @@ def cli(ctx: click.Context, api_url: str):
return
# 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:
token = get_token(ctx)
ctx.obj["headers"] = {"Authorization": f"Bearer {token}"}
except Exception:
# If we can't get token, commands will rely on API service
pass
ctx.obj["api_url"] = api_url

62
leggen/utils/auth.py
View File

@@ -1,62 +0,0 @@
import json
from pathlib import Path
import click
import requests
from leggen.utils.text import warning
def create_token(ctx: click.Context) -> str:
"""
Create a new token
"""
res = requests.post(
f"{ctx.obj['gocardless']['url']}/token/new/",
json={
"secret_id": ctx.obj["gocardless"]["key"],
"secret_key": ctx.obj["gocardless"]["secret"],
},
)
res.raise_for_status()
auth = res.json()
save_auth(auth)
return auth["access"]
def get_token(ctx: click.Context) -> str:
"""
Get the token from the auth file or request a new one
"""
auth_file = Path.home() / ".config" / "leggen" / "auth.json"
if auth_file.exists():
with click.open_file(str(auth_file), "r") as f:
auth = json.load(f)
if not auth.get("access"):
return create_token(ctx)
res = requests.post(
f"{ctx.obj['gocardless']['url']}/token/refresh/",
json={"refresh": auth["refresh"]},
)
try:
res.raise_for_status()
auth.update(res.json())
save_auth(auth)
return auth["access"]
except requests.exceptions.HTTPError:
warning(
f"Token probably expired, requesting a new one.\nResponse: {res.status_code}\n{res.text}"
)
return create_token(ctx)
else:
return create_token(ctx)
def save_auth(d: dict):
auth_dir = Path.home() / ".config" / "leggen"
auth_dir.mkdir(parents=True, exist_ok=True)
auth_file = auth_dir / "auth.json"
with click.open_file(str(auth_file), "w") as f:
json.dump(d, f)

15
leggen/utils/database.py
View File

@@ -3,7 +3,6 @@ from datetime import datetime
import click
import leggen.database.sqlite as sqlite_engine
from leggen.utils.network import get
from leggen.utils.text import info, warning
@@ -32,15 +31,21 @@ def persist_transactions(ctx: click.Context, account: str, transactions: list) -
def save_transactions(ctx: click.Context, account: str) -> list:
import requests
api_url = ctx.obj.get("api_url", "http://localhost:8000")
info(f"[{account}] Getting account details")
account_info = get(ctx, f"/accounts/{account}")
res = requests.get(f"{api_url}/accounts/{account}")
res.raise_for_status()
account_info = res.json()
info(f"[{account}] Getting transactions")
transactions = []
account_transactions = get(ctx, f"/accounts/{account}/transactions/").get(
"transactions", []
)
res = requests.get(f"{api_url}/accounts/{account}/transactions/")
res.raise_for_status()
account_transactions = res.json().get("transactions", [])
for transaction in account_transactions.get("booked", []):
booked_date = transaction.get("bookingDateTime") or transaction.get(

64
leggen/utils/network.py
View File

@@ -1,64 +0,0 @@
import click
import requests
from leggen.utils.text import error
def get(ctx: click.Context, path: str, params: dict = {}):
"""
GET request to the GoCardless API
"""
url = f"{ctx.obj['gocardless']['url']}{path}"
res = requests.get(url, headers=ctx.obj["headers"], params=params)
try:
res.raise_for_status()
except Exception as e:
error(f"Error: {e}\n{res.text}")
ctx.abort()
return res.json()
def post(ctx: click.Context, path: str, data: dict = {}):
"""
POST request to the GoCardless API
"""
url = f"{ctx.obj['gocardless']['url']}{path}"
res = requests.post(url, headers=ctx.obj["headers"], json=data)
try:
res.raise_for_status()
except Exception as e:
error(f"Error: {e}\n{res.text}")
ctx.abort()
return res.json()
def put(ctx: click.Context, path: str, data: dict = {}):
"""
PUT request to the GoCardless API
"""
url = f"{ctx.obj['gocardless']['url']}{path}"
res = requests.put(url, headers=ctx.obj["headers"], json=data)
try:
res.raise_for_status()
except Exception as e:
error(f"Error: {e}\n{res.text}")
ctx.abort()
return res.json()
def delete(ctx: click.Context, path: str):
"""
DELETE request to the GoCardless API
"""
url = f"{ctx.obj['gocardless']['url']}{path}"
res = requests.delete(url, headers=ctx.obj["headers"])
try:
res.raise_for_status()
except Exception as e:
error(f"Error: {e}\n{res.text}")
ctx.abort()
return res.json()

0
leggend/__init__.py Normal file
View File

1
leggend/api/models/common.py
View File

@@ -1,4 +1,3 @@
from datetime import datetime
from typing import Any, Dict, Optional
from pydantic import BaseModel

2
leggend/api/models/notifications.py
View File

@@ -1,4 +1,4 @@
from typing import Dict, Any, Optional, List
from typing import Dict, Optional, List
from pydantic import BaseModel

2
leggend/api/models/sync.py
View File

@@ -1,5 +1,5 @@
from datetime import datetime
from typing import Optional, Dict, Any
from typing import Optional
from pydantic import BaseModel

32
leggend/api/routes/accounts.py
View File

@@ -1,4 +1,4 @@
from typing import List, Optional
from typing import Optional, List, Union
from fastapi import APIRouter, HTTPException, Query
from loguru import logger
@@ -74,7 +74,9 @@ async def get_all_accounts() -> APIResponse:
except Exception as e:
logger.error(f"Failed to get accounts: {e}")
raise HTTPException(status_code=500, detail=f"Failed to get accounts: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Failed to get accounts: {str(e)}"
) from e
@router.get("/accounts/{account_id}", response_model=APIResponse)
@@ -117,7 +119,9 @@ async def get_account_details(account_id: str) -> APIResponse:
except Exception as e:
logger.error(f"Failed to get account details for {account_id}: {e}")
raise HTTPException(status_code=404, detail=f"Account not found: {str(e)}")
raise HTTPException(
status_code=404, detail=f"Account not found: {str(e)}"
) from e
@router.get("/accounts/{account_id}/balances", response_model=APIResponse)
@@ -146,7 +150,9 @@ async def get_account_balances(account_id: str) -> APIResponse:
except Exception as e:
logger.error(f"Failed to get balances for account {account_id}: {e}")
raise HTTPException(status_code=404, detail=f"Failed to get balances: {str(e)}")
raise HTTPException(
status_code=404, detail=f"Failed to get balances: {str(e)}"
) from e
@router.get("/accounts/{account_id}/transactions", response_model=APIResponse)
@@ -172,11 +178,17 @@ async def get_account_transactions(
# Apply pagination
total_transactions = len(processed_transactions)
paginated_transactions = processed_transactions[offset : offset + limit]
actual_offset = offset or 0
actual_limit = limit or 100
paginated_transactions = processed_transactions[
actual_offset : actual_offset + actual_limit
]
data: Union[List[TransactionSummary], List[Transaction]]
if summary_only:
# Return simplified transaction summaries
summaries = [
data = [
TransactionSummary(
internal_transaction_id=txn["internalTransactionId"],
date=txn["transactionDate"],
@@ -188,10 +200,9 @@ async def get_account_transactions(
)
for txn in paginated_transactions
]
data = summaries
else:
# Return full transaction details
transactions = [
data = [
Transaction(
internal_transaction_id=txn["internalTransactionId"],
institution_id=txn["institutionId"],
@@ -206,16 +217,15 @@ async def get_account_transactions(
)
for txn in paginated_transactions
]
data = transactions
return APIResponse(
success=True,
data=data,
message=f"Retrieved {len(data)} transactions (showing {offset + 1}-{offset + len(data)} of {total_transactions})",
message=f"Retrieved {len(data)} transactions (showing {actual_offset + 1}-{actual_offset + len(data)} of {total_transactions})",
)
except Exception as e:
logger.error(f"Failed to get transactions for account {account_id}: {e}")
raise HTTPException(
status_code=404, detail=f"Failed to get transactions: {str(e)}"
)
) from e

16
leggend/api/routes/banks.py
View File

@@ -1,8 +1,7 @@
from typing import List, Optional
from fastapi import APIRouter, HTTPException, Query
from loguru import logger
from leggend.api.models.common import APIResponse, ErrorResponse
from leggend.api.models.common import APIResponse
from leggend.api.models.banks import (
BankInstitution,
BankConnectionRequest,
@@ -46,15 +45,16 @@ async def get_bank_institutions(
logger.error(f"Failed to get institutions for {country}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get institutions: {str(e)}"
)
) from e
@router.post("/banks/connect", response_model=APIResponse)
async def connect_to_bank(request: BankConnectionRequest) -> APIResponse:
"""Create a connection to a bank (requisition)"""
try:
redirect_url = request.redirect_url or "http://localhost:8000/"
requisition_data = await gocardless_service.create_requisition(
request.institution_id, request.redirect_url
request.institution_id, redirect_url
)
requisition = BankRequisition(
@@ -69,14 +69,14 @@ async def connect_to_bank(request: BankConnectionRequest) -> APIResponse:
return APIResponse(
success=True,
data=requisition,
message=f"Bank connection created. Please visit the link to authorize.",
message="Bank connection created. Please visit the link to authorize.",
)
except Exception as e:
logger.error(f"Failed to connect to bank {request.institution_id}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to connect to bank: {str(e)}"
)
) from e
@router.get("/banks/status", response_model=APIResponse)
@@ -114,7 +114,7 @@ async def get_bank_connections_status() -> APIResponse:
logger.error(f"Failed to get bank connection status: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get bank status: {str(e)}"
)
) from e
@router.delete("/banks/connections/{requisition_id}", response_model=APIResponse)
@@ -132,7 +132,7 @@ async def delete_bank_connection(requisition_id: str) -> APIResponse:
logger.error(f"Failed to delete bank connection {requisition_id}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to delete connection: {str(e)}"
)
) from e
@router.get("/banks/countries", response_model=APIResponse)

14
leggend/api/routes/notifications.py
View File

@@ -1,4 +1,4 @@
from typing import Optional
from typing import Dict, Any
from fastapi import APIRouter, HTTPException
from loguru import logger
@@ -60,7 +60,7 @@ async def get_notification_settings() -> APIResponse:
logger.error(f"Failed to get notification settings: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get notification settings: {str(e)}"
)
) from e
@router.put("/notifications/settings", response_model=APIResponse)
@@ -84,7 +84,7 @@ async def update_notification_settings(settings: NotificationSettings) -> APIRes
}
# Update filters config
filters_config = {}
filters_config: Dict[str, Any] = {}
if settings.filters.case_insensitive:
filters_config["case-insensitive"] = settings.filters.case_insensitive
if settings.filters.case_sensitive:
@@ -110,7 +110,7 @@ async def update_notification_settings(settings: NotificationSettings) -> APIRes
logger.error(f"Failed to update notification settings: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to update notification settings: {str(e)}"
)
) from e
@router.post("/notifications/test", response_model=APIResponse)
@@ -137,7 +137,7 @@ async def test_notification(test_request: NotificationTest) -> APIResponse:
logger.error(f"Failed to send test notification: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to send test notification: {str(e)}"
)
) from e
@router.get("/notifications/services", response_model=APIResponse)
@@ -179,7 +179,7 @@ async def get_notification_services() -> APIResponse:
logger.error(f"Failed to get notification services: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get notification services: {str(e)}"
)
) from e
@router.delete("/notifications/settings/{service}", response_model=APIResponse)
@@ -206,4 +206,4 @@ async def delete_notification_service(service: str) -> APIResponse:
logger.error(f"Failed to delete notification service {service}: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to delete notification service: {str(e)}"
)
) from e

22
leggend/api/routes/sync.py
View File

@@ -3,7 +3,7 @@ from fastapi import APIRouter, HTTPException, BackgroundTasks
from loguru import logger
from leggend.api.models.common import APIResponse
from leggend.api.models.sync import SyncRequest, SyncStatus, SyncResult, SchedulerConfig
from leggend.api.models.sync import SyncRequest, SchedulerConfig
from leggend.services.sync_service import SyncService
from leggend.background.scheduler import scheduler
from leggend.config import config
@@ -31,7 +31,7 @@ async def get_sync_status() -> APIResponse:
logger.error(f"Failed to get sync status: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get sync status: {str(e)}"
)
) from e
@router.post("/sync", response_model=APIResponse)
@@ -78,7 +78,9 @@ async def trigger_sync(
except Exception as e:
logger.error(f"Failed to trigger sync: {e}")
raise HTTPException(status_code=500, detail=f"Failed to trigger sync: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Failed to trigger sync: {str(e)}"
) from e
@router.post("/sync/now", response_model=APIResponse)
@@ -104,7 +106,9 @@ async def sync_now(sync_request: Optional[SyncRequest] = None) -> APIResponse:
except Exception as e:
logger.error(f"Failed to run sync: {e}")
raise HTTPException(status_code=500, detail=f"Failed to run sync: {str(e)}")
raise HTTPException(
status_code=500, detail=f"Failed to run sync: {str(e)}"
) from e
@router.get("/sync/scheduler", response_model=APIResponse)
@@ -134,7 +138,7 @@ async def get_scheduler_config() -> APIResponse:
logger.error(f"Failed to get scheduler config: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get scheduler config: {str(e)}"
)
) from e
@router.put("/sync/scheduler", response_model=APIResponse)
@@ -152,7 +156,7 @@ async def update_scheduler_config(scheduler_config: SchedulerConfig) -> APIRespo
except Exception as e:
raise HTTPException(
status_code=400, detail=f"Invalid cron expression: {str(e)}"
)
) from e
# Update configuration
schedule_data = scheduler_config.dict(exclude_none=True)
@@ -171,7 +175,7 @@ async def update_scheduler_config(scheduler_config: SchedulerConfig) -> APIRespo
logger.error(f"Failed to update scheduler config: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to update scheduler config: {str(e)}"
)
) from e
@router.post("/sync/scheduler/start", response_model=APIResponse)
@@ -188,7 +192,7 @@ async def start_scheduler() -> APIResponse:
logger.error(f"Failed to start scheduler: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to start scheduler: {str(e)}"
)
) from e
@router.post("/sync/scheduler/stop", response_model=APIResponse)
@@ -205,4 +209,4 @@ async def stop_scheduler() -> APIResponse:
logger.error(f"Failed to stop scheduler: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to stop scheduler: {str(e)}"
)
) from e

16
leggend/api/routes/transactions.py
View File

@@ -1,4 +1,4 @@
from typing import List, Optional
from typing import Optional, List, Union
from datetime import datetime, timedelta
from fastapi import APIRouter, HTTPException, Query
from loguru import logger
@@ -120,7 +120,13 @@ async def get_all_transactions(
# Apply pagination
total_transactions = len(filtered_transactions)
paginated_transactions = filtered_transactions[offset : offset + limit]
actual_offset = offset or 0
actual_limit = limit or 100
paginated_transactions = filtered_transactions[
actual_offset : actual_offset + actual_limit
]
data: Union[List[TransactionSummary], List[Transaction]]
if summary_only:
# Return simplified transaction summaries
@@ -157,14 +163,14 @@ async def get_all_transactions(
return APIResponse(
success=True,
data=data,
message=f"Retrieved {len(data)} transactions (showing {offset + 1}-{offset + len(data)} of {total_transactions})",
message=f"Retrieved {len(data)} transactions (showing {actual_offset + 1}-{actual_offset + len(data)} of {total_transactions})",
)
except Exception as e:
logger.error(f"Failed to get transactions: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get transactions: {str(e)}"
)
) from e
@router.get("/transactions/stats", response_model=APIResponse)
@@ -270,4 +276,4 @@ async def get_transaction_stats(
logger.error(f"Failed to get transaction stats: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to get transaction stats: {str(e)}"
)
) from e

19
leggend/config.py
View File

@@ -17,7 +17,7 @@ class Config:
cls._instance = super().__new__(cls)
return cls._instance
def load_config(self, config_path: str = None) -> Dict[str, Any]:
def load_config(self, config_path: Optional[str] = None) -> Dict[str, Any]:
if self._config is not None:
return self._config
@@ -43,7 +43,9 @@ class Config:
return self._config
def save_config(
self, config_data: Dict[str, Any] = None, config_path: str = None
self,
config_data: Optional[Dict[str, Any]] = None,
config_path: Optional[str] = None,
) -> None:
"""Save configuration to TOML file"""
if config_data is None:
@@ -55,6 +57,11 @@ class Config:
str(Path.home() / ".config" / "leggen" / "config.toml"),
)
if config_path is None:
raise ValueError("No config path specified")
if config_data is None:
raise ValueError("No config data to save")
# Ensure directory exists
Path(config_path).parent.mkdir(parents=True, exist_ok=True)
@@ -75,6 +82,9 @@ class Config:
if self._config is None:
self.load_config()
if self._config is None:
raise RuntimeError("Failed to load config")
if section not in self._config:
self._config[section] = {}
@@ -86,6 +96,9 @@ class Config:
if self._config is None:
self.load_config()
if self._config is None:
raise RuntimeError("Failed to load config")
self._config[section] = data
self.save_config()
@@ -93,6 +106,8 @@ class Config:
def config(self) -> Dict[str, Any]:
if self._config is None:
self.load_config()
if self._config is None:
raise RuntimeError("Failed to load config")
return self._config
@property

1
leggend/main.py
View File

@@ -1,4 +1,3 @@
import asyncio
from contextlib import asynccontextmanager
from importlib import metadata

7
leggend/services/database_service.py
View File

@@ -1,5 +1,5 @@
from datetime import datetime
from typing import List, Dict, Any, Optional
from typing import List, Dict, Any
from loguru import logger
@@ -75,7 +75,10 @@ class DatabaseService:
datetime.fromisoformat(booked_date), datetime.fromisoformat(value_date)
)
else:
min_date = datetime.fromisoformat(booked_date or value_date)
date_str = booked_date or value_date
if not date_str:
raise ValueError("No valid date found in transaction")
min_date = datetime.fromisoformat(date_str)
# Extract amount and currency
transaction_amount = transaction.get("transactionAmount", {})

3
leggend/services/gocardless_service.py
View File

@@ -1,8 +1,7 @@
import asyncio
import json
import httpx
from pathlib import Path
from typing import Dict, Any, List, Optional
from typing import Dict, Any, List
from loguru import logger

2
leggend/services/notification_service.py
View File

@@ -69,7 +69,7 @@ class NotificationService:
description = transaction.get("description", "").lower()
# Check case-insensitive filters
for filter_name, filter_value in filters_case_insensitive.items():
for _filter_name, filter_value in filters_case_insensitive.items():
if filter_value.lower() in description:
matching.append(
{

4
leggend/services/sync_service.py
View File

@@ -1,10 +1,8 @@
import asyncio
from datetime import datetime
from typing import List, Dict, Any
from typing import List
from loguru import logger
from leggend.config import config
from leggend.api.models.sync import SyncResult, SyncStatus
from leggend.services.gocardless_service import GoCardlessService
from leggend.services.database_service import DatabaseService

22
pyproject.toml
View File

@@ -49,9 +49,12 @@ dev = [
"pre-commit>=3.6.0",
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"pytest-mock>=3.12.0",
"pytest-mock>=3.12.0",
"respx>=0.21.0",
"requests-mock>=1.12.0",
"mypy>=1.17.1",
"types-tabulate>=0.9.0.20241207",
"types-requests>=2.32.4.20250809",
]
[tool.hatch.build.targets.sdist]
@@ -71,20 +74,19 @@ lint.extend-select = ["B", "C4", "PIE", "T20", "SIM", "TCH"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = "test_*.py"
python_classes = "Test*"
python_classes = "Test*"
python_functions = "test_*"
addopts = [
"-v",
"--tb=short",
"--strict-markers",
"--disable-warnings"
]
addopts = ["-v", "--tb=short", "--strict-markers", "--disable-warnings"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
markers = [
"unit: Unit tests",
"integration: Integration tests",
"slow: Slow running tests",
"slow: Slow running tests",
"api: API endpoint tests",
"cli: CLI command tests"
"cli: CLI command tests",
]
[[tool.mypy.overrides]]
module = ["apscheduler.*"]
ignore_missing_imports = true

2
tests/unit/test_api_banks.py
View File

@@ -5,8 +5,6 @@ import respx
import httpx
from unittest.mock import patch
from leggend.services.gocardless_service import GoCardlessService
@pytest.mark.api
class TestBanksAPI:

5
tests/unit/test_api_client.py
View File

@@ -1,6 +1,7 @@
"""Tests for CLI API client."""
import pytest
import requests
import requests_mock
from unittest.mock import patch
@@ -85,7 +86,7 @@ class TestLeggendAPIClient:
"""Test handling of connection errors."""
client = LeggendAPIClient("http://localhost:9999") # Non-existent service
with pytest.raises(Exception):
with pytest.raises((requests.ConnectionError, requests.RequestException)):
client.get_accounts()
def test_http_error_handling(self):
@@ -99,7 +100,7 @@ class TestLeggendAPIClient:
json={"detail": "Internal server error"},
)
with pytest.raises(Exception):
with pytest.raises((requests.HTTPError, requests.RequestException)):
client.get_accounts()
def test_custom_api_url(self):

10
tests/unit/test_config.py
View File

@@ -1,8 +1,6 @@
"""Tests for configuration management."""
import pytest
import tempfile
from pathlib import Path
from unittest.mock import patch
from leggend.config import Config
@@ -164,9 +162,11 @@ class TestConfig:
config = Config()
config._config = None
with patch("builtins.open", side_effect=FileNotFoundError):
with pytest.raises(FileNotFoundError):
config.load_config()
with (
patch("builtins.open", side_effect=FileNotFoundError),
pytest.raises(FileNotFoundError),
):
config.load_config()
def test_notifications_config(self):
"""Test notifications configuration access."""

5
tests/unit/test_scheduler.py
View File

@@ -1,13 +1,10 @@
"""Tests for background scheduler."""
import pytest
import asyncio
from unittest.mock import Mock, patch, AsyncMock, MagicMock
from unittest.mock import patch, AsyncMock, MagicMock
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from leggend.background.scheduler import BackgroundScheduler
from leggend.services.sync_service import SyncService
@pytest.mark.unit

65
uv.lock generated
View File

@@ -240,6 +240,7 @@ dependencies = [
[package.dev-dependencies]
dev = [
{ name = "mypy" },
{ name = "pre-commit" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
@@ -247,6 +248,8 @@ dev = [
{ name = "requests-mock" },
{ name = "respx" },
{ name = "ruff" },
{ name = "types-requests" },
{ name = "types-tabulate" },
]
[package.metadata]
@@ -265,6 +268,7 @@ requires-dist = [
[package.metadata.requires-dev]
dev = [
{ name = "mypy", specifier = ">=1.17.1" },
{ name = "pre-commit", specifier = ">=3.6.0" },
{ name = "pytest", specifier = ">=8.0.0" },
{ name = "pytest-asyncio", specifier = ">=0.23.0" },
@@ -272,6 +276,8 @@ dev = [
{ name = "requests-mock", specifier = ">=1.12.0" },
{ name = "respx", specifier = ">=0.21.0" },
{ name = "ruff", specifier = ">=0.6.1" },
{ name = "types-requests", specifier = ">=2.32.4.20250809" },
{ name = "types-tabulate", specifier = ">=0.9.0.20241207" },
]
[[package]]
@@ -287,6 +293,35 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0c/29/0348de65b8cc732daa3e33e67806420b2ae89bdce2b04af740289c5c6c8c/loguru-0.7.3-py3-none-any.whl", hash = "sha256:31a33c10c8e1e10422bfd431aeb5d351c7cf7fa671e3c4df004162264b28220c", size = 61595, upload-time = "2024-12-06T11:20:54.538Z" },
]
[[package]]
name = "mypy"
version = "1.17.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "mypy-extensions" },
{ name = "pathspec" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/8e/22/ea637422dedf0bf36f3ef238eab4e455e2a0dcc3082b5cc067615347ab8e/mypy-1.17.1.tar.gz", hash = "sha256:25e01ec741ab5bb3eec8ba9cdb0f769230368a22c959c4937360efb89b7e9f01", size = 3352570, upload-time = "2025-07-31T07:54:19.204Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/17/a2/7034d0d61af8098ec47902108553122baa0f438df8a713be860f7407c9e6/mypy-1.17.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:69e83ea6553a3ba79c08c6e15dbd9bfa912ec1e493bf75489ef93beb65209aeb", size = 11086295, upload-time = "2025-07-31T07:53:28.124Z" },
{ url = "https://files.pythonhosted.org/packages/14/1f/19e7e44b594d4b12f6ba8064dbe136505cec813549ca3e5191e40b1d3cc2/mypy-1.17.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:1b16708a66d38abb1e6b5702f5c2c87e133289da36f6a1d15f6a5221085c6403", size = 10112355, upload-time = "2025-07-31T07:53:21.121Z" },
{ url = "https://files.pythonhosted.org/packages/5b/69/baa33927e29e6b4c55d798a9d44db5d394072eef2bdc18c3e2048c9ed1e9/mypy-1.17.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:89e972c0035e9e05823907ad5398c5a73b9f47a002b22359b177d40bdaee7056", size = 11875285, upload-time = "2025-07-31T07:53:55.293Z" },
{ url = "https://files.pythonhosted.org/packages/90/13/f3a89c76b0a41e19490b01e7069713a30949d9a6c147289ee1521bcea245/mypy-1.17.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:03b6d0ed2b188e35ee6d5c36b5580cffd6da23319991c49ab5556c023ccf1341", size = 12737895, upload-time = "2025-07-31T07:53:43.623Z" },
{ url = "https://files.pythonhosted.org/packages/23/a1/c4ee79ac484241301564072e6476c5a5be2590bc2e7bfd28220033d2ef8f/mypy-1.17.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:c837b896b37cd103570d776bda106eabb8737aa6dd4f248451aecf53030cdbeb", size = 12931025, upload-time = "2025-07-31T07:54:17.125Z" },
{ url = "https://files.pythonhosted.org/packages/89/b8/7409477be7919a0608900e6320b155c72caab4fef46427c5cc75f85edadd/mypy-1.17.1-cp312-cp312-win_amd64.whl", hash = "sha256:665afab0963a4b39dff7c1fa563cc8b11ecff7910206db4b2e64dd1ba25aed19", size = 9584664, upload-time = "2025-07-31T07:54:12.842Z" },
{ url = "https://files.pythonhosted.org/packages/1d/f3/8fcd2af0f5b806f6cf463efaffd3c9548a28f84220493ecd38d127b6b66d/mypy-1.17.1-py3-none-any.whl", hash = "sha256:a9f52c0351c21fe24c21d8c0eb1f62967b262d6729393397b6f443c3b773c3b9", size = 2283411, upload-time = "2025-07-31T07:53:24.664Z" },
]
[[package]]
name = "mypy-extensions"
version = "1.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/a2/6e/371856a3fb9d31ca8dac321cda606860fa4548858c0cc45d9d1d4ca2628b/mypy_extensions-1.1.0.tar.gz", hash = "sha256:52e68efc3284861e772bbcd66823fde5ae21fd2fdb51c62a211403730b916558", size = 6343, upload-time = "2025-04-22T14:54:24.164Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/79/7b/2c79738432f5c924bef5071f933bcc9efd0473bac3b4aa584a6f7c1c8df8/mypy_extensions-1.1.0-py3-none-any.whl", hash = "sha256:1be4cccdb0f2482337c4743e60421de3a356cd97508abadd57d47403e94f5505", size = 4963, upload-time = "2025-04-22T14:54:22.983Z" },
]
[[package]]
name = "nodeenv"
version = "1.9.1"
@@ -305,6 +340,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/20/12/38679034af332785aac8774540895e234f4d07f7545804097de4b666afd8/packaging-25.0-py3-none-any.whl", hash = "sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484", size = 66469, upload-time = "2025-04-19T11:48:57.875Z" },
]
[[package]]
name = "pathspec"
version = "0.12.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ca/bc/f35b8446f4531a7cb215605d100cd88b7ac6f44ab3fc94870c120ab3adbf/pathspec-0.12.1.tar.gz", hash = "sha256:a482d51503a1ab33b1c67a6c3813a26953dbdc71c31dacaef9a838c4e29f5712", size = 51043, upload-time = "2023-12-10T22:30:45Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cc/20/ff623b09d963f88bfde16306a54e12ee5ea43e9b597108672ff3a408aad6/pathspec-0.12.1-py3-none-any.whl", hash = "sha256:a0d503e138a4c123b27490a4f7beda6a01c6f288df0e4a8b79c7eb0dc7b4cc08", size = 31191, upload-time = "2023-12-10T22:30:43.14Z" },
]
[[package]]
name = "platformdirs"
version = "4.3.6"
@@ -558,6 +602,27 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/18/c86eb8e0202e32dd3df50d43d7ff9854f8e0603945ff398974c1d91ac1ef/tomli_w-1.2.0-py3-none-any.whl", hash = "sha256:188306098d013b691fcadc011abd66727d3c414c571bb01b1a174ba8c983cf90", size = 6675, upload-time = "2025-01-15T12:07:22.074Z" },
]
[[package]]
name = "types-requests"
version = "2.32.4.20250809"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ed/b0/9355adb86ec84d057fea765e4c49cce592aaf3d5117ce5609a95a7fc3dac/types_requests-2.32.4.20250809.tar.gz", hash = "sha256:d8060de1c8ee599311f56ff58010fb4902f462a1470802cf9f6ed27bc46c4df3", size = 23027, upload-time = "2025-08-09T03:17:10.664Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2b/6f/ec0012be842b1d888d46884ac5558fd62aeae1f0ec4f7a581433d890d4b5/types_requests-2.32.4.20250809-py3-none-any.whl", hash = "sha256:f73d1832fb519ece02c85b1f09d5f0dd3108938e7d47e7f94bbfa18a6782b163", size = 20644, upload-time = "2025-08-09T03:17:09.716Z" },
]
[[package]]
name = "types-tabulate"
version = "0.9.0.20241207"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/3f/43/16030404a327e4ff8c692f2273854019ed36718667b2993609dc37d14dd4/types_tabulate-0.9.0.20241207.tar.gz", hash = "sha256:ac1ac174750c0a385dfd248edc6279fa328aaf4ea317915ab879a2ec47833230", size = 8195, upload-time = "2024-12-07T02:54:42.554Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5e/86/a9ebfd509cbe74471106dffed320e208c72537f9aeb0a55eaa6b1b5e4d17/types_tabulate-0.9.0.20241207-py3-none-any.whl", hash = "sha256:b8dad1343c2a8ba5861c5441370c3e35908edd234ff036d4298708a1d4cf8a85", size = 8307, upload-time = "2024-12-07T02:54:41.031Z" },
]
[[package]]
name = "typing-extensions"
version = "4.15.0"