367 lines
11 KiB
Python
367 lines
11 KiB
Python
"""
|
|
Response handlers for SQLAlchemy query results with pagination support.
|
|
|
|
This module provides a set of response classes for handling different types of data:
|
|
- Single PostgreSQL records
|
|
- Multiple SQLAlchemy records
|
|
- List data
|
|
- Dictionary data
|
|
|
|
Each response includes pagination information and supports data transformation
|
|
through response models.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
from typing import Any, Dict, List, Optional, Type, TypeVar, Protocol, Generic
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from Services.PostgresDb.Models.response import PostgresResponse
|
|
from ErrorHandlers.ErrorHandlers.api_exc_handler import HTTPExceptionApi
|
|
from Services.pagination import Pagination, PaginationConfig
|
|
|
|
|
|
T = TypeVar("T")
|
|
DataT = TypeVar("DataT")
|
|
|
|
|
|
@dataclass
|
|
class ResponseConfig(Generic[T]):
|
|
"""Configuration for response formatting.
|
|
|
|
Attributes:
|
|
status_code: HTTP status code (default: "HTTP_200_OK")
|
|
message: Response message to include in the response
|
|
completed: Operation completion status flag
|
|
cls_object: Class object for error handling context
|
|
response_model: Optional response model class for data transformation
|
|
"""
|
|
|
|
status_code: str = "HTTP_200_OK"
|
|
message: str = ""
|
|
completed: bool = True
|
|
cls_object: Optional[Any] = None
|
|
response_model: Optional[Type[T]] = None
|
|
|
|
|
|
class ResponseProtocol(Protocol):
|
|
"""Protocol defining required methods for response models."""
|
|
|
|
def dump(self) -> Dict[str, Any]:
|
|
"""Convert model to dictionary format."""
|
|
...
|
|
|
|
|
|
class BaseJsonResponse(Generic[T]):
|
|
"""Base class for JSON response handling.
|
|
|
|
Provides common functionality for all response types including:
|
|
- Response formatting with consistent structure
|
|
- Pagination handling and configuration
|
|
- Data transformation through response models
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
result: Any,
|
|
response_model: Optional[Type[T]] = None,
|
|
status_code: str = "HTTP_200_OK",
|
|
completed: bool = True,
|
|
cls_object: Optional[Any] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
) -> None:
|
|
"""Initialize response handler.
|
|
|
|
Args:
|
|
message: Response message
|
|
result: Query result or data
|
|
response_model: Optional model for data transformation
|
|
status_code: HTTP status code
|
|
completed: Operation completion status
|
|
cls_object: Class object for error context
|
|
filter_attributes: Optional pagination and filtering attributes
|
|
"""
|
|
self.status_code = getattr(status, status_code, status.HTTP_200_OK)
|
|
self.message = message
|
|
self.completed = completed
|
|
self.filter_attributes = filter_attributes
|
|
self.response_model = response_model
|
|
self.cls_object = cls_object
|
|
self.result = result
|
|
|
|
def _create_pagination(self) -> Pagination:
|
|
"""Create and configure pagination instance.
|
|
|
|
Returns:
|
|
Configured Pagination instance
|
|
"""
|
|
pagination = Pagination()
|
|
if self.filter_attributes:
|
|
pagination.change(
|
|
PaginationConfig(
|
|
page=self.filter_attributes.page,
|
|
size=self.filter_attributes.size,
|
|
order_field=self.filter_attributes.order_field,
|
|
order_type=self.filter_attributes.order_type,
|
|
)
|
|
)
|
|
return pagination
|
|
|
|
def _format_response(self, pagination: Pagination, data: Any) -> JSONResponse:
|
|
"""Format final JSON response with pagination.
|
|
|
|
Args:
|
|
pagination: Pagination instance with configuration
|
|
data: Response data to include
|
|
|
|
Returns:
|
|
Formatted JSONResponse
|
|
"""
|
|
return JSONResponse(
|
|
status_code=self.status_code,
|
|
content={
|
|
"pagination": pagination.as_dict(),
|
|
"completed": self.completed,
|
|
"message": self.message,
|
|
"data": data,
|
|
},
|
|
)
|
|
|
|
def _transform_data(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Transform data using response model if provided.
|
|
|
|
Args:
|
|
data: Raw data dictionary
|
|
|
|
Returns:
|
|
Transformed data dictionary
|
|
"""
|
|
if self.response_model:
|
|
return self.response_model(**data).dump()
|
|
return data
|
|
|
|
@staticmethod
|
|
def _validate_data(data: Any, expected_type: Type, cls_object: Any) -> None:
|
|
"""Validate data type and raise exception if invalid.
|
|
|
|
Args:
|
|
data: Data to validate
|
|
expected_type: Expected type of data
|
|
cls_object: Class object for error context
|
|
|
|
Raises:
|
|
HTTPExceptionApi: If data type is invalid
|
|
"""
|
|
if not isinstance(data, expected_type):
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
|
|
class SinglePostgresResponse(BaseJsonResponse[T]):
|
|
"""Handler for single record responses from PostgreSQL queries."""
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
result: PostgresResponse,
|
|
response_model: Optional[Type[T]] = None,
|
|
status_code: str = "HTTP_200_OK",
|
|
completed: bool = True,
|
|
cls_object: Optional[Any] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
) -> JSONResponse:
|
|
"""Create response for single PostgreSQL record.
|
|
|
|
Args:
|
|
message: Response message
|
|
result: PostgreSQL query result
|
|
response_model: Optional model for data transformation
|
|
status_code: HTTP status code
|
|
completed: Operation completion status
|
|
cls_object: Class object for error context
|
|
filter_attributes: Optional pagination and filtering attributes
|
|
|
|
Returns:
|
|
Formatted JSON response
|
|
|
|
Raises:
|
|
HTTPExceptionApi: If result is invalid or empty
|
|
"""
|
|
cls._validate_data(result, PostgresResponse, cls_object)
|
|
|
|
if not result.first:
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
instance = super().__new__(cls)
|
|
instance.__init__(
|
|
message=message,
|
|
result=result,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = instance._transform_data(result.data.get_dict())
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class AlchemyJsonResponse(BaseJsonResponse[T]):
|
|
"""Handler for multiple record responses from SQLAlchemy queries."""
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
result: PostgresResponse,
|
|
response_model: Optional[Type[T]] = None,
|
|
status_code: str = "HTTP_200_OK",
|
|
completed: bool = True,
|
|
cls_object: Optional[Any] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
) -> JSONResponse:
|
|
"""Create response for multiple SQLAlchemy records.
|
|
|
|
Args:
|
|
message: Response message
|
|
result: PostgreSQL query result
|
|
response_model: Optional model for data transformation
|
|
status_code: HTTP status code
|
|
completed: Operation completion status
|
|
cls_object: Class object for error context
|
|
filter_attributes: Optional pagination and filtering attributes
|
|
|
|
Returns:
|
|
Formatted JSON response
|
|
|
|
Raises:
|
|
HTTPExceptionApi: If result is invalid
|
|
"""
|
|
cls._validate_data(result, PostgresResponse, cls_object)
|
|
|
|
if result.first:
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
instance = super().__new__(cls)
|
|
instance.__init__(
|
|
message=message,
|
|
result=result,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = [instance._transform_data(item.get_dict()) for item in result.data]
|
|
pagination.feed(data)
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class ListJsonResponse(BaseJsonResponse[T]):
|
|
"""Handler for list data responses."""
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
result: List[Any],
|
|
response_model: Optional[Type[T]] = None,
|
|
status_code: str = "HTTP_200_OK",
|
|
completed: bool = True,
|
|
cls_object: Optional[Any] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
) -> JSONResponse:
|
|
"""Create response for list data.
|
|
|
|
Args:
|
|
message: Response message
|
|
result: List of data items
|
|
response_model: Optional model for data transformation
|
|
status_code: HTTP status code
|
|
completed: Operation completion status
|
|
cls_object: Class object for error context
|
|
filter_attributes: Optional pagination and filtering attributes
|
|
|
|
Returns:
|
|
Formatted JSON response
|
|
"""
|
|
cls._validate_data(result, list, cls_object)
|
|
|
|
instance = super().__new__(cls)
|
|
instance.__init__(
|
|
message=message,
|
|
result=result,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = [instance._transform_data(item) for item in result]
|
|
pagination.feed(data)
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class DictJsonResponse(BaseJsonResponse[T]):
|
|
"""Handler for dictionary data responses."""
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
result: Dict[str, Any],
|
|
response_model: Optional[Type[T]] = None,
|
|
status_code: str = "HTTP_200_OK",
|
|
completed: bool = True,
|
|
cls_object: Optional[Any] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
) -> JSONResponse:
|
|
"""Create response for dictionary data.
|
|
|
|
Args:
|
|
message: Response message
|
|
result: Dictionary data
|
|
response_model: Optional model for data transformation
|
|
status_code: HTTP status code
|
|
completed: Operation completion status
|
|
cls_object: Class object for error context
|
|
filter_attributes: Optional pagination and filtering attributes
|
|
|
|
Returns:
|
|
Formatted JSON response
|
|
"""
|
|
cls._validate_data(result, dict, cls_object)
|
|
|
|
instance = super().__new__(cls)
|
|
instance.__init__(
|
|
message=message,
|
|
result=result,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = instance._transform_data(result)
|
|
|
|
return instance._format_response(pagination, data)
|