Move seasons business logic to the services

Author: koldakov

futuramaapi/routers/rest/seasons/api.py

@@ -1,14 +1,18 @@
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, HTTPException, Path, status
+from fastapi import APIRouter, Path, status
 from fastapi_pagination import Page
-from sqlalchemy.ext.asyncio.session import AsyncSession
 
 from futuramaapi.repositories import INT32
-from futuramaapi.repositories.session import get_async_session
-from futuramaapi.routers.exceptions import ModelNotFoundError, NotFoundResponse
-
-from .schemas import Season
+from futuramaapi.routers.exceptions import NotFoundResponse
+from futuramaapi.routers.services.seasons.get_season import (
+    GetSeasonResponse,
+    GetSeasonService,
+)
+from futuramaapi.routers.services.seasons.list_seasons import (
+    ListSeasonResponse,
+    ListSeasonsService,
+)
 
 router = APIRouter(
     prefix="/seasons",
@@ -24,7 +28,7 @@
             "model": NotFoundResponse,
         },
     },
-    response_model=Season,
+    response_model=GetSeasonResponse,
     name="season",
 )
 async def get_season(
@@ -34,30 +38,25 @@ async def get_season(
             le=INT32,
         ),
     ],
-    session: AsyncSession = Depends(get_async_session),  # noqa: B008
-) -> Season:
+) -> GetSeasonResponse:
     """Retrieve specific season.
 
     Utilize this endpoint to retrieve detailed information about a specific Futurama season by providing its unique ID.
     The response includes details such as the list of seasons, season ID, and more.
 
     Can be used to gain in-depth insights into a particular season of Futurama.
     """
-    try:
-        return await Season.get(session, season_id)
-    except ModelNotFoundError:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) from None
+    service: GetSeasonService = GetSeasonService(pk=season_id)
+    return await service()
 
 
 @router.get(
     "",
     status_code=status.HTTP_200_OK,
-    response_model=Page[Season],
+    response_model=Page[ListSeasonResponse],
     name="seasons",
 )
-async def get_seasons(
-    session: AsyncSession = Depends(get_async_session),  # noqa: B008
-) -> Page[Season]:
+async def get_seasons() -> Page[ListSeasonResponse]:
     """Retrieve specific seasons.
 
     Access a comprehensive list of all Futurama seasons using this endpoint,
@@ -67,4 +66,5 @@ async def get_seasons(
     This endpoint is valuable for those interested in exploring the entirety of Futurama's seasons or implementing
     features like season browsing on your site.
     """
-    return await Season.paginate(session)
+    service: ListSeasonsService = ListSeasonsService()
+    return await service()

futuramaapi/routers/services/__init__.py

@@ -0,0 +1,13 @@
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+from typing import Any
+
+from futuramaapi.helpers.pydantic import BaseModel
+
+
+class BaseService(BaseModel, ABC):
+    context: dict[str, Any] | None = None
+
+    @abstractmethod
+    async def __call__(self, *args, **kwargs) -> BaseModel | Sequence[BaseModel] | None:
+        pass

futuramaapi/routers/services/_base.py

@@ -0,0 +1,47 @@
+from abc import ABC
+
+from fastapi import HTTPException, status
+from pydantic import Field
+from sqlalchemy import Select, select
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import selectinload
+
+from futuramaapi.helpers.pydantic import BaseModel
+from futuramaapi.repositories.models import SeasonModel
+from futuramaapi.repositories.session import session_manager
+from futuramaapi.routers.services._base import BaseService
+
+
+class GetSeasonResponse(BaseModel):
+    class Episode(BaseModel):
+        id: int
+        name: str
+        broadcast_number: int = Field(alias="number")
+        production_code: str = Field(
+            examples=[
+                "1ACV01",
+            ],
+        )
+
+    id: int
+    episodes: list[Episode]
+
+
+class GetSeasonService(BaseService, ABC):
+    pk: int
+
+    @property
+    def statement(self) -> Select:
+        return select(SeasonModel).where(SeasonModel.id == self.pk).options(selectinload(SeasonModel.episodes))
+
+    async def __call__(self, *args, **kwargs) -> GetSeasonResponse:
+        async with session_manager.session() as session:
+            try:
+                season_model: SeasonModel = (await session.execute(self.statement)).scalars().one()
+            except NoResultFound:
+                raise HTTPException(
+                    detail="Season not found",
+                    status_code=status.HTTP_404_NOT_FOUND,
+                ) from None
+
+        return GetSeasonResponse.model_validate(season_model)

futuramaapi/routers/services/seasons/__init__.py

@@ -0,0 +1,28 @@
+from abc import ABC
+
+from fastapi_pagination import Page
+from fastapi_pagination.ext.sqlalchemy import paginate
+from sqlalchemy import Select, select
+from sqlalchemy.orm import selectinload
+
+from futuramaapi.repositories.models import SeasonModel
+from futuramaapi.repositories.session import session_manager
+from futuramaapi.routers.services._base import BaseService
+from futuramaapi.routers.services.seasons.get_season import GetSeasonResponse
+
+
+class ListSeasonResponse(GetSeasonResponse):
+    pass
+
+
+class ListSeasonsService(BaseService, ABC):
+    @property
+    def statement(self) -> Select:
+        return select(SeasonModel).filter().options(selectinload(SeasonModel.episodes))
+
+    async def __call__(self, *args, **kwargs) -> Page[ListSeasonResponse]:
+        async with session_manager.session() as session:
+            return await paginate(
+                session,
+                self.statement,
+            )