I'm building a FastAPI application with a custom RouteLogger that logs route information during startup. The logger works for static routes but fails when processing dynamic routes with path parameters.
Error logs show:
{
"error_type": "KeyError",
"error_message": "'column'",
"context": {
"context": "log_route_info",
"route_path": "/api/general/distribution/{column}",
"route_name": "get_distribution"
}
}
{
"error_type": "KeyError",
"error_message": "'column1'",
"context": {
"context": "log_route_info",
"route_path": "/api/general/crosstab/{column1}/{column2}",
"route_name": "get_crosstab"
}
}
{
"error_type": "KeyError",
"error_message": "'column'",
"context": {
"context": "log_route_info",
"route_path": "/api/general/filters/{column}",
"route_name": "get_filter_values"
}
}
My route definitions:
@router.get("/distribution/{column}", response_model=DistributionResponse)
async def get_distribution(
column: str = Path(..., description="Column to analyze"),
filters: Optional[str] = Query(None, description="Optional filters as JSON"),
export_format: Optional[ResultFormat] = Query(None, description="Export format if needed")
) -> DistributionResponse:
# Route implementation
@router.get("/crosstab/{column1}/{column2}", response_model=CrosstabResponse)
async def get_crosstab(
column1: str = Path(..., description="First column to analyze"),
column2: str = Path(..., description="Second column to analyze"),
filters: Optional[str] = Query(None, description="Optional filters as JSON"),
export_format: Optional[ResultFormat] = Query(None, description="Export format if needed")
) -> CrosstabResponse:
# Route implementation
RouteLogger successfully processes static routes like /api/health/ but fails for dynamic routes with path parameters. The logger attempts to extract parameter information but can't handle path parameters correctly.
How can I modify the RouteLogger to properly handle FastAPI path parameters without throwing KeyErrors?
Environment:
- FastAPI latest version
- Python 3.12
- Windows/Linux
- APILogger built on loguru
RouteLogger code:
from fastapi import FastAPI, routing, Depends
from fastapi.routing import APIRoute
from typing import Dict, Any, List, Set, Optional, Type, Union, get_type_hints
from enum import Enum
import inspect
import re
from pydantic import BaseModel
from api.core.logging import APILogger
from api.core.config import LOGGING_CONFIG
class RouteLogger:
"""Enhanced route logger with proper parameter extraction."""
def __init__(self):
self.logger = APILogger("RouteLogger", LOGGING_CONFIG)
self.logger.log_info("Route logger initialized")
def __init__(self):
self.logger = APILogger("RouteLogger", LOGGING_CONFIG)
self.logger.log_info("Route logger initialized")
def _get_path_parameters(self, route: APIRoute) -> Dict[str, Any]:
"""Extract path parameters with proper error handling."""
path_params = {}
try:
# Extract path parameter names using regex
param_pattern = r"{([^}]+)}"
path_matches = re.finditer(param_pattern, route.path)
for match in path_matches:
param_name = match.group(1)
param_info = {
'name': param_name,
'required': True,
'kind': 'PATH',
'type': 'str' # Default type for path parameters
}
# Try to get more specific type information from endpoint
if hasattr(route, 'endpoint'):
sig = inspect.signature(route.endpoint)
if param_name in sig.parameters:
param = sig.parameters[param_name]
if param.annotation != inspect.Parameter.empty:
param_info['type'] = str(param.annotation)
path_params[param_name] = param_info
return path_params
except Exception as e:
self.logger.log_error(e, {
"context": "_get_path_parameters",
"route_path": route.path
})
return {}
def _get_endpoint_parameters(self, route: APIRoute) -> List[Dict[str, Any]]:
"""Get endpoint parameters with improved error handling."""
parameters = []
try:
# Get path parameters first
path_params = self._get_path_parameters(route)
parameters.extend(list(path_params.values()))
# Get query parameters from endpoint signature
if hasattr(route, 'endpoint') and route.endpoint:
signature = inspect.signature(route.endpoint)
type_hints = get_type_hints(route.endpoint)
for name, param in signature.parameters.items():
# Skip special parameters and path parameters
if name in {'request', 'background_tasks'} or name in path_params:
continue
param_info = {
'name': name,
'required': param.default == inspect.Parameter.empty,
'kind': str(param.kind),
'type': self._get_parameter_type(param, type_hints.get(name))
}
parameters.append(param_info)
return parameters
except Exception as e:
self.logger.log_error(e, {
"context": "_get_endpoint_parameters",
"route_path": getattr(route, "path", "unknown"),
"route_name": getattr(route, "name", "unknown"),
"parameters_collected": parameters
})
return []
def _get_parameter_type(
self,
param: inspect.Parameter,
type_hint: Optional[Type] = None
) -> str:
"""Get parameter type with proper null handling."""
try:
if type_hint:
if hasattr(type_hint, '__origin__'):
origin = type_hint.__origin__
args = getattr(type_hint, '__args__', [])
if origin is Union and type(None) in args:
non_none_type = next(arg for arg in args if arg is not type(None))
return f"Optional[{self._get_parameter_type(param, non_none_type)}]"
return str(origin.__name__)
return str(type_hint.__name__)
if param.annotation != inspect.Parameter.empty:
return str(param.annotation)
return 'unknown'
except Exception as e:
self.logger.log_error(e, {"context": "_get_parameter_type"})
return 'unknown'
def log_route_info(self, route: APIRoute) -> None:
"""Log route information with proper error handling."""
try:
route_info = {
'path': route.path,
'name': route.name,
'methods': sorted(list(route.methods)) if route.methods else [],
'parameters': self._get_endpoint_parameters(route) or []
}
self.logger.log_info(
f"Registered route: {route.path}",
extra=route_info
)
if route_info['parameters']:
self.logger.log_debug(
f"Route parameters for {route_info['path']}:",
extra={'parameters': route_info['parameters']}
)
except Exception as e:
self.logger.log_error(e, {
"context": "log_route_info",
"route_path": getattr(route, "path", "unknown"),
"route_name": getattr(route, "name", "unknown")
})
def setup_route_logging(self, app: FastAPI) -> None:
"""Set up route logging with proper error handling."""
try:
self.logger.log_info("Setting up route logging")
for route in app.routes:
if isinstance(route, APIRoute):
self.log_route_info(route)
self.logger.log_info("Route logging setup completed")
except Exception as e:
self.logger.log_error(e, {"context": "setup_route_logging"})