jessikitty/lego-instructions-manager
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add owner-only logging and Brickset API call tracking

Browse Source 
LIM configured no logging at all, so the instruction fetcher had no trail to follow. Adds a rotating file handler on the root logger plus two owner-only admin pages.

- Admin > Logs: filter by level, source and text; tracebacks kept with their error; auto-refresh, raw download, rotated-file browsing and rotate-now

- Admin > Logs > Brickset API calls: every outbound call recorded with method, params, caller, duration, result count and error, plus daily quota accounting (only getSets counts) and a comparison against getKeyUsageStats

- New owner concept (services/ownership.py): a single account, distinct from is_admin, since these pages expose other admins activity. Defaults to the first admin; flask owner show/set to change

- Services now log through module loggers; retention setting and pruning for call records

- New CLI: flask owner show/set, flask logs path/prune

- New table brickset_api_calls, created automatically at boot; LOG_* env vars, /data/logs in Docker
This commit is contained in:
jessikitty committed 2026-09-27 10:42:49 +10:00
1 parent 89f65e2d74
commit f7c20de5ac
27 files changed
+1914 -35
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/services/brickset_api.py
+108 -19
View File
@@ -17,10 +17,15 @@ Created by: Jess Rogerson (yelling commands at Claude.AI)
"""
import json
import logging
import time
from typing import Any, Dict, List, Optional
import requests
from flask import current_app
from flask import current_app, has_request_context, request
from flask_login import current_user
logger = logging.getLogger(__name__)
class BricksetAPI:
@@ -28,11 +33,49 @@ class BricksetAPI:
BASE_URL = 'https://brickset.com/api/v3.asmx'
def __init__(self):
def __init__(self, context: Optional[str] = None):
self.api_key = current_app.config.get('BRICKSET_API_KEY')
self.username = current_app.config.get('BRICKSET_USERNAME')
self.password = current_app.config.get('BRICKSET_PASSWORD')
self._user_hash = None
# Recorded against every call so the log shows which part of LIM
# was responsible. Falls back to the endpoint name in a request.
self.context = context or self._default_context()
@staticmethod
def _default_context() -> str:
if has_request_context() and request.endpoint:
return request.endpoint.split('.')[-1][:40]
return 'background'
def _log_call(self, method, params=None, ok=True, http_status=None,
result_count=None, started=None, message=None):
"""Record one call. Never raises."""
duration = None
if started is not None:
duration = int((time.monotonic() - started) * 1000)
if ok:
logger.info('Brickset %s (%s) -> %s result(s) in %sms',
method, params or '-',
'?' if result_count is None else result_count,
duration)
else:
logger.warning('Brickset %s (%s) failed: %s',
method, params or '-', message)
try:
from app.models.api_call import BricksetApiCall
user_id = None
if has_request_context() and getattr(current_user,
'is_authenticated', False):
user_id = current_user.id
BricksetApiCall.record(
method=method, params=params, ok=ok, http_status=http_status,
result_count=result_count, duration_ms=duration,
message=message, context=self.context, user_id=user_id)
except Exception:
logger.debug('Could not record Brickset call', exc_info=True)
@staticmethod
def is_configured() -> bool:
@@ -52,6 +95,7 @@ class BricksetAPI:
current_app.logger.warning('Brickset API credentials not configured')
return None
started = time.monotonic()
try:
response = requests.get(
f'{self.BASE_URL}/login',
@@ -67,13 +111,19 @@ class BricksetAPI:
if data.get('status') == 'success':
self._user_hash = data.get('hash')
self._log_call('login', self.username, ok=True,
http_status=response.status_code,
started=started)
return self._user_hash
else:
current_app.logger.error(f"Brickset login failed: {data.get('message')}")
return None
self._log_call('login', self.username, ok=False,
http_status=response.status_code, started=started,
message=str(data.get('message')))
return None
except Exception as e:
current_app.logger.error(f'Brickset API authentication error: {str(e)}')
self._log_call('login', self.username, ok=False, started=started,
message=f'{type(e).__name__}: {e}')
return None
def search_sets(self,
@@ -123,6 +173,9 @@ class BricksetAPI:
'params': json.dumps(search_params),
}
summary = json.dumps(search_params, separators=(',', ':'))
started = time.monotonic()
try:
response = requests.get(
f'{self.BASE_URL}/getSets',
@@ -133,10 +186,16 @@ class BricksetAPI:
data = response.json()
if data.get('status') == 'success':
return data.get('sets', [])
sets = data.get('sets', [])
self._log_call('getSets', summary, ok=True,
http_status=response.status_code,
result_count=len(sets), started=started)
return sets
message = data.get('message', 'unknown error')
current_app.logger.error(f'Brickset search failed: {message}')
self._log_call('getSets', summary, ok=False,
http_status=response.status_code, started=started,
message=message)
if 'limit exceeded' in message.lower():
raise RuntimeError('Brickset API limit exceeded')
return []
@@ -144,7 +203,8 @@ class BricksetAPI:
except RuntimeError:
raise
except Exception as e:
current_app.logger.error(f'Brickset API search error: {str(e)}')
self._log_call('getSets', summary, ok=False, started=started,
message=f'{type(e).__name__}: {e}')
return []
def get_set_by_number(self, set_number: str) -> Optional[Dict[str, Any]]:
@@ -174,6 +234,7 @@ class BricksetAPI:
if not user_hash:
return []
started = time.monotonic()
try:
response = requests.get(
f'{self.BASE_URL}/getThemes',
@@ -188,19 +249,31 @@ class BricksetAPI:
if data.get('status') == 'success':
themes = data.get('themes', [])
return [theme.get('theme') for theme in themes if theme.get('theme')]
else:
return []
names = [theme.get('theme') for theme in themes
if theme.get('theme')]
self._log_call('getThemes', ok=True,
http_status=response.status_code,
result_count=len(names), started=started)
return names
self._log_call('getThemes', ok=False,
http_status=response.status_code, started=started,
message=str(data.get('message')))
return []
except Exception as e:
current_app.logger.error(f'Brickset API themes error: {str(e)}')
self._log_call('getThemes', ok=False, started=started,
message=f'{type(e).__name__}: {e}')
return []
# -- instructions -------------------------------------------------------
def _instructions_request(self, method: str,
params: Dict[str, Any]) -> List[Dict[str, Any]]:
summary = ', '.join(f'{k}={v}' for k, v in params.items())
params = dict(params, apiKey=self.api_key)
started = time.monotonic()
try:
response = requests.get(f'{self.BASE_URL}/{method}',
params=params, timeout=20)
@@ -208,14 +281,20 @@ class BricksetAPI:
data = response.json()
if data.get('status') == 'success':
return data.get('instructions', []) or []
instructions = data.get('instructions', []) or []
self._log_call(method, summary, ok=True,
http_status=response.status_code,
result_count=len(instructions), started=started)
return instructions
current_app.logger.error(
f"Brickset {method} failed: {data.get('message')}")
self._log_call(method, summary, ok=False,
http_status=response.status_code, started=started,
message=str(data.get('message')))
return []
except Exception as e:
current_app.logger.error(f'Brickset {method} error: {str(e)}')
self._log_call(method, summary, ok=False, started=started,
message=f'{type(e).__name__}: {e}')
return []
def get_instructions(self, set_number: str) -> List[Dict[str, Any]]:
@@ -246,6 +325,7 @@ class BricksetAPI:
"""
if not self.api_key:
return []
started = time.monotonic()
try:
response = requests.get(f'{self.BASE_URL}/getKeyUsageStats',
params={'apiKey': self.api_key},
@@ -253,8 +333,17 @@ class BricksetAPI:
response.raise_for_status()
data = response.json()
if data.get('status') == 'success':
return data.get('apiKeyUsage', []) or []
usage = data.get('apiKeyUsage', []) or []
self._log_call('getKeyUsageStats', ok=True,
http_status=response.status_code,
result_count=len(usage), started=started)
return usage
self._log_call('getKeyUsageStats', ok=False,
http_status=response.status_code, started=started,
message=str(data.get('message')))
return []
except Exception as e:
current_app.logger.error(f'Brickset key usage error: {str(e)}')
self._log_call('getKeyUsageStats', ok=False, started=started,
message=f'{type(e).__name__}: {e}')
return []