uriyyo/fastapi-pagination

Getting `ResponseValidationError`

Chiggy-Playz opened this issue · 7 comments

Chiggy-Playz commented

Searched the issues but did not find anyone facing this issue. I'm trying to paginate my results from my database, using sqlalchemy ORM
Python version: 3.11.4

Error:

api-1 | INFO:     Started server process [8]
api-1 | INFO:     Waiting for application startup.
api-1 | INFO:     Application startup complete.
api-1 | INFO:     172.18.0.1:53714 - "GET /api/servers/?query=2 HTTP/1.1" 500 Internal Server Error
api-1 | ERROR:    Exception in ASGI application
api-1 | Traceback (most recent call last):
api-1 |   File "/usr/local/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 435, in run_asgi
api-1 |     result = await app(  # type: ignore[func-returns-value]
api-1 |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
api-1 |   File "/usr/local/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__
api-1 |     return await self.app(scope, receive, send)
api-1 |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
api-1 |   File "/usr/local/lib/python3.11/site-packages/fastapi/applications.py", line 289, in __call__
api-1 |     await super().__call__(scope, receive, send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/applications.py", line 122, in __call__
api-1 |     await self.middleware_stack(scope, receive, send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 184, in __call__
api-1 |     raise exc
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 162, in __call__
api-1 |     await self.app(scope, receive, _send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
api-1 |     raise exc
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
api-1 |     await self.app(scope, receive, sender)
api-1 |   File "/usr/local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
api-1 |     raise e
api-1 |   File "/usr/local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
api-1 |     await self.app(scope, receive, send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
api-1 |     await route.handle(scope, receive, send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
api-1 |     await self.app(scope, receive, send)
api-1 |   File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
api-1 |     response = await func(request)
api-1 |                ^^^^^^^^^^^^^^^^^^^
api-1 |   File "/usr/local/lib/python3.11/site-packages/fastapi/routing.py", line 291, in app
api-1 |     content = await serialize_response(
api-1 |               ^^^^^^^^^^^^^^^^^^^^^^^^^
api-1 |   File "/usr/local/lib/python3.11/site-packages/fastapi/routing.py", line 154, in serialize_response
api-1 |     raise ResponseValidationError(
api-1 | fastapi.exceptions.ResponseValidationError

Code:

from fastapi_pagination.ext.sqlalchemy import paginate

@router.get("/", response_model=Page[ServerResponse])
async def filter_servers(
    query: str = Query(description="Query to filter servers"),
    game_type: GameType | None = Query(None, description="Game type of server"),
    platform: list[Platform] | None = Query(None, description="Platform(s) of server"),
    official: bool | None = Query(None, description="Whether server is official"),
    session: AsyncSession = Depends(get_db_session),
):
    logger.info(
        f"Filtering servers with query: {query}, game_type: {game_type}, platform: {platform}, official: {official}"
    )
    statement = select(ServerModel).where(ServerModel.name.ilike(f"%{query}%"))
    if game_type:
        statement = statement.where(ServerModel.game_type == game_type)
    if platform:
        statement = statement.where(ServerModel.platforms == platform)
    if official is not None:
        statement = statement.where(ServerModel.official == official)

    return paginate(
        session,
        statement,
    )
uriyyo commented

Hi @Chiggy-Playz,

Could you please show me full exception traceback? Also, it will be helpful to see definitions of ServerResponse and ServerModel.

Chiggy-Playz commented

Hey @uriyyo
I continued working on my project and have added a bit more stuff since then, so right now my models look like this:

class GameType(enum.Enum):
    """Enum for game type."""

    ase = "ase"
    asa = "asa"


class Platform(enum.Enum):
    """Enum for platform."""

    xbox = "Xbox"
    playstation = "Playstation"
    pc = "PC"
    unknown = "Unknown"
    any = "Any"


class PlatformType(TypeDecorator):
    """Platform type for server model."""

    impl = JSONB
    cache_ok = True

    def process_bind_param(self, value, dialect):
        """Used to convert enum to string."""
        if value:
            value = [platform.value for platform in value]
        return value

    def process_result_value(self, value, dialect):
        """Used to convert string to enum."""
        if value:
            value = [Platform(platform) for platform in value]
        return value


class ServerModel(Base):
    """Model for server."""

    __tablename__ = "servers"
    __table_args__ = (
        UniqueConstraint(
            "game_type",
            "name",
            "platforms",
            "official",
            name="unique_server",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True)
    game_type: Mapped[GameType] = mapped_column(Enum(GameType), nullable=False)
    name: Mapped[str] = mapped_column(String, nullable=False)
    platforms: Mapped[list[Platform]] = mapped_column(
        PlatformType,
        nullable=False,
    )
    official: Mapped[bool] = mapped_column(Boolean, nullable=False)
    map: Mapped[str] = mapped_column(String, nullable=False)

    details: Mapped["ServerDetailsModel"] = relationship(
        back_populates="server",
        lazy="joined",
    )


class ServerDetailsModel(Base):
    """Model for server_details."""

    __tablename__ = "server_details"

    id: Mapped[int] = mapped_column(ForeignKey("servers.id"), primary_key=True)
    players: Mapped[int] = mapped_column(Integer, nullable=False)
    ip: Mapped[str] = mapped_column(String, nullable=False)
    port: Mapped[int] = mapped_column(Integer, nullable=False)
    day: Mapped[int] = mapped_column(Integer, nullable=False)
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
    )
    download_characters: Mapped[bool] = mapped_column(Boolean, nullable=False)
    download_items: Mapped[bool] = mapped_column(Boolean, nullable=False)

    server: Mapped[ServerModel] = relationship(back_populates="details")

And here's the response model:

class ServerResponse(BaseModel):
    """Server Response."""

    id: int
    game_type: GameType
    name: str
    platforms: list[str]
    official: bool
    map: str
    # Commented rn because I don't know how to make this work with pagination, since the below properties are not on the sqlalchemy model directly but instead on .details attribute (ServerDetailsModel)
    # players: int
    # ip: str
    # port: int
    # day: int
    # download_items: bool
    # download_characters: bool
    # last_seen_at: datetime

Here's my endpoint:

@router.get("/", response_model=Page[ServerResponse])
async def filter_servers(
    query: str = Query(description="Query to filter servers"),
    game_type: GameType = Query(None, description="Game type of server"),
    platform: list[Platform] = Query(None, description="Platform(s) of server"),
    official: bool = Query(None, description="Whether server is official"),
    db: AsyncSession = Depends(get_db_session),
):
    statement = select(ServerModel).where(
        ServerModel.name.ilike(f"%{query}%"),
    )
    if game_type is not None:
        statement = statement.where(ServerModel.game_type == game_type)
    if platform is not None:
        statement = statement.where(ServerModel.platforms == platform)
    if official is not None:
        statement = statement.where(ServerModel.official == official)
    return paginate(db, statement)

Here's the full exception thats getting logged (It feels like its cut at the end but thats all there is):

2024-02-20 13:40:14.440 | INFO     | logging:callHandlers:1706 - 172.18.0.1:59614 - "GET /api/servers/?query=knights HTTP/1.1" 500
2024-02-20 13:40:14.441 | ERROR    | logging:callHandlers:1706 - Exception in ASGI application

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/usr/local/lib/python3.11/multiprocessing/spawn.py", line 120, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "/usr/local/lib/python3.11/multiprocessing/spawn.py", line 133, in _main
    return self._bootstrap(parent_sentinel)
  File "/usr/local/lib/python3.11/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.11/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.11/site-packages/uvicorn/_subprocess.py", line 76, in subprocess_started
    target(sockets=sockets)
  File "/usr/local/lib/python3.11/site-packages/uvicorn/server.py", line 61, in run
    return asyncio.run(self.serve(sockets=sockets))
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
> File "/usr/local/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 435, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
  File "/usr/local/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 78, in __call__
    return await self.app(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/fastapi/applications.py", line 289, in __call__
    await super().__call__(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/starlette/applications.py", line 122, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 184, in __call__
    raise exc
  File "/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File "/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
    raise exc
  File "/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "/usr/local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
    raise e
  File "/usr/local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
    await self.app(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
    await route.handle(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "/usr/local/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
  File "/usr/local/lib/python3.11/site-packages/fastapi/routing.py", line 291, in app
    content = await serialize_response(
  File "/usr/local/lib/python3.11/site-packages/fastapi/routing.py", line 154, in serialize_response
    raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError
uriyyo commented

@Chiggy-Playz I used code that you provided and I can't see any issue:

import enum
from datetime import datetime

from fastapi import APIRouter, Depends, Query
from pydantic import AliasPath, BaseModel, Field
from sqlalchemy import Boolean, DateTime, Enum, ForeignKey, Integer, String, TypeDecorator, UniqueConstraint, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column, relationship

from fastapi_pagination import Page, add_pagination
from fastapi_pagination.ext.sqlalchemy import paginate


class GameType(enum.Enum):
    """Enum for game type."""

    ase = "ase"
    asa = "asa"


class Platform(enum.Enum):
    """Enum for platform."""

    xbox = "Xbox"
    playstation = "Playstation"
    pc = "PC"
    unknown = "Unknown"
    any = "Any"


class PlatformType(TypeDecorator):
    """Platform type for server model."""

    impl = JSONB
    cache_ok = True

    def process_bind_param(self, value, dialect):
        """Used to convert enum to string."""
        if value:
            value = [platform.value for platform in value]
        return value

    def process_result_value(self, value, dialect):
        """Used to convert string to enum."""
        if value:
            value = [Platform(platform) for platform in value]
        return value


class Base(DeclarativeBase, MappedAsDataclass, kw_only=True):
    pass


class ServerModel(Base):
    """Model for server."""

    __tablename__ = "servers"
    __table_args__ = (
        UniqueConstraint(
            "game_type",
            "name",
            "platforms",
            "official",
            name="unique_server",
        ),
    )

    id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True)
    game_type: Mapped[GameType] = mapped_column(Enum(GameType), nullable=False)
    name: Mapped[str] = mapped_column(String, nullable=False)
    platforms: Mapped[list[Platform]] = mapped_column(
        PlatformType,
        nullable=False,
    )
    official: Mapped[bool] = mapped_column(Boolean, nullable=False)
    map: Mapped[str] = mapped_column(String, nullable=False)

    details: Mapped["ServerDetailsModel"] = relationship(
        back_populates="server",
        lazy="joined",
    )


class ServerDetailsModel(Base):
    """Model for server_details."""

    __tablename__ = "server_details"

    id: Mapped[int] = mapped_column(ForeignKey("servers.id"), primary_key=True)
    players: Mapped[int] = mapped_column(Integer, nullable=False)
    ip: Mapped[str] = mapped_column(String, nullable=False)
    port: Mapped[int] = mapped_column(Integer, nullable=False)
    day: Mapped[int] = mapped_column(Integer, nullable=False)
    last_seen_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
    )
    download_characters: Mapped[bool] = mapped_column(Boolean, nullable=False)
    download_items: Mapped[bool] = mapped_column(Boolean, nullable=False)

    server: Mapped[ServerModel] = relationship(back_populates="details", default=None)


def _details_alias(name: str) -> AliasPath:
    return AliasPath("details", name)


class ServerResponse(BaseModel):
    """Server Response."""

    model_config = {
        "populate_by_name": True,
    }

    id: int
    game_type: GameType
    name: str
    platforms: list[str]
    official: bool
    map: str

    players: int = Field(validation_alias=_details_alias("players"))
    ip: str = Field(validation_alias=_details_alias("ip"))
    port: int = Field(validation_alias=_details_alias("port"))
    day: int = Field(validation_alias=_details_alias("day"))
    download_items: bool = Field(validation_alias=_details_alias("download_items"))
    download_characters: bool = Field(validation_alias=_details_alias("download_characters"))
    last_seen_at: datetime = Field(validation_alias=_details_alias("last_seen_at"))


engine = create_async_engine(
    "postgresql+asyncpg://postgres:postgres@localhost:5433/postgres",
    future=True,
)


async def get_db_session():
    async with AsyncSession(engine) as session:
        yield session


router = APIRouter()


@router.get("/", response_model=Page[ServerResponse])
async def filter_servers(
    query: str = Query(description="Query to filter servers"),
    game_type: GameType = Query(None, description="Game type of server"),
    platform: list[Platform] = Query(None, description="Platform(s) of server"),
    official: bool = Query(None, description="Whether server is official"),
    db: AsyncSession = Depends(get_db_session),
):
    statement = select(ServerModel).where(ServerModel.name.ilike(f"%{query}%"))

    if game_type is not None:
        statement = statement.where(ServerModel.game_type == game_type)
    if platform is not None:
        statement = statement.where(ServerModel.platforms == platform)
    if official is not None:
        statement = statement.where(ServerModel.official == official)

    return await paginate(db, statement)


async def lifespan(_):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        async with session.begin():
            # Create some servers
            session.add_all(
                [
                    ServerModel(
                        id=1,
                        game_type=GameType.ase,
                        name="server1",
                        platforms=[Platform.xbox, Platform.playstation],
                        official=True,
                        map="map1",
                        details=ServerDetailsModel(
                            id=1,
                            players=10,
                            ip="1.1.1.1",
                            port=1234,
                            day=1,
                            last_seen_at=datetime.now(),
                            download_characters=True,
                            download_items=True,
                        ),
                    ),
                ],
            )

    yield


app = APIRouter(lifespan=lifespan)
add_pagination(app)

app.include_router(router, prefix="/servers")

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

When I call endpoint it returns me:

{
  "items": [
    {
      "id": 1,
      "game_type": "ase",
      "name": "server1",
      "platforms": [
        "Xbox",
        "Playstation"
      ],
      "official": true,
      "map": "map1",
      "players": 10,
      "ip": "1.1.1.1",
      "port": 1234,
      "day": 1,
      "download_items": true,
      "download_characters": true,
      "last_seen_at": "2024-02-20T18:23:23.298450Z"
    }
  ],
  "total": 1,
  "page": 1,
  "size": 50,
  "pages": 1
}

Could you please check code, maybe you missed stmth on your side?
One thing that I noticed, that you didn't await paginate, that might be an issue.

@router.get("/", response_model=Page[ServerResponse])
async def filter_servers(
    query: str = Query(description="Query to filter servers"),
    game_type: GameType = Query(None, description="Game type of server"),
    platform: list[Platform] = Query(None, description="Platform(s) of server"),
    official: bool = Query(None, description="Whether server is official"),
    db: AsyncSession = Depends(get_db_session),
):
    statement = select(ServerModel).where(
        ServerModel.name.ilike(f"%{query}%"),
    )
    if game_type is not None:
        statement = statement.where(ServerModel.game_type == game_type)
    if platform is not None:
        statement = statement.where(ServerModel.platforms == platform)
    if official is not None:
        statement = statement.where(ServerModel.official == official)
    return await paginate(db, statement)
Chiggy-Playz commented

Oh....
It seems the missing await fixed the issue 😅
I feel like an idiot now :)
Tysm!

uriyyo commented

No worries, I am happy to help you)

Chiggy-Playz commented

Also, thank you for the validation_alias=_details_alias("field") thingy! It was exactly what i was looking for 😁

  • ❤️1
uriyyo commented

Always happy to help)

  • ❤️1