feat: recruitment periods + dynamic form system

Phase 2 of auth-and-beginner-flow plan:

- Recruitment periods: CRUD with active period by type
- Dynamic forms: admin-editable questions (short_text, long_text,
  radio, checkbox), linked to recruitment periods
- Form questions: create/update/delete with sort order
- Migration 004: recruitment_periods, forms, form_questions tables
- GraphQL: periods, form queries + createPeriod, updatePeriod,
  createForm, updateForm mutations

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ImTotem

alembic/versions/004_recruit_and_forms.py

@@ -0,0 +1,55 @@
+"""recruitment periods and forms
+
+Revision ID: 004
+Revises: 003
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "004"
+down_revision = "003"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "recruitment_periods",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("title", sa.String, nullable=False),
+        sa.Column("type", sa.String, nullable=False),
+        sa.Column("start_date", sa.String, nullable=False),
+        sa.Column("end_date", sa.String, nullable=False),
+        sa.Column("is_active", sa.String, server_default="true"),
+        sa.Column("created_by", sa.String, sa.ForeignKey("members.id")),
+        sa.Column("created_at", sa.String),
+        sa.Column("updated_at", sa.String),
+    )
+    op.create_table(
+        "forms",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("title", sa.String, nullable=False),
+        sa.Column("description", sa.Text),
+        sa.Column("recruitment_id", sa.String, sa.ForeignKey("recruitment_periods.id")),
+        sa.Column("type", sa.String, nullable=False),
+        sa.Column("is_active", sa.String, server_default="true"),
+        sa.Column("created_by", sa.String, sa.ForeignKey("members.id")),
+        sa.Column("created_at", sa.String),
+        sa.Column("updated_at", sa.String),
+    )
+    op.create_table(
+        "form_questions",
+        sa.Column("id", sa.String, primary_key=True),
+        sa.Column("form_id", sa.String, sa.ForeignKey("forms.id", ondelete="CASCADE")),
+        sa.Column("label", sa.String, nullable=False),
+        sa.Column("type", sa.String, nullable=False),
+        sa.Column("options", sa.Text),
+        sa.Column("required", sa.String, server_default="true"),
+        sa.Column("sort_order", sa.Integer, nullable=False),
+        sa.Column("created_at", sa.String),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("form_questions")
+    op.drop_table("forms")
+    op.drop_table("recruitment_periods")

src/bcsd_api/dependencies.py

@@ -11,7 +11,9 @@
 from .email import ResendSender
 from .email.sender import EmailSender
 from .exception import Unauthorized
+from .form.pg_repository import PgFormRepository, PgQuestionRepository
 from .member.pg_repository import PgMemberRepository
+from .recruit.pg_repository import PgRecruitRepository
 from .setting.pg_repository import PgSettingRepository
 from .sheets.client import SheetsClient
 from .shorten.pg_repository import PgLinkRepository
@@ -72,6 +74,18 @@ def get_setting_repo(conn: Connection = Depends(get_conn)) -> PgSettingRepositor
     return PgSettingRepository(conn)
 
 
+def get_recruit_repo(conn: Connection = Depends(get_conn)) -> PgRecruitRepository:
+    return PgRecruitRepository(conn)
+
+
+def get_form_repo(conn: Connection = Depends(get_conn)) -> PgFormRepository:
+    return PgFormRepository(conn)
+
+
+def get_question_repo(conn: Connection = Depends(get_conn)) -> PgQuestionRepository:
+    return PgQuestionRepository(conn)
+
+
 def current_user(
     request: Request,
     settings: Settings = Depends(get_settings),

src/bcsd_api/form/__init__.py

@@ -0,0 +1,42 @@
+from sqlalchemy import Connection, delete, insert, select, update
+
+from bcsd_api.repository import BaseRepository
+from bcsd_api.tables import form_questions, forms
+
+
+class PgFormRepository(BaseRepository):
+    def __init__(self, conn: Connection):
+        super().__init__(conn, forms)
+
+    def find_by_recruitment(self, recruitment_id: str) -> list[dict]:
+        stmt = select(forms).where(forms.c.recruitment_id == recruitment_id)
+        return [row._asdict() for row in self._conn.execute(stmt)]
+
+    def create(self, row: dict) -> None:
+        self._conn.execute(insert(forms).values(**row))
+
+    def update_fields(self, form_id: str, updates: dict) -> None:
+        self._conn.execute(
+            update(forms).where(forms.c.id == form_id).values(**updates),
+        )
+
+
+class PgQuestionRepository(BaseRepository):
+    def __init__(self, conn: Connection):
+        super().__init__(conn, form_questions)
+
+    def find_by_form(self, form_id: str) -> list[dict]:
+        stmt = (
+            select(form_questions)
+            .where(form_questions.c.form_id == form_id)
+            .order_by(form_questions.c.sort_order)
+        )
+        return [row._asdict() for row in self._conn.execute(stmt)]
+
+    def create(self, row: dict) -> None:
+        self._conn.execute(insert(form_questions).values(**row))
+
+    def delete_by_form(self, form_id: str) -> None:
+        self._conn.execute(
+            delete(form_questions).where(form_questions.c.form_id == form_id),
+        )

src/bcsd_api/form/pg_repository.py

@@ -0,0 +1,72 @@
+import strawberry
+from strawberry.types import Info
+
+from bcsd_api.graphql.context import GqlContext, require_user
+from bcsd_api.graphql.convert import from_model
+
+from . import service
+from .schema import CreateFormRequest, QuestionRequest, UpdateFormRequest
+from .types import CreateFormInput, FormType, QuestionType, UpdateFormInput
+
+
+def _to_questions(inputs: list) -> list[QuestionRequest]:
+    return [
+        QuestionRequest(
+            label=q.label, type=q.type,
+            options=q.options, required=q.required,
+            sort_order=q.sort_order,
+        )
+        for q in inputs
+    ]
+
+
+def _to_form_type(f) -> FormType:
+    data = f.model_dump()
+    data["questions"] = [QuestionType(**q) for q in data["questions"]]
+    return FormType(**data)
+
+
+def resolve_form(info: Info[GqlContext, None], id: strawberry.ID) -> FormType:
+    require_user(info.context)
+    ctx = info.context
+    f = service.get_form(ctx.form_repo, ctx.question_repo, id)
+    return _to_form_type(f)
+
+
+def resolve_forms(
+    info: Info[GqlContext, None], recruitment_id: str | None = None,
+) -> list[FormType]:
+    require_user(info.context)
+    ctx = info.context
+    forms = service.list_forms(ctx.form_repo, ctx.question_repo, recruitment_id)
+    return [_to_form_type(f) for f in forms]
+
+
+def resolve_create_form(
+    info: Info[GqlContext, None], input: CreateFormInput,
+) -> FormType:
+    user = require_user(info.context)
+    ctx = info.context
+    req = CreateFormRequest(
+        title=input.title, description=input.description,
+        recruitment_id=input.recruitment_id, type=input.type,
+        questions=_to_questions(input.questions),
+    )
+    f = service.create_form(ctx.form_repo, ctx.question_repo, req, user["sub"])
+    return _to_form_type(f)
+
+
+def resolve_update_form(
+    info: Info[GqlContext, None], id: strawberry.ID, input: UpdateFormInput,
+) -> FormType:
+    require_user(info.context)
+    ctx = info.context
+    questions = None
+    if input.questions is not None:
+        questions = _to_questions(input.questions)
+    req = UpdateFormRequest(
+        title=input.title, description=input.description,
+        is_active=input.is_active, questions=questions,
+    )
+    f = service.update_form(ctx.form_repo, ctx.question_repo, id, req)
+    return _to_form_type(f)

src/bcsd_api/form/resolvers.py

@@ -0,0 +1,48 @@
+from pydantic import BaseModel
+
+
+class QuestionRequest(BaseModel):
+    label: str
+    type: str
+    options: str | None = None
+    required: str = "true"
+    sort_order: int = 0
+
+
+class QuestionResponse(BaseModel):
+    id: str
+    form_id: str
+    label: str
+    type: str
+    options: str | None = None
+    required: str = "true"
+    sort_order: int = 0
+    created_at: str | None = None
+
+
+class CreateFormRequest(BaseModel):
+    title: str
+    description: str | None = None
+    recruitment_id: str
+    type: str
+    questions: list[QuestionRequest] = []
+
+
+class UpdateFormRequest(BaseModel):
+    title: str | None = None
+    description: str | None = None
+    is_active: str | None = None
+    questions: list[QuestionRequest] | None = None
+
+
+class FormResponse(BaseModel):
+    id: str
+    title: str
+    description: str | None = None
+    recruitment_id: str
+    type: str
+    is_active: str
+    created_by: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+    questions: list[QuestionResponse] = []

src/bcsd_api/form/schema.py

@@ -0,0 +1,110 @@
+from datetime import datetime
+
+from bcsd_api.exception import NotFound
+from bcsd_api.id_gen import generate_id
+from bcsd_api.timezone import KST
+
+from .pg_repository import PgFormRepository, PgQuestionRepository
+from .schema import (
+    CreateFormRequest,
+    FormResponse,
+    QuestionRequest,
+    QuestionResponse,
+    UpdateFormRequest,
+)
+
+
+def _now() -> str:
+    return datetime.now(KST).isoformat()
+
+
+def create_form(
+    form_repo: PgFormRepository,
+    q_repo: PgQuestionRepository,
+    req: CreateFormRequest,
+    creator_id: str,
+) -> FormResponse:
+    now = _now()
+    form_id = generate_id("F")
+    row = {
+        "id": form_id, "title": req.title,
+        "description": req.description or "",
+        "recruitment_id": req.recruitment_id,
+        "type": req.type, "is_active": "true",
+        "created_by": creator_id,
+        "created_at": now, "updated_at": now,
+    }
+    form_repo.create(row)
+    questions = _save_questions(q_repo, form_id, req.questions)
+    return FormResponse(**row, questions=questions)
+
+
+def update_form(
+    form_repo: PgFormRepository,
+    q_repo: PgQuestionRepository,
+    form_id: str,
+    req: UpdateFormRequest,
+) -> FormResponse:
+    _get_or_raise(form_repo, form_id)
+    updates = {}
+    if req.title is not None:
+        updates["title"] = req.title
+    if req.description is not None:
+        updates["description"] = req.description
+    if req.is_active is not None:
+        updates["is_active"] = req.is_active
+    updates["updated_at"] = _now()
+    form_repo.update_fields(form_id, updates)
+    if req.questions is not None:
+        q_repo.delete_by_form(form_id)
+        _save_questions(q_repo, form_id, req.questions)
+    return get_form(form_repo, q_repo, form_id)
+
+
+def get_form(
+    form_repo: PgFormRepository, q_repo: PgQuestionRepository, form_id: str,
+) -> FormResponse:
+    row = _get_or_raise(form_repo, form_id)
+    questions = [QuestionResponse(**q) for q in q_repo.find_by_form(form_id)]
+    return FormResponse(**row, questions=questions)
+
+
+def list_forms(
+    form_repo: PgFormRepository,
+    q_repo: PgQuestionRepository,
+    recruitment_id: str | None = None,
+) -> list[FormResponse]:
+    if recruitment_id:
+        rows = form_repo.find_by_recruitment(recruitment_id)
+    else:
+        rows = form_repo.find_all()
+    result = []
+    for row in rows:
+        qs = [QuestionResponse(**q) for q in q_repo.find_by_form(row["id"])]
+        result.append(FormResponse(**row, questions=qs))
+    return result
+
+
+def _get_or_raise(form_repo: PgFormRepository, form_id: str) -> dict:
+    row = form_repo.find_by_id(form_id)
+    if not row:
+        raise NotFound(f"form {form_id} not found")
+    return row
+
+
+def _save_questions(
+    q_repo: PgQuestionRepository, form_id: str, questions: list[QuestionRequest],
+) -> list[QuestionResponse]:
+    result = []
+    for i, q in enumerate(questions):
+        row = {
+            "id": generate_id("FQ"), "form_id": form_id,
+            "label": q.label, "type": q.type,
+            "options": q.options or "",
+            "required": q.required,
+            "sort_order": q.sort_order or i,
+            "created_at": _now(),
+        }
+        q_repo.create(row)
+        result.append(QuestionResponse(**row))
+    return result