115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""
|
|
Base request models for API endpoints.
|
|
|
|
This module provides base request models that can be used across different endpoints
|
|
to ensure consistent request handling and validation.
|
|
"""
|
|
|
|
from typing import Union, Optional, Any
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ApiLayers.ApiValidations.Custom.token_objects import (
|
|
EmployeeTokenObject,
|
|
OccupantTokenObject,
|
|
)
|
|
from ApiLayers.ApiValidations.Custom.wrapper_contexts import AuthContext, EventContext
|
|
from ApiLayers.AllConfigs.Token.config import Auth
|
|
from Services.PostgresDb.Models.pagination import Pagination, PaginationResult, QueryOptions
|
|
|
|
|
|
TokenDictType = Union[EmployeeTokenObject, OccupantTokenObject]
|
|
|
|
|
|
class EndpointBaseRequestModel(BaseModel):
|
|
|
|
data: dict = Field(..., description="Data to be sent with the request")
|
|
|
|
class Config:
|
|
json_schema_extra = {"data": {"key": "value"}}
|
|
|
|
|
|
class ContextRetrievers:
|
|
"""Utility class to retrieve context from functions."""
|
|
|
|
is_auth: bool = False
|
|
is_event: bool = False
|
|
key_: str = ""
|
|
|
|
def __init__(self, func):
|
|
self.func = func
|
|
if hasattr(self.func, "auth_context"):
|
|
self.is_auth = True
|
|
self.key_ = "auth_context"
|
|
elif hasattr(self.func, "event_context"):
|
|
self.is_event = hasattr(self.func, "event_context")
|
|
self.key_ = "event_context"
|
|
|
|
@property
|
|
def key(self) -> Union[str, None]:
|
|
"""Retrieve key context from a function."""
|
|
return self.key_
|
|
|
|
@property
|
|
def context(self) -> Union[AuthContext, EventContext, None]:
|
|
"""Retrieve authentication or event context from a function."""
|
|
return getattr(self.func, self.key, None)
|
|
|
|
@property
|
|
def request(self) -> Union[Any, None]:
|
|
"""Retrieve request context from a function."""
|
|
return getattr(self.context, "request", None)
|
|
|
|
@property
|
|
def token(self) -> TokenDictType:
|
|
"""Retrieve token context from a function."""
|
|
if self.is_auth or self.is_event:
|
|
return getattr(self.context, "auth", None)
|
|
|
|
@property
|
|
def url(self) -> Union[str, None]:
|
|
"""Retrieve URL context from a function."""
|
|
return getattr(self.context, "url", None)
|
|
|
|
@property
|
|
def code(self) -> Union[str, None]:
|
|
"""Retrieve code context from a function."""
|
|
if self.is_event:
|
|
return getattr(self.context, "code", None)
|
|
return None
|
|
|
|
@property
|
|
def base(self) -> Optional[dict[str, Any]]:
|
|
"""Retrieve base request model from a function."""
|
|
return getattr(self.context, "base", None)
|
|
|
|
@property
|
|
def get_token(self) -> Optional[str]:
|
|
"""Retrieve access key from a function."""
|
|
return getattr(self.request, "headers", {}).get(Auth.ACCESS_TOKEN_TAG, None)
|
|
|
|
|
|
class BaseRouteModel:
|
|
|
|
context_retriever: Union[ContextRetrievers] = None
|
|
|
|
|
|
class ListOptionsBase:
|
|
|
|
def __init__(self, table, list_options, model_query: Optional[BaseModel] = None):
|
|
self.table = table
|
|
self.list_options = list_options
|
|
self.model_query = model_query
|
|
|
|
def init_list_options(self) -> tuple:
|
|
db_session = self.table.new_session()
|
|
query_options = QueryOptions(table=self.table, data=self.list_options, model_query=self.model_query)
|
|
return db_session, query_options
|
|
|
|
def paginated_result(self, records) -> PaginationResult:
|
|
pagination = Pagination(data=records)
|
|
if isinstance(self.list_options, dict):
|
|
pagination.change(**self.list_options)
|
|
elif isinstance(self.list_options, BaseModel):
|
|
pagination.change(**self.list_options.model_dump())
|
|
return PaginationResult(data=records, pagination=pagination)
|