✨ feat: awesome jmy pagination

Author: Zerohertz

app/api/v1/endpoints/jmy.py

@@ -1,35 +1,55 @@
+from typing import Literal
+
 from dependency_injector.wiring import Provide, inject
-from fastapi import Depends, status
+from fastapi import Depends, Query, status
 from fastapi.responses import ORJSONResponse
 
 from app.core.auth import (
-    AdminAuthDeps,
     GitHubOAuthDeps,
     GoogleOAuthDeps,
     PasswordOAuthDeps,
 )
 from app.core.container import Container
 from app.core.router import CoreAPIRouter
-from app.schemas.jmy import JmyCompanyOut
+from app.schemas.jmy import JmyCompanyOut, JmyPageResponse
 from app.services.jmy import JmyService
 
-router = CoreAPIRouter(
-    prefix="/jmy",
-    tags=["jmy"],
-    dependencies=[AdminAuthDeps, PasswordOAuthDeps, GoogleOAuthDeps, GitHubOAuthDeps],
+router = CoreAPIRouter(prefix="/jmy", tags=["jmy"])
+
+
+@router.get(
+    "",
+    response_model=JmyPageResponse,
+    response_class=ORJSONResponse,
+    status_code=status.HTTP_200_OK,
+    dependencies=[PasswordOAuthDeps, GoogleOAuthDeps, GitHubOAuthDeps],
+    summary="",
+    description="",
 )
+@inject
+async def get_jmy_page(
+    page: int = Query(1, ge=1),
+    size: int = Query(10, ge=10, le=100),
+    sort_by: Literal["old", "new"] = "old",
+    filter_by: Literal["ms", "phd", "all"] = "all",
+    service: JmyService = Depends(Provide[Container.jmy_service]),
+):
+    return await service.get_page(
+        page=page, size=size, sort_by=sort_by, filter_by=filter_by
+    )
 
 
 @router.get(
     "/{id}",
     response_model=JmyCompanyOut,
     response_class=ORJSONResponse,
     status_code=status.HTTP_200_OK,
+    dependencies=[PasswordOAuthDeps, GoogleOAuthDeps, GitHubOAuthDeps],
     summary="",
     description="",
 )
 @inject
-async def get_jmy(
+async def get_jmy_by_id(
     id: int,
     service: JmyService = Depends(Provide[Container.jmy_service]),
 ):

app/repositories/jmy.py

@@ -1,7 +1,11 @@
-from sqlalchemy import select
+import math
+from operator import and_
+from typing import Literal, Sequence
+
+from sqlalchemy import func, select
 
 from app.core.database import database
-from app.models.jmy import JmyCompany
+from app.models.jmy import JmyCompany, JmyTimeSeries
 from app.repositories.base import BaseRepository
 
 
@@ -18,3 +22,65 @@ async def read_by_name(self, name: str, eager: bool = False) -> JmyCompany | Non
         result = await session.execute(stmt)
         entity = result.scalar_one_or_none()
         return entity
+
+    async def read_page(
+        self,
+        page: int,
+        size: int,
+        sort_by: Literal["old", "new"],
+        filter_by: Literal["ms", "phd", "all"],
+    ) -> tuple[Sequence[JmyCompany], int]:
+        jmy_ms = [
+            "벤처기업부설연구소",
+            "중견기업부설연구소",
+            "중소기업부설연구소",
+        ]
+        latest_date_subq = (
+            select(
+                JmyTimeSeries.company_id, func.max(JmyTimeSeries.date).label("max_date")
+            )
+            .group_by(JmyTimeSeries.company_id)
+            .subquery()
+        )
+        total_expr = (
+            (JmyTimeSeries.b_old + JmyTimeSeries.a_old)
+            if sort_by == "old"
+            else (JmyTimeSeries.b_new + JmyTimeSeries.a_new)
+        )
+        total_subq = (
+            select(
+                JmyTimeSeries.company_id,
+                total_expr.label("total"),
+            )
+            .join(
+                latest_date_subq,
+                and_(
+                    JmyTimeSeries.company_id == latest_date_subq.c.company_id,
+                    JmyTimeSeries.date == latest_date_subq.c.max_date,
+                ),
+            )
+            .subquery()
+        )
+        stmt = select(self.model).outerjoin(
+            total_subq, self.model.id == total_subq.c.company_id
+        )
+        if filter_by == "ms":
+            stmt = stmt.where(self.model.type_.in_(jmy_ms))
+        elif filter_by == "phd":
+            stmt = stmt.where(self.model.type_.notin_(jmy_ms))
+        offset = (page - 1) * size
+        stmt = stmt.order_by(total_subq.c.total.desc()).offset(offset).limit(size)
+        session = database.scoped_session()
+        results = await session.execute(stmt)
+        entities = results.scalars().all()
+        _stmt = select(func.count(self.model.id)).outerjoin(
+            total_subq, self.model.id == total_subq.c.company_id
+        )
+        if filter_by == "ms":
+            _stmt = _stmt.where(self.model.type_.in_(jmy_ms))
+        elif filter_by == "phd":
+            _stmt = _stmt.where(self.model.type_.notin_(jmy_ms))
+        result = await session.execute(_stmt)
+        total_count = result.scalar_one()
+        total_pages = math.ceil(total_count / size)
+        return entities, total_pages

app/schemas/auth.py

@@ -5,13 +5,13 @@
 from pydantic import BaseModel
 
 from app.models.enums import OAuthProvider
-from app.schemas.base import BaseRequest, BaseResponse
+from app.schemas.base import BaseDBResponse, BaseRequest
 
 
 class AuthRequest(BaseRequest): ...
 
 
-class AuthResponse(BaseResponse):
+class AuthResponse(BaseDBResponse):
     provider: OAuthProvider
 
 

app/schemas/base.py

@@ -14,6 +14,9 @@ class BaseRequest(BaseModel, abc.ABC):
 
 class BaseResponse(BaseModel, abc.ABC):
     model_config = ConfigDict(from_attributes=True)
+
+
+class BaseDBResponse(BaseResponse):
     id: int
     created_at: datetime
     updated_at: datetime

app/schemas/jmy.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from app.schemas.base import BaseRequest, BaseResponse
+from app.schemas.base import BaseDBResponse, BaseRequest, BaseResponse
 
 
 class JmyCompanyRequest(BaseRequest):
@@ -20,7 +20,7 @@ class JmyCompanyRequest(BaseRequest):
     a_old: int
 
 
-class JmyCompanyResponse(BaseResponse):
+class JmyCompanyResponse(BaseDBResponse):
     name: str
     year: int
     location: str
@@ -30,7 +30,22 @@ class JmyCompanyResponse(BaseResponse):
     research: str
 
 
-class JmyTimeSeriesOut(BaseResponse):
+class JmyCompanyInfo(JmyCompanyResponse):
+    date: datetime
+    b_assigned: int
+    b_new: int
+    b_old: int
+    a_assigned: int
+    a_new: int
+    a_old: int
+
+
+class JmyPageResponse(BaseResponse):
+    info: list[JmyCompanyInfo]
+    total_pages: int
+
+
+class JmyTimeSeriesOut(BaseDBResponse):
     date: datetime
     b_assigned: int
     b_new: int

app/schemas/users.py

@@ -4,7 +4,7 @@
 
 from app.models.enums import Role
 from app.schemas.auth import AuthOut, AuthResponse
-from app.schemas.base import BaseRequest, BaseResponse
+from app.schemas.base import BaseDBResponse, BaseRequest
 
 
 class UserRequest(BaseRequest):
@@ -20,7 +20,7 @@ class UserPasswordAdminRequest(BaseRequest):
     password: Annotated[str, StringConstraints(min_length=8, max_length=30)]
 
 
-class UserResponse(BaseResponse):
+class UserResponse(BaseDBResponse):
     name: Annotated[str, StringConstraints(min_length=3, max_length=30)]
     email: EmailStr
     role: Role

app/services/jmy.py

@@ -1,7 +1,14 @@
+from typing import Literal
+
 from app.core.database import database
 from app.models.jmy import JmyCompany, JmyTimeSeries
 from app.repositories.jmy import JmyRepository
-from app.schemas.jmy import JmyCompanyOut, JmyCompanyRequest
+from app.schemas.jmy import (
+    JmyCompanyInfo,
+    JmyCompanyOut,
+    JmyCompanyRequest,
+    JmyPageResponse,
+)
 from app.services.base import BaseService
 
 
@@ -36,3 +43,40 @@ async def create(self, schema: JmyCompanyRequest) -> JmyCompanyOut:
         )
         entity = await self.repository.create(entity=jmy_company)
         return self.mapper(entity)
+
+    @database.transactional
+    async def get_page(
+        self,
+        page: int,
+        size: int,
+        sort_by: Literal["old", "new"],
+        filter_by: Literal["ms", "phd", "all"],
+    ) -> JmyPageResponse:
+        entities, total_pages = await self.repository.read_page(
+            page=page, size=size, sort_by=sort_by, filter_by=filter_by
+        )
+        jmy_company_info = []
+        for entity in entities:
+            time_series = entity.time_series[-1]
+            jmy_company_info.append(
+                JmyCompanyInfo(
+                    id=entity.id,
+                    created_at=entity.created_at,
+                    updated_at=entity.updated_at,
+                    name=entity.name,
+                    year=entity.year,
+                    location=entity.location,
+                    address=entity.address,
+                    type_=entity.type_,
+                    size=entity.size,
+                    research=entity.research,
+                    date=time_series.date,
+                    b_assigned=time_series.b_assigned,
+                    b_new=time_series.b_new,
+                    b_old=time_series.b_old,
+                    a_assigned=time_series.a_assigned,
+                    a_new=time_series.a_new,
+                    a_old=time_series.a_old,
+                )
+            )
+        return JmyPageResponse(info=jmy_company_info, total_pages=total_pages)

pyproject.toml

@@ -59,6 +59,7 @@ disable = [
     "R0801",
     "R0903",
     "R0913",
+    "R0914",
     "W0221",
     "W0511",
     "W0622",
@@ -75,6 +76,7 @@ disable = [
 # R0801: Similar lines in * files
 # R0903: Too few public methods (*/*) (too-few-public-methods)
 # R0913: Too many arguments (*/*) (too-many-arguments)
+# R0914: Too many local variables (*/*) (too-many-locals)
 # W0221: Number of parameters was * in '*' and is now * in overriding '*' method (arguments-differ)
 # W0221: Variadics removed in overriding '*' method (arguments-differ)
 # W0511: TODO, FIXME