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

feat(api): Add bank logo support and fix banks endpoint type errors.

Browse Source 
Backend changes:
- Add logo field to AccountDetails model
- Update accounts API endpoints to include logo data
- Add database migration for logo column in accounts table
- Implement institution details fetching from GoCardless API
- Enrich account data with institution logos during sync
- Fix type errors in banks endpoint with proper response parsing

Frontend changes:
- Add failedImages state to track logo loading failures
- Implement conditional rendering to show bank logos when available
- Add proper error handling with fallback to Building2 icon
- Fix image sizing to w-6 h-6 sm:w-8 sm:h-8 for proper display
- Update Account interface to include optional logo field
- Remove unused useState import from System component

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Elisiário Couto
2025-09-24 19:57:03 +01:00
parent a8f704129b
commit b9ca74e7e6
12 changed files with 154 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
leggen/api/models/accounts.py
View File

@@ -26,6 +26,7 @@ class AccountDetails(BaseModel):
name: Optional[str] = None
display_name: Optional[str] = None
currency: Optional[str] = None
logo: Optional[str] = None
created: datetime
last_accessed: Optional[datetime] = None
balances: List[AccountBalance] = []

2
leggen/api/routes/accounts.py
View File

@@ -55,6 +55,7 @@ async def get_all_accounts() -> APIResponse:
name=db_account.get("name"),
display_name=db_account.get("display_name"),
currency=db_account.get("currency"),
logo=db_account.get("logo"),
created=db_account["created"],
last_accessed=db_account.get("last_accessed"),
balances=balances,
@@ -115,6 +116,7 @@ async def get_account_details(account_id: str) -> APIResponse:
name=db_account.get("name"),
display_name=db_account.get("display_name"),
currency=db_account.get("currency"),
logo=db_account.get("logo"),
created=db_account["created"],
last_accessed=db_account.get("last_accessed"),
balances=balances,

7
leggen/api/routes/banks.py
View File

@@ -21,18 +21,19 @@ async def get_bank_institutions(
) -> APIResponse:
"""Get available bank institutions for a country"""
try:
institutions_data = await gocardless_service.get_institutions(country)
institutions_response = await gocardless_service.get_institutions(country)
institutions_data = institutions_response.get("results", [])
institutions = [
BankInstitution(
id=inst["id"],
name=inst["name"],
bic=inst.get("bic"),
transaction_total_days=inst["transaction_total_days"],
transaction_total_days=int(inst["transaction_total_days"]),
countries=inst["countries"],
logo=inst.get("logo"),
)
for inst in institutions_data.get("results", [])
for inst in institutions_data
]
return APIResponse(

83
leggen/services/database_service.py
View File

@@ -217,6 +217,7 @@ class DatabaseService:
await self._migrate_to_composite_key_if_needed()
await self._migrate_add_display_name_if_needed()
await self._migrate_add_sync_operations_if_needed()
await self._migrate_add_logo_if_needed()
async def _migrate_balance_timestamps_if_needed(self):
"""Check and migrate balance timestamps if needed"""
@@ -1133,7 +1134,8 @@ class DatabaseService:
created DATETIME,
last_accessed DATETIME,
last_updated DATETIME,
display_name TEXT
display_name TEXT,
logo TEXT
)"""
)
@@ -1170,8 +1172,9 @@ class DatabaseService:
created,
last_accessed,
last_updated,
display_name
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
display_name,
logo
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
account_data["id"],
account_data["institution_id"],
@@ -1183,6 +1186,7 @@ class DatabaseService:
account_data.get("last_accessed"),
account_data.get("last_updated", account_data["created"]),
display_name,
account_data.get("logo"),
),
)
conn.commit()
@@ -1516,6 +1520,79 @@ class DatabaseService:
logger.error(f"Sync operations table migration failed: {e}")
raise
async def _migrate_add_logo_if_needed(self):
"""Check and add logo column to accounts table if needed"""
try:
if await self._check_logo_migration_needed():
logger.info("Logo column migration needed, starting...")
await self._migrate_add_logo()
logger.info("Logo column migration completed")
else:
logger.info("Logo column already exists")
except Exception as e:
logger.error(f"Logo column migration failed: {e}")
raise
async def _check_logo_migration_needed(self) -> bool:
"""Check if logo column needs to be added to accounts table"""
db_path = path_manager.get_database_path()
if not db_path.exists():
return False
try:
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
# Check if accounts table exists
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='accounts'"
)
if not cursor.fetchone():
conn.close()
return False
# Check if logo column exists
cursor.execute("PRAGMA table_info(accounts)")
columns = cursor.fetchall()
# Check if logo column exists
has_logo = any(col[1] == "logo" for col in columns)
conn.close()
return not has_logo
except Exception as e:
logger.error(f"Failed to check logo migration status: {e}")
return False
async def _migrate_add_logo(self):
"""Add logo column to accounts table"""
db_path = path_manager.get_database_path()
if not db_path.exists():
logger.warning("Database file not found, skipping migration")
return
try:
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
logger.info("Adding logo column to accounts table...")
# Add the logo column
cursor.execute("""
ALTER TABLE accounts
ADD COLUMN logo TEXT
""")
conn.commit()
conn.close()
logger.info("Logo column migration completed successfully")
except Exception as e:
logger.error(f"Logo column migration failed: {e}")
raise
async def persist_sync_operation(self, sync_operation: Dict[str, Any]) -> int:
"""Persist sync operation to database and return the ID"""
if not self.sqlite_enabled:

17
leggen/services/gocardless_service.py
View File

@@ -11,14 +11,13 @@ from leggen.utils.paths import path_manager
def _log_rate_limits(response):
"""Log GoCardless API rate limit headers"""
limit = response.headers.get("X-RateLimit-Limit")
remaining = response.headers.get("X-RateLimit-Remaining")
reset = response.headers.get("X-RateLimit-Reset")
account_success_reset = response.headers.get("X-RateLimit-Account-Success-Reset")
limit = response.headers.get("http_x_ratelimit_limit")
remaining = response.headers.get("http_x_ratelimit_remaining")
reset = response.headers.get("http_x_ratelimit_reset")
if limit or remaining or reset or account_success_reset:
if limit or remaining or reset:
logger.info(
f"GoCardless rate limits - Limit: {limit}, Remaining: {remaining}, Reset: {reset}s, Account Success Reset: {account_success_reset}"
f"GoCardless rate limits - Limit: {limit}, Remaining: {remaining}, Reset: {reset}s"
)
@@ -162,3 +161,9 @@ class GoCardlessService:
return await self._make_authenticated_request(
"GET", f"{self.base_url}/accounts/{account_id}/transactions/"
)
async def get_institution_details(self, institution_id: str) -> Dict[str, Any]:
"""Get institution details by ID"""
return await self._make_authenticated_request(
"GET", f"{self.base_url}/institutions/{institution_id}/"
)

23
leggen/services/sync_service.py
View File

@@ -15,6 +15,7 @@ class SyncService:
self.database = DatabaseService()
self.notifications = NotificationService()
self._sync_status = SyncStatus(is_running=False)
self._institution_logos = {} # Cache for institution logos
async def get_sync_status(self) -> SyncStatus:
"""Get current sync status"""
@@ -77,7 +78,7 @@ class SyncService:
# Get balances to extract currency information
balances = await self.gocardless.get_account_balances(account_id)
# Enrich account details with currency and persist
# Enrich account details with currency and institution logo
if account_details and balances:
enriched_account_details = account_details.copy()
@@ -90,6 +91,26 @@ class SyncService:
if currency:
enriched_account_details["currency"] = currency
# Get institution details to fetch logo
institution_id = enriched_account_details.get("institution_id")
if institution_id:
try:
institution_details = (
await self.gocardless.get_institution_details(
institution_id
)
)
enriched_account_details["logo"] = (
institution_details.get("logo", "")
)
logger.info(
f"Fetched logo for institution {institution_id}: {enriched_account_details.get('logo', 'No logo')}"
)
except Exception as e:
logger.warning(
f"Failed to fetch institution details for {institution_id}: {e}"
)
# Persist enriched account details to database
await self.database.persist_account_details(
enriched_account_details