python - FastAPI RouteLogger fails to process path parameters, throwing KeyError for dynamic routes - Stack Overflow

admin管理员组

文章数量:1435859

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"})

本文标签：