Feature/grad 2025 view

app/features/flags.toml

@@ -3,12 +3,15 @@
 
 [default]
 "grad_2025.enabled" = true
+"grad_2025_viewer.enabled" = false
 
 [environments.local]
 "grad_2025.enabled" = true
+"grad_2025_viewer.enabled" = true
 
 [environments.dev]
 "grad_2025.enabled" = true
+"grad_2025_viewer.enabled" = true
 
 [environments.prod]
 # production은 기본값 사용

app/grad_2025/router.py

@@ -1,6 +1,13 @@
+from typing import Annotated
+
+from fastapi import Query
+
+from app.category.enums import Category
+from app.common.schemas import Page
 from app.core.router import create_router
+from app.posts.schemas import PostCompactRead
 
-from .schemas import Grad2025Read
+from .schemas import Grad2025Cursor, Grad2025Read
 from .service import Grad2025Service
 
 router = create_router()
@@ -11,3 +18,21 @@ async def read_univ_majors(
     grad_2025_service: Grad2025Service,
 ) -> list[Grad2025Read]:
     return await grad_2025_service.get_univ_majors()
+
+
+@router.get("/posts")
+async def search_grad_2025_posts(
+    grad_2025_service: Grad2025Service,
+    cursor: Annotated[
+        str | None,
+        Query(description="Base64 인코딩된 커서 (응답의 next_cursor 값)"),
+    ] = None,
+    category: Category | None = Query(None),
+    univ_major: str | None = Query(None),
+) -> Page[PostCompactRead]:
+    decoded_cursor = Grad2025Cursor.decode(cursor)
+    return await grad_2025_service.search_posts(
+        cursor=decoded_cursor,
+        category=category,
+        univ_major=univ_major,
+    )

app/grad_2025/schemas.py

@@ -1,3 +1,5 @@
+import base64
+
 from app.schemas import APISchema
 
 from .models import Grad2025
@@ -11,3 +13,18 @@ def from_grad_2025(cls, grad_2025: Grad2025):
         return cls(
             name=grad_2025.name,
         )
+
+
+class Grad2025Cursor(APISchema):
+    id: int
+
+    def encode(self) -> str:
+        payload = self.model_dump_json().encode()
+        return base64.urlsafe_b64encode(payload).decode()
+
+    @classmethod
+    def decode(cls, token: str | None) -> "Grad2025Cursor | None":
+        if not token:
+            return None
+        data = base64.urlsafe_b64decode(token.encode()).decode()
+        return cls.model_validate_json(data)

app/grad_2025/service.py

@@ -1,13 +1,44 @@
+from app.category.enums import Category
+from app.common.schemas import Page
+from app.posts.repository import PostRepository
+from app.posts.schemas import PostCompactRead
 from app.utils.dependency import dependency
 
 from .repository import Grad2025Repository
-from .schemas import Grad2025Read
+from .schemas import Grad2025Cursor, Grad2025Read
+
+POST_CURSOR_LIMIT = 10
 
 
 @dependency
 class Grad2025Service:
     grad_2025_repository: Grad2025Repository
+    post_repository: PostRepository
 
     async def get_univ_majors(self) -> list[Grad2025Read]:
         univ_majors = await self.grad_2025_repository.list_univ_majors()
         return [Grad2025Read.from_grad_2025(univ_major) for univ_major in univ_majors]
+
+    async def search_posts(
+        self,
+        *,
+        cursor: Grad2025Cursor | None,
+        limit: int = POST_CURSOR_LIMIT,
+        category: Category | None = None,
+        univ_major: str | None = None,
+    ) -> Page[PostCompactRead]:
+        cursor_id = cursor.id if cursor else None
+        posts = await self.post_repository.find_grad_2025_posts(
+            cursor=cursor_id,
+            limit=limit,
+            category=category,
+            univ_major=univ_major,
+        )
+        compact_posts = [PostCompactRead.from_post(post) for post in posts]
+        next_cursor = (
+            Grad2025Cursor(id=posts[-1].id).encode() if len(posts) == limit else None
+        )
+        return Page[PostCompactRead](
+            list=compact_posts,
+            next_cursor=next_cursor,
+        )

app/posts/repository.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy import String, and_, case, delete, exists, func, or_, select, update
 from sqlalchemy.dialects.postgresql import insert
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, selectinload
 
 from app.category.enums import Category
 from app.common.schemas import FeedCursor
@@ -281,3 +281,47 @@ async def find_univ_major_by_post_id(self, *, post_id: int) -> Grad2025 | None:
             .join(post_grad_2025_table)
             .where(post_grad_2025_table.c.post_id == post_id)
         )
+
+    async def find_grad_2025_posts(
+        self,
+        *,
+        cursor: int | None,
+        limit: int,
+        category: Category | None = None,
+        univ_major: str | None = None,
+    ) -> list[Post]:
+        """post_grad_2025_table에 존재하는 Post만 조회합니다."""
+        stmt = (
+            select(Post)
+            .options(
+                joinedload(Post.author),  # 1:1 관계 - joinedload OK
+                selectinload(Post.images),  # 1:N 관계 - selectinload로 별도 쿼리
+            )
+            .join(post_grad_2025_table, Post.id == post_grad_2025_table.c.post_id)
+            .where(Post.deleted_at.is_(None))
+            .order_by(Post.created_at.desc(), Post.id.desc())
+            .limit(limit)
+        )
+
+        if cursor is not None:
+            # 커서 id의 created_at을 가져와서 그 이전 것들만 조회
+            cursor_created_at = (
+                select(Post.created_at).where(Post.id == cursor).scalar_subquery()
+            )
+            stmt = stmt.where(
+                or_(
+                    Post.created_at < cursor_created_at,
+                    and_(Post.created_at == cursor_created_at, Post.id < cursor),
+                )
+            )
+
+        if category is not None:
+            stmt = stmt.where(Post.category == category)
+
+        if univ_major is not None:
+            stmt = stmt.join(
+                Grad2025, post_grad_2025_table.c.grad_2025_id == Grad2025.id
+            ).where(Grad2025.name == univ_major)
+
+        result = await self.session.scalars(stmt)
+        return list(result.unique().all())