mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-04-15 22:26:25 +02:00
feat: PaginatedResponse[T] as discriminated union annotation (#142)
This commit is contained in:
@@ -102,7 +102,9 @@ async def list_events(
|
|||||||
|
|
||||||
#### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
|
#### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
|
||||||
|
|
||||||
Base class and return type for endpoints that support **both** pagination strategies via a `pagination_type` query parameter (using [`paginate()`](crud.md#unified-paginate--both-strategies-on-one-endpoint))
|
Return type for endpoints that support **both** pagination strategies via a `pagination_type` query parameter (using [`paginate()`](crud.md#unified-paginate--both-strategies-on-one-endpoint)).
|
||||||
|
|
||||||
|
When used as a return annotation, `PaginatedResponse[T]` automatically expands to `Annotated[Union[CursorPaginatedResponse[T], OffsetPaginatedResponse[T]], Field(discriminator="pagination_type")]`, so FastAPI emits a proper `oneOf` + discriminator in the OpenAPI schema with no extra boilerplate:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from fastapi_toolsets.crud import PaginationType
|
from fastapi_toolsets.crud import PaginationType
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
"""Base Pydantic schemas for API responses."""
|
"""Base Pydantic schemas for API responses."""
|
||||||
|
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Any, ClassVar, Generic, Literal
|
from typing import Annotated, Any, ClassVar, Generic, Literal, TypeVar, Union
|
||||||
|
|
||||||
from pydantic import BaseModel, ConfigDict
|
from pydantic import BaseModel, ConfigDict, Field
|
||||||
|
|
||||||
from .types import DataT
|
from .types import DataT
|
||||||
|
|
||||||
@@ -138,9 +138,11 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
|
|||||||
|
|
||||||
Base class and return type for endpoints that support both pagination
|
Base class and return type for endpoints that support both pagination
|
||||||
strategies. Use :class:`OffsetPaginatedResponse` or
|
strategies. Use :class:`OffsetPaginatedResponse` or
|
||||||
:class:`CursorPaginatedResponse` when the strategy is fixed; use
|
:class:`CursorPaginatedResponse` when the strategy is fixed.
|
||||||
``PaginatedResponse`` as the return annotation for unified endpoints that
|
|
||||||
dispatch via :meth:`~fastapi_toolsets.crud.factory.AsyncCrud.paginate`.
|
When used as ``PaginatedResponse[T]`` in a return annotation, subscripting
|
||||||
|
returns ``Annotated[Union[CursorPaginatedResponse[T], OffsetPaginatedResponse[T]], Field(discriminator="pagination_type")]``
|
||||||
|
so FastAPI emits a proper ``oneOf`` + discriminator in the OpenAPI schema.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
data: list[DataT]
|
data: list[DataT]
|
||||||
@@ -148,6 +150,16 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
|
|||||||
pagination_type: PaginationType | None = None
|
pagination_type: PaginationType | None = None
|
||||||
filter_attributes: dict[str, list[Any]] | None = None
|
filter_attributes: dict[str, list[Any]] | None = None
|
||||||
|
|
||||||
|
def __class_getitem__( # type: ignore[invalid-method-override]
|
||||||
|
cls, item: type[Any] | tuple[type[Any], ...]
|
||||||
|
) -> type[Any]:
|
||||||
|
if cls is PaginatedResponse and not isinstance(item, TypeVar):
|
||||||
|
return Annotated[ # type: ignore[invalid-return-type]
|
||||||
|
Union[CursorPaginatedResponse[item], OffsetPaginatedResponse[item]], # type: ignore[invalid-type-form]
|
||||||
|
Field(discriminator="pagination_type"),
|
||||||
|
]
|
||||||
|
return super().__class_getitem__(item)
|
||||||
|
|
||||||
|
|
||||||
class OffsetPaginatedResponse(PaginatedResponse[DataT]):
|
class OffsetPaginatedResponse(PaginatedResponse[DataT]):
|
||||||
"""Paginated response with typed offset-based pagination metadata.
|
"""Paginated response with typed offset-based pagination metadata.
|
||||||
|
|||||||
@@ -304,7 +304,7 @@ class TestPaginatedResponse:
|
|||||||
page=1,
|
page=1,
|
||||||
has_more=False,
|
has_more=False,
|
||||||
)
|
)
|
||||||
response = PaginatedResponse[dict](
|
response = PaginatedResponse(
|
||||||
data=[],
|
data=[],
|
||||||
pagination=pagination,
|
pagination=pagination,
|
||||||
)
|
)
|
||||||
@@ -313,6 +313,26 @@ class TestPaginatedResponse:
|
|||||||
assert response.data == []
|
assert response.data == []
|
||||||
assert response.pagination.total_count == 0
|
assert response.pagination.total_count == 0
|
||||||
|
|
||||||
|
def test_class_getitem_with_concrete_type_returns_discriminated_union(self):
|
||||||
|
"""PaginatedResponse[T] with a concrete type returns a discriminated Annotated union."""
|
||||||
|
import typing
|
||||||
|
|
||||||
|
alias = PaginatedResponse[dict]
|
||||||
|
args = typing.get_args(alias)
|
||||||
|
# args[0] is the Union, args[1] is the FieldInfo discriminator
|
||||||
|
union_args = typing.get_args(args[0])
|
||||||
|
assert CursorPaginatedResponse[dict] in union_args
|
||||||
|
assert OffsetPaginatedResponse[dict] in union_args
|
||||||
|
|
||||||
|
def test_class_getitem_with_typevar_returns_generic(self):
|
||||||
|
"""PaginatedResponse[TypeVar] falls through to Pydantic generic parametrisation."""
|
||||||
|
from typing import TypeVar
|
||||||
|
|
||||||
|
T = TypeVar("T")
|
||||||
|
alias = PaginatedResponse[T]
|
||||||
|
# Should be a generic alias, not an Annotated union
|
||||||
|
assert not hasattr(alias, "__metadata__")
|
||||||
|
|
||||||
def test_generic_type_hint(self):
|
def test_generic_type_hint(self):
|
||||||
"""PaginatedResponse supports generic type hints."""
|
"""PaginatedResponse supports generic type hints."""
|
||||||
pagination = OffsetPagination(
|
pagination = OffsetPagination(
|
||||||
|
|||||||
Reference in New Issue
Block a user