380 lines
11 KiB
Python
380 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, Union, TypeVar, Protocol
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import status
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy.orm import Query
|
|
|
|
from ApiValidations.Request.base_validations import PydanticBaseModel
|
|
from ApiValidations.handler import BaseModelRegular
|
|
from Services.Postgres.Models.response import PostgresResponse
|
|
from ErrorHandlers.ApiErrorHandlers.api_exc_handler import HTTPExceptionApi
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class DataValidator(Protocol):
|
|
"""Protocol for data validation methods."""
|
|
|
|
@staticmethod
|
|
def validate_data(data: Any, cls_object: Any) -> None:
|
|
"""Validate data and raise HTTPExceptionApi if invalid."""
|
|
...
|
|
|
|
|
|
@dataclass
|
|
class PaginationConfig:
|
|
"""
|
|
Configuration for pagination settings.
|
|
|
|
Attributes:
|
|
page: Current page number (default: 1)
|
|
size: Items per page (default: 10)
|
|
order_field: Field to order by (default: "id")
|
|
order_type: Order direction (default: "asc")
|
|
"""
|
|
|
|
page: int = 1
|
|
size: int = 10
|
|
order_field: str = "id"
|
|
order_type: str = "asc"
|
|
|
|
|
|
class Pagination:
|
|
"""
|
|
Handles pagination logic for query results.
|
|
|
|
Manages page size, current page, ordering, and calculates total pages
|
|
and items based on the data source.
|
|
|
|
Attributes:
|
|
DEFAULT_SIZE: Default number of items per page (10)
|
|
MIN_SIZE: Minimum allowed page size (10)
|
|
MAX_SIZE: Maximum allowed page size (40)
|
|
"""
|
|
|
|
DEFAULT_SIZE = 10
|
|
MIN_SIZE = 10
|
|
MAX_SIZE = 40
|
|
|
|
def __init__(self):
|
|
self.size: int = self.DEFAULT_SIZE
|
|
self.page: int = 1
|
|
self.order_field: str = "id"
|
|
self.order_type: str = "asc"
|
|
self.page_count: int = 1
|
|
self.total_count: int = 0
|
|
self.total_pages: int = 1
|
|
|
|
def change(self, config: PaginationConfig) -> None:
|
|
"""Update pagination settings from config."""
|
|
self.size = (
|
|
config.size
|
|
if self.MIN_SIZE <= config.size <= self.MAX_SIZE
|
|
else self.DEFAULT_SIZE
|
|
)
|
|
self.page = config.page
|
|
self.order_field = config.order_field
|
|
self.order_type = config.order_type
|
|
self._update_page_counts()
|
|
|
|
def feed(self, data: Union[List[Any], PostgresResponse, Query]) -> None:
|
|
"""Calculate pagination based on data source."""
|
|
self.total_count = (
|
|
len(data)
|
|
if isinstance(data, list)
|
|
else data.count if isinstance(data, PostgresResponse) else data.count()
|
|
)
|
|
self._update_page_counts()
|
|
|
|
def _update_page_counts(self) -> None:
|
|
"""Update page counts and validate current page."""
|
|
self.total_pages = max(1, (self.total_count + self.size - 1) // self.size)
|
|
self.page = max(1, min(self.page, self.total_pages))
|
|
self.page_count = (
|
|
self.total_count % self.size
|
|
if self.page == self.total_pages and self.total_count % self.size
|
|
else self.size
|
|
)
|
|
|
|
def as_dict(self) -> Dict[str, Any]:
|
|
"""Convert pagination state to dictionary format."""
|
|
return {
|
|
"size": self.size,
|
|
"page": self.page,
|
|
"totalCount": self.total_count,
|
|
"totalPages": self.total_pages,
|
|
"pageCount": self.page_count,
|
|
"orderField": self.order_field,
|
|
"orderType": self.order_type,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ResponseConfig:
|
|
"""
|
|
Configuration for response formatting.
|
|
|
|
Attributes:
|
|
status_code: HTTP status code (default: "HTTP_200_OK")
|
|
message: Response message
|
|
completed: Operation completion status
|
|
cls_object: Class object for error handling
|
|
"""
|
|
|
|
status_code: str = "HTTP_200_OK"
|
|
message: str = ""
|
|
completed: bool = True
|
|
cls_object: Optional[Any] = None
|
|
|
|
|
|
class BaseJsonResponse:
|
|
"""
|
|
Base class for JSON response handling.
|
|
|
|
Provides common functionality for all response types:
|
|
- Response formatting
|
|
- Pagination handling
|
|
- Data transformation
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
config: ResponseConfig,
|
|
response_model: Optional[Type[T]] = None,
|
|
filter_attributes: Optional[Any] = None,
|
|
):
|
|
self.status_code = getattr(status, config.status_code, status.HTTP_200_OK)
|
|
self.message = config.message
|
|
self.completed = config.completed
|
|
self.filter_attributes = filter_attributes
|
|
self.response_model = response_model
|
|
self.cls_object = config.cls_object
|
|
|
|
def _create_pagination(self) -> Pagination:
|
|
"""Create and configure 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."""
|
|
return JSONResponse(
|
|
status_code=self.status_code,
|
|
content={
|
|
"pagination": pagination.as_dict(),
|
|
"completed": self.completed,
|
|
"message": self.message,
|
|
"data": data,
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def _validate_data(data: Any, expected_type: Type, cls_object: Any) -> None:
|
|
"""Validate data type and raise exception if invalid."""
|
|
if not isinstance(data, expected_type):
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
|
|
class SinglePostgresResponse(BaseJsonResponse):
|
|
"""
|
|
Handles single record responses from PostgreSQL queries.
|
|
|
|
Used when expecting a single record from a database query.
|
|
Validates that the result is a PostgresResponse and contains exactly one record.
|
|
"""
|
|
|
|
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,
|
|
) -> JSONResponse:
|
|
cls._validate_data(result, PostgresResponse, cls_object)
|
|
|
|
if not result.first:
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
instance = cls()
|
|
instance.__init__(
|
|
ResponseConfig(
|
|
status_code=status_code,
|
|
message=message,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
),
|
|
response_model=response_model,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = result.data.get_dict()
|
|
if response_model:
|
|
data = response_model(**data).dump()
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class AlchemyJsonResponse(BaseJsonResponse):
|
|
"""
|
|
Handles multiple record responses from SQLAlchemy queries.
|
|
|
|
Used for database queries returning multiple records.
|
|
Validates that the result is a PostgresResponse and contains multiple records.
|
|
Supports pagination and data transformation through response models.
|
|
"""
|
|
|
|
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:
|
|
cls._validate_data(result, PostgresResponse, cls_object)
|
|
|
|
if result.first:
|
|
raise HTTPExceptionApi(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
instance = cls()
|
|
instance.__init__(
|
|
ResponseConfig(
|
|
status_code=status_code,
|
|
message=message,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
),
|
|
response_model=response_model,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = [
|
|
(
|
|
response_model(**item.get_dict()).dump()
|
|
if response_model
|
|
else item.get_dict()
|
|
)
|
|
for item in result.data
|
|
]
|
|
pagination.feed(data)
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class ListJsonResponse(BaseJsonResponse):
|
|
"""
|
|
Handles responses for list data.
|
|
|
|
Used when working with Python lists that need to be paginated
|
|
and optionally transformed through a response model.
|
|
Validates that the input is a list.
|
|
"""
|
|
|
|
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:
|
|
cls._validate_data(result, list, cls_object)
|
|
|
|
instance = cls()
|
|
instance.__init__(
|
|
ResponseConfig(
|
|
status_code=status_code,
|
|
message=message,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
),
|
|
response_model=response_model,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = [
|
|
response_model(**item).dump() if response_model else item for item in result
|
|
]
|
|
pagination.feed(data)
|
|
|
|
return instance._format_response(pagination, data)
|
|
|
|
|
|
class DictJsonResponse(BaseJsonResponse):
|
|
"""
|
|
Handles responses for dictionary data.
|
|
|
|
Used when working with single dictionary objects that need to be
|
|
transformed through a response model. Validates that the input
|
|
is a dictionary.
|
|
"""
|
|
|
|
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:
|
|
cls._validate_data(result, dict, cls_object)
|
|
|
|
instance = cls()
|
|
instance.__init__(
|
|
ResponseConfig(
|
|
status_code=status_code,
|
|
message=message,
|
|
completed=completed,
|
|
cls_object=cls_object,
|
|
),
|
|
response_model=response_model,
|
|
filter_attributes=filter_attributes,
|
|
)
|
|
|
|
pagination = instance._create_pagination()
|
|
data = response_model(**result).dump() if response_model else result
|
|
|
|
return instance._format_response(pagination, data)
|