""" Route configuration and factory module. Handles dynamic route creation based on configurations. """ from typing import Optional, Dict, Any, List, Callable, TypeVar, ParamSpec P = ParamSpec("P") # For function parameters R = TypeVar("R") # For return type from dataclasses import dataclass from functools import wraps from fastapi import APIRouter, Request from fastapi.routing import APIRoute from middleware.auth_middleware import MiddlewareModule from pydantic import BaseModel from AllConfigs.main import MainConfig as Config @dataclass class EndpointFactoryConfig: endpoint: str method: str summary: str description: str endpoint_function: Callable[P, R] # Now accepts any parameters and return type is_auth_required: bool = True is_event_required: bool = False extra_options: Dict[str, Any] = None def __post_init__(self): if self.extra_options is None: self.extra_options = {} class EnhancedEndpointFactory: def __init__(self, router_config: dict): self.router = APIRouter( prefix=router_config["prefix"], tags=router_config["tags"], include_in_schema=router_config.get("include_in_schema", True), ) self.endpoints = router_config["endpoints"] self.protected_routes: Dict[str, List[str]] = {} def create_endpoint(self, config: EndpointFactoryConfig): """ Create an endpoint directly from the configuration. Args: config: EndpointFactoryConfig instance containing endpoint configuration """ endpoint_path = config.endpoint endpoint_function = config.endpoint_function if config.is_auth_required: # endpoint_function = MiddlewareModule.auth_required(endpoint_function) # Track protected routes full_path = f"{self.router.prefix}{endpoint_path}" if full_path not in self.protected_routes: self.protected_routes[full_path] = [] self.protected_routes[full_path].append(config.method.lower()) # Register the endpoint with FastAPI router getattr(self.router, config.method.lower())( endpoint_path, summary=config.summary, description=config.description, **config.extra_options, )(endpoint_function) def get_router(self) -> APIRouter: """Get the configured router.""" return self.router def get_protected_routes(self) -> Dict[str, List[str]]: """Get the protected routes mapping.""" return self.protected_routes async def health_check(request: Request): """Default health check endpoint.""" return {"status": "healthy", "message": "Service is running"} async def ping_test(request: Request, service_name: str = "base-router"): """Default ping test endpoint.""" return {"ping": "pong", "service": service_name} def get_all_routers() -> tuple[List[APIRouter], Dict[str, List[str]]]: """ Get all routers and protected routes from route configurations. Returns: tuple: (routers, protected_routes) """ from ApiEvents.route_configs import get_route_configs routers = [] all_protected_routes = {} # Get route configurations from the registry route_configs = get_route_configs() factory_all = [] for config in route_configs: factory = EnhancedEndpointFactory(config) # Create endpoints from configuration for endpoint_dict in config["endpoints"]: endpoint_config = EndpointFactoryConfig( endpoint=endpoint_dict["endpoint"], method=endpoint_dict["method"], summary=endpoint_dict["summary"], description=endpoint_dict["description"], endpoint_function=endpoint_dict["endpoint_function"], is_auth_required=endpoint_dict["is_auth_required"], is_event_required=endpoint_dict["is_event_required"], extra_options=endpoint_dict.get("extra_options", {}), ) factory.create_endpoint(endpoint_config) factory_all.append(endpoint_config.__dict__) # Add router and protected routes routers.append(factory.get_router()) all_protected_routes.update(factory.get_protected_routes()) return routers, all_protected_routes