""" Abstract base classes for API route and event handling. This module provides core abstractions for route configuration and factory, with support for authentication and event handling. """ from typing import ( Tuple, TypeVar, Optional, Callable, Dict, Any, List, Type, ClassVar, Union, Awaitable, ) from dataclasses import dataclass, field from pydantic import BaseModel from fastapi import Request, Depends, APIRouter from functools import wraps import inspect from ApiLibrary.common.line_number import get_line_number_for_error from ErrorHandlers.Exceptions.api_exc import HTTPExceptionApi from Schemas.rules.rules import EndpointRestriction ResponseModel = TypeVar("ResponseModel", bound=BaseModel) def endpoint_wrapper(url_of_endpoint: Optional[str] = None): """Create a wrapper for endpoints that stores url_of_endpoint in closure. Args: url_of_endpoint: Optional URL path for the endpoint """ def decorator(func: Callable[..., Any]) -> Callable[..., Any]: @wraps(func) async def wrapper( *args: Any, **kwargs: Any ) -> Union[Dict[str, Any], BaseModel]: # Handle both async and sync functions if inspect.iscoroutinefunction(func): result = await func(*args, **kwargs) else: result = func(*args, **kwargs) # If result is a coroutine, await it if inspect.iscoroutine(result): result = await result # Add endpoint to the result if isinstance(result, dict): result["endpoint"] = url_of_endpoint return result elif isinstance(result, BaseModel): # Convert Pydantic model to dict and add endpoint result_dict = result.model_dump() result_dict["endpoint"] = url_of_endpoint return result_dict return result wrapper.url_of_endpoint = url_of_endpoint return wrapper return decorator @dataclass class EndpointFactoryConfig: """Configuration class for API endpoints. Attributes: url_of_endpoint: Full URL path for this endpoint endpoint: URL path for this endpoint method: HTTP method (GET, POST, etc.) summary: Short description for API documentation description: Detailed description for API documentation endpoint_function: Function to handle the endpoint is_auth_required: Whether authentication is required response_model: Optional response model for OpenAPI schema request_model: Optional request model for OpenAPI schema is_event_required: Whether event handling is required extra_options: Additional endpoint options """ url_prefix: str url_endpoint: str url_of_endpoint: str endpoint: str method: str summary: str description: str endpoint_function: Callable[..., Any] # Now accepts any parameters and return type response_model: Optional[type] = None request_model: Optional[type] = None is_auth_required: bool = True is_event_required: bool = False extra_options: Dict[str, Any] = field(default_factory=dict) def __post_init__(self): """Post initialization hook. Wraps endpoint function with appropriate middleware based on configuration: - If auth and event required -> wrap with TokenEventMiddleware - If only event required -> wrap with EventMiddleware - If only auth required -> wrap with MiddlewareModule.auth_required """ # First apply auth/event middleware if self.is_event_required: from middleware import TokenEventMiddleware self.endpoint_function = TokenEventMiddleware.event_required( self.endpoint_function ) elif self.is_auth_required: from middleware import MiddlewareModule self.endpoint_function = MiddlewareModule.auth_required( self.endpoint_function ) # Then wrap with endpoint_wrapper to store url_of_endpoint self.endpoint_function = endpoint_wrapper(self.url_of_endpoint)( self.endpoint_function ) class RouteFactoryConfig: """Configuration class for API route factories. Attributes: name: Route name tags: List of tags for API documentation prefix: URL prefix for all endpoints in this route include_in_schema: Whether to include in OpenAPI schema endpoints: List of endpoint configurations extra_options: Additional route options """ def __init__( self, name: str, tags: List[str], prefix: str, include_in_schema: bool = True, endpoints: List[EndpointFactoryConfig] = None, extra_options: Dict[str, Any] = None, ): self.name = name self.tags = tags self.prefix = prefix self.include_in_schema = include_in_schema self.endpoints = endpoints or [] self.extra_options = extra_options or {} def __post_init__(self): """Validate and normalize configuration after initialization.""" if self.endpoints is None: self.endpoints = [] if self.extra_options is None: self.extra_options = {} def as_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary format.""" return { "name": self.name, "tags": self.tags, "prefix": self.prefix, "include_in_schema": self.include_in_schema, "endpoints": [endpoint.__dict__ for endpoint in self.endpoints], "extra_options": self.extra_options, } class ActionsSchema: """Base class for defining API action schemas. This class handles endpoint registration and validation in the database. Subclasses should implement specific validation logic. """ def __init__(self, endpoint: str): """Initialize with an API endpoint path. Args: endpoint: The API endpoint path (e.g. "/users/create") """ self.endpoint = endpoint def retrieve_action_from_endpoint(self) -> Dict[str, Any]: """Retrieve the endpoint registration from the database. Returns: Dict containing the endpoint registration data Raises: HTTPException: If endpoint is not found in database """ raise NotImplementedError( "Subclasses must implement retrieve_action_from_endpoint" ) class ActionsSchemaFactory: """Factory class for creating and validating action schemas. This class ensures proper initialization and validation of API endpoints through their action schemas. """ def __init__(self, action: ActionsSchema): """Initialize with an action schema. Args: action: The action schema to initialize Raises: HTTPException: If action initialization fails """ self.action = action self.action_match = self.action.retrieve_action_from_endpoint() class MethodToEvent: """Base class for mapping methods to API events with type safety. This class provides a framework for handling API events with proper type checking for tokens and response models. Type Parameters: TokenType: Type of authentication token ResponseModel: Type of response model """ action_key: ClassVar[Optional[str]] = None event_type: ClassVar[Optional[str]] = None event_description: ClassVar[str] = "" event_category: ClassVar[str] = "" __event_keys__: ClassVar[Dict[str, str]] = {} __event_validation__: Dict[str, Tuple[Type, Union[List, tuple]]] = {} @classmethod def retrieve_event_response_model(cls, function_code: str) -> Tuple: """Retrieve event validation for a specific function. Args: function_code: Function identifier Returns: Tuple containing response model and language models """ event_validation_list = cls.__event_validation__.get(function_code, None) if not event_validation_list: raise HTTPExceptionApi( error_code="", lang="en", loc=get_line_number_for_error(), sys_msg="Function not found", ) return event_validation_list[0] @classmethod def retrieve_event_languages(cls, function_code: str) -> Union[List, tuple]: """Retrieve event description for a specific function. Args: function_code: Function identifier Returns: Event description """ event_keys_list = cls.__event_validation__.get(function_code, None) print('event_keys_list', event_keys_list) if not event_keys_list: raise HTTPExceptionApi( error_code="", lang="en", loc=get_line_number_for_error(), sys_msg="Function not found", ) function_language_models: Union[List, tuple] = event_keys_list[1] if not function_language_models: raise HTTPExceptionApi( error_code="", lang="en", loc=get_line_number_for_error(), sys_msg="Function not found", ) return function_language_models @staticmethod def merge_models(language_model: List) -> Tuple: merged_models = {"tr": {}, "en": {}} for model in language_model: for lang in dict(model).keys(): if lang not in merged_models: merged_models[lang] = model[lang] else: merged_models[lang].update(model[lang]) return merged_models @classmethod def retrieve_event_function(cls, function_code: str) -> Dict[str, str]: """Retrieve event parameters for a specific function. Args: function_code: Function identifier Returns: Dictionary of event parameters """ function_event = cls.__event_keys__[function_code] function_itself = getattr(cls, function_event, None) if not function_itself: raise HTTPExceptionApi( error_code="", lang="en", loc=get_line_number_for_error(), sys_msg="Function not found", ) return function_itself @classmethod def retrieve_language_parameters(cls, function_code: str, language: str = "tr") -> Dict[str, str]: """Retrieve language-specific parameters for an event. Args: language: Language code (e.g. 'tr', 'en') function_code: Function identifier Returns: Dictionary of language-specific field mappings """ event_language_models = cls.retrieve_event_languages(function_code) event_response_model = cls.retrieve_event_response_model(function_code) event_response_model_merged = cls.merge_models(event_language_models) event_response_model_merged_lang = event_response_model_merged[language] # Map response model fields to language-specific values print('event_response_model', dict( event_response_model=event_response_model, event_response_model_merged_lang=event_response_model_merged_lang, event_response_model_merged=event_response_model_merged, language=language, function_code=function_code, )) only_language_dict = { field: event_response_model_merged_lang[field] for field in event_response_model.model_fields if field in event_response_model_merged_lang } return { "language_model": only_language_dict, "language_models": event_response_model_merged, }