feat: post public_id를 8자 base62로 변경

app/admin/router.py

@@ -4,12 +4,13 @@
 from fastapi import Depends, HTTPException, Query
 from fastapi.security import APIKeyHeader
 from sqlalchemy import func, select, text
-from ulid import ULID
+from sqlalchemy.exc import IntegrityError
 
 from app.core.router import create_router
 from app.database.deps import SessionDep
 from app.posts.models import Post
 from app.schemas import APISchema
+from app.utils.public_id import generate_public_id
 
 from .config import admin_settings
 
@@ -46,13 +47,11 @@ async def fill_public_ids(
     batch_size: int = Query(default=100, ge=1, le=1000),
 ) -> FillPublicIdsResponse:
     """
-    public_id가 NULL인 Post에 ULID를 채웁니다.
-    created_at 기반으로 ULID를 생성합니다.
+    public_id가 NULL인 Post에 8자 public_id를 채웁니다.
 
     - batch_size: 한 번에 처리할 개수 (기본값: 100, 최대: 1000)
     - 여러 번 호출해서 점진적으로 처리할 수 있습니다.
     """
-    # NULL인 Post 개수 확인
     total_remaining = await session.scalar(
         select(func.count()).select_from(Post).where(Post.public_id.is_(None))
     )
@@ -65,33 +64,42 @@ async def fill_public_ids(
             remaining=0,
         )
 
-    # 배치 처리: SELECT로 id, created_at 조회
-    posts = await session.execute(
-        select(Post.id, Post.created_at)
-        .where(Post.public_id.is_(None))
-        .order_by(Post.id)
-        .limit(batch_size)
-    )
-    posts_list = list(posts)
+    post_ids = (
+        await session.scalars(
+            select(Post.id)
+            .where(Post.public_id.is_(None))
+            .order_by(Post.id)
+            .limit(batch_size)
+        )
+    ).all()
 
-    if not posts_list:
+    if not post_ids:
         return FillPublicIdsResponse(
             success=True,
             message="모든 Post에 public_id가 이미 있습니다.",
             processed=0,
             remaining=0,
         )
 
-    # Raw SQL로 업데이트 (onupdate 트리거 우회)
-    for post in posts_list:
-        ulid_value = str(ULID.from_datetime(post.created_at))
-        await session.execute(
-            text("UPDATE post SET public_id = :ulid WHERE id = :id"),
-            {"ulid": ulid_value, "id": post.id},
-        )
+    for post_id in post_ids:
+        for _ in range(5):
+            try:
+                async with session.begin_nested():
+                    await session.execute(
+                        text("UPDATE post SET public_id = :pid WHERE id = :id"),
+                        {"pid": generate_public_id(), "id": post_id},
+                    )
+                break
+            except IntegrityError:
+                continue
+        else:
+            raise HTTPException(
+                status_code=500,
+                detail=f"public_id 충돌 재시도 한도 초과 (post id={post_id})",
+            )
     await session.commit()
 
-    processed = len(posts_list)
+    processed = len(post_ids)
     remaining = total_remaining - processed
 
     return FillPublicIdsResponse(

app/posts/models.py

@@ -18,7 +18,7 @@
 
 from app.category.enums import Category
 from app.models import Base
-from app.utils.ulid import generate_ulid
+from app.utils.public_id import generate_public_id
 
 if TYPE_CHECKING:
     from app.users.models import User
@@ -53,12 +53,12 @@ class Post(Base):
     )
 
     public_id: Mapped[str | None] = mapped_column(
-        String(26),
+        String(8),
         unique=True,
         index=True,
         nullable=True,
         init=False,
-        default_factory=generate_ulid,
+        default_factory=generate_public_id,
     )
 
     author_id: Mapped[int] = mapped_column(ForeignKey("user.id"))

app/posts/repository.py

@@ -2,6 +2,7 @@
 
 from sqlalchemy import String, and_, case, delete, exists, func, or_, select, update
 from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload, selectinload
 
 from app.category.enums import Category
@@ -10,6 +11,7 @@
 from app.grad_2025.models import Grad2025
 from app.users.models import User
 from app.utils.dependency import dependency
+from app.utils.public_id import generate_public_id
 
 from .models import Post, PostImage, post_grad_2025_table
 
@@ -147,7 +149,15 @@ async def feed(
 
     async def save(self, *, post: Post, univ_major: str | None = None) -> SaveResult:
         self.session.add(post)
-        await self.session.flush()
+        for _ in range(5):
+            try:
+                async with self.session.begin_nested():
+                    await self.session.flush()
+                break
+            except IntegrityError:
+                post.public_id = generate_public_id()
+        else:
+            raise RuntimeError("public_id 충돌 재시도 한도 초과")
 
         new_univ_major: str | None = None
 

app/utils/public_id.py

@@ -0,0 +1,9 @@
+import secrets
+import string
+
+PUBLIC_ID_ALPHABET = string.ascii_letters + string.digits
+PUBLIC_ID_LENGTH = 8
+
+
+def generate_public_id() -> str:
+    return "".join(secrets.choice(PUBLIC_ID_ALPHABET) for _ in range(PUBLIC_ID_LENGTH))

migrations/versions/2026-05-10_change_post_public_id_to_8char.py

@@ -0,0 +1,77 @@
+"""change post public_id to 8-char base62
+
+Revision ID: dd329e7814b1
+Revises: 40fea8c77b2e
+Create Date: 2026-05-10 00:00:00.000000
+
+"""
+
+import secrets
+import string
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "dd329e7814b1"
+down_revision: Union[str, None] = "40fea8c77b2e"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+_PUBLIC_ID_ALPHABET = string.ascii_letters + string.digits
+_PUBLIC_ID_LENGTH = 8
+
+
+def _generate_public_id() -> str:
+    return "".join(
+        secrets.choice(_PUBLIC_ID_ALPHABET) for _ in range(_PUBLIC_ID_LENGTH)
+    )
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+
+    op.drop_index(op.f("ix_post_public_id"), table_name="post")
+
+    rows = bind.execute(sa.text("SELECT id FROM post")).fetchall()
+
+    used: set[str] = set()
+    updates: list[dict[str, object]] = []
+    for row in rows:
+        while True:
+            candidate = _generate_public_id()
+            if candidate not in used:
+                used.add(candidate)
+                break
+        updates.append({"id": row.id, "pid": candidate})
+
+    if updates:
+        bind.execute(
+            sa.text("UPDATE post SET public_id = :pid WHERE id = :id"),
+            updates,
+        )
+
+    op.alter_column(
+        "post",
+        "public_id",
+        existing_type=sa.String(length=26),
+        type_=sa.String(length=8),
+        existing_nullable=True,
+    )
+
+    op.create_index(op.f("ix_post_public_id"), "post", ["public_id"], unique=True)
+
+
+def downgrade() -> None:
+    # 기존 ULID 값은 복원하지 않습니다. 컬럼 길이만 되돌립니다.
+    op.drop_index(op.f("ix_post_public_id"), table_name="post")
+    op.alter_column(
+        "post",
+        "public_id",
+        existing_type=sa.String(length=8),
+        type_=sa.String(length=26),
+        existing_nullable=True,
+    )
+    op.create_index(op.f("ix_post_public_id"), "post", ["public_id"], unique=True)

pyproject.toml

@@ -17,7 +17,6 @@ dependencies = [
     "psycopg[binary,pool]>=3.2.6",
     "pydantic-settings>=2.8.1",
     "pyjwt[crypto]>=2.10.1",
-    "python-ulid>=3.1.0",
     "sqlalchemy[asyncio]>=2.0.39",
     "types-aioboto3[full]>=14.1.0",
 ]