feat(cli): Add log level configuration with flag and environment variable.

- Add --log-level flag with LEGGEN_LOG_LEVEL environment variable support
- Configure loguru to respect log level setting across the application
- Pass log level to uvicorn server for consistent logging
- Log GoCardless API responses and transaction data at debug level
- Supported levels: TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL

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

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Elisiário Couto
2026-01-07 01:31:26 +00:00
parent 9e9b1cf15f
commit c765accfd7
4 changed files with 35 additions and 6 deletions

View File

@@ -133,12 +133,14 @@ def create_app() -> FastAPI:
def server(ctx: click.Context, reload: bool, host: str, port: int):
"""Start the Leggen API server"""
# Get config_dir and database from main CLI context
# Get config_dir, database, and log_level from main CLI context
config_dir = None
database = None
log_level = "info"
if ctx.parent:
config_dir = ctx.parent.params.get("config_dir")
database = ctx.parent.params.get("database")
log_level = ctx.parent.params.get("log_level", "info").lower()
# Set up path manager with user-provided paths
if config_dir:
@@ -153,7 +155,7 @@ def server(ctx: click.Context, reload: bool, host: str, port: int):
factory=True,
host=host,
port=port,
log_level="info",
log_level=log_level,
access_log=True,
reload=True,
reload_dirs=["leggen"], # Watch leggen directory
@@ -164,6 +166,6 @@ def server(ctx: click.Context, reload: bool, host: str, port: int):
app,
host=host,
port=port,
log_level="info",
log_level=log_level,
access_log=True,
)

View File

@@ -4,6 +4,7 @@ from gettext import gettext as _
from pathlib import Path
import click
from loguru import logger
from leggen.utils.config import load_config
from leggen.utils.paths import path_manager
@@ -109,13 +110,25 @@ class Group(click.Group):
show_envvar=True,
help="URL of the leggen API service",
)
@click.option(
"--log-level",
type=click.Choice(
["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
),
default="INFO",
envvar="LEGGEN_LOG_LEVEL",
show_envvar=True,
help="Set the logging level",
)
@click.group(
cls=Group,
context_settings={"help_option_names": ["-h", "--help"]},
)
@click.version_option(package_name="leggen")
@click.pass_context
def cli(ctx: click.Context, config_dir: Path, database: Path, api_url: str):
def cli(
ctx: click.Context, config_dir: Path, database: Path, api_url: str, log_level: str
):
"""
Leggen: An Open Banking CLI
"""
@@ -124,11 +137,20 @@ def cli(ctx: click.Context, config_dir: Path, database: Path, api_url: str):
if "--help" in sys.argv[1:] or "-h" in sys.argv[1:]:
return
# Configure loguru log level
logger.remove() # Remove default handler
logger.add(
sys.stderr,
level=log_level.upper(),
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
)
# Set up path manager with user-provided paths
if config_dir:
path_manager.set_config_dir(config_dir)
if database:
path_manager.set_database_path(database)
# Store API URL in context for commands to use
# Store API URL and log level in context for commands to use
ctx.obj["api_url"] = api_url
ctx.obj["log_level"] = log_level.lower()

View File

@@ -1,6 +1,8 @@
from datetime import datetime
from typing import Any, Dict, List
from loguru import logger
class TransactionProcessor:
"""Handles processing and transformation of raw transaction data"""
@@ -13,6 +15,7 @@ class TransactionProcessor:
) -> List[Dict[str, Any]]:
"""Process raw transaction data into standardized format"""
transactions = []
logger.debug(transaction_data)
# Process booked transactions
for transaction in transaction_data.get("transactions", {}).get("booked", []):

View File

@@ -57,7 +57,9 @@ class GoCardlessService:
_log_rate_limits(response, method, url)
response.raise_for_status()
return response.json()
response_data = response.json()
logger.debug(f"{method} {url} response: {response_data}")
return response_data
async def _get_auth_headers(self) -> Dict[str, str]:
"""Get authentication headers for GoCardless API"""