Schemas and repositories

สิ่งที่จะสร้าง

สองชั้นที่ตั้งบน model ของบทที่แล้ว: app/schemas/ — Pydantic v2 model ที่ define ว่าข้อมูลอะไรข้าม boundary ของ API — และ app/repositories/ — async class ที่เป็นเจ้าของทุก database query schema validate และจัดรูปร่าง; repository อ่านและเขียน สองชั้นนี้รวมกันเปลี่ยน raw SQLAlchemy model เป็น read/write surface ที่สะอาด ให้ router แค่เรียกใช้

คุณจะเขียน ExerciseCreate และ ExerciseRead, WorkoutCreate ที่มี list SetInput ซ้อนและ WorkoutRead ที่ match กัน แล้ว ExerciseRepo และ WorkoutRepo — แต่ละตัวรับ AsyncSession (จาก Async database →) และ expose method อย่าง create, get, และ list นี่คือชิ้นสุดท้ายของ domain layer: เมื่อชั้นนี้เสร็จ The Exercises API → ส่วนใหญ่คือการต่อสาย route เข้ากับการเรียก repository

ทำไม

SQLAlchemy model คือรูปร่างของ database จึงไม่ควรรับตรง ๆ จาก client และไม่ควรยื่นกลับไปตรง ๆ การรับ raw model เป็น input จะเปิดให้ผู้เรียกตั้ง id, created_at, หรือ created_by — field ที่ server เป็นเจ้าของ ส่วนการ return model ตรง ๆ ก็ leak internal structure และกับดัก lazy-load ดังนั้น FitTrack แยก concern ตรงนี้ออก: Pydantic schema คือ boundary และมาเป็นคู่เสมอ

• ExerciseCreate คือสิ่งที่ client ส่ง ได้ — name, muscle_group, is_public — และไม่มีอย่างอื่น ตัว schema validate ให้: name ต้องไม่ว่าง และจำกัดความยาว ไม่มี id ไม่มี created_by เพราะ server เป็นคนตั้งค่าพวกนั้น
• ExerciseRead คือสิ่งที่ API return — row เต็มรวม field ที่ server เป็นเจ้าของ — พร้อม model_config = ConfigDict(from_attributes=True) จึงสร้างจาก SQLAlchemy object ตรง ๆ ได้

WorkoutCreate แสดงว่าทำไม nesting ถึงสำคัญ workout ถูก log เป็น payload เดียวพร้อม set อยู่ข้างใน — {"notes": ..., "sets": [{...}, {...}]} — เพราะ session และ set ของ session นั้นสร้างขึ้นด้วยกันในหนึ่ง transaction ดังนั้น WorkoutCreate ถือ list[SetInput] และ Pydantic validate ทั้ง tree ในทีเดียว: อย่างน้อยหนึ่ง set แต่ละตัวมี reps ที่เป็นบวกและ weight_kg ที่ fit numeric(6,2) สังเกตว่า SetInput ไม่มี set_index — client ส่ง set ตามลำดับ และ server assign index ตามตำแหน่ง ลำดับเป็น truth ที่ server เป็นเจ้าของ ไม่ใช่สิ่งที่ควร trust ให้ client เป็นคนนับ

Repository ตอบคำถามคนละอัน: query อยู่ที่ไหน? ไม่ใช่ใน router — router ควรอ่านเหมือน “validate input, ทำสิ่งนั้น, return output,” ไม่ใช่แบก noise ของ select(...).where(...) repository เป็น class เล็ก ๆ ที่ถือ AsyncSession เดียวและทุก query สำหรับ resource หนึ่ง ExerciseRepo.list_visible(user_id) encapsulate “exercise สาธารณะบวกของ user คนนี้เอง”; WorkoutRepo.get(id, user_id) encapsulate “workout นี้ แต่เฉพาะถ้าเป็นของคุณ พร้อม set ที่ load แล้ว” router เรียก method; SQL มีบ้านเดียวเป๊ะ; test exercise repo ได้โดยไม่ต้องยก HTTP ขึ้นมา

กฎที่เก็บให้ write ทนทาน: write method แต่ละตัว commit unit of work ของตัวเอง get_session ของ Async database → ตั้งใจ ไม่ commit — แค่ yield session สะอาดแล้วปิดให้ — ดังนั้น method ที่เขียนต้องเรียก commit() เอง ส่วน read method ไม่ต้องเลย สิ่งที่ทำให้ “สร้าง workout และ set ทั้งชุด” atomic ไม่ใช่ shared request-level transaction แต่คือการ add ทั้ง object graph แล้ว commit ใน commit() เดียว ภายใน create: SQLAlchemy flush parent และ cascaded children ด้วยกัน และ commit เดียวนั้นทำให้ทั้งหมดทนทานหรือไม่มีเลย การ commit ต่อ set หรือใน 2 ขั้น คือสิ่งที่จะให้ session ที่ log ไปครึ่งเดียวมีอยู่ได้

ข้อดีข้อเสีย

แยก input และ output schema (Create เทียบกับ Read) เทียบกับ schema เดียวสำหรับทั้งสองทิศทาง

• Pros: input schema expose เฉพาะ field ที่ client ตั้งได้ ดังนั้นค่าที่ server เป็นเจ้าของ (id, created_by, timestamp) spoof ไม่ได้และไม่ต้อง strip; กฎ validation อยู่เป๊ะที่ที่ข้อมูลที่ไม่ trust เข้ามา; output schema รวมทุกอย่างที่ client ควรเห็นได้อย่างปลอดภัย รูปร่างทั้งสอง evolve อิสระกัน
• Cons: class เยอะขึ้นสำหรับสิ่งที่ดูเหมือน entity เดียวกัน และมีการซ้ำเล็กน้อยระหว่าง Create และ Read การซ้ำนั้นคือประเด็น — รูปร่างต่างกันจริง — แต่ก็ต้องพิมพ์เพิ่มจริง ๆ และคุณต้องต้านความอยากที่จะ “DRY” ให้เหลือ model เดียวที่ permissive เพราะนั่นดึงความเสี่ยงเรื่อง spoof กลับมา

Repository class เทียบกับ query เขียน inline ใน route handler

• Pros: ทุก SQL สำหรับ resource อยู่ที่ testable เดียว; router ยังบางและอ่านเป็น intent ไม่ใช่ query mechanics; query ที่ใช้โดยสอง endpoint (get-by-id ทั้งใน “read” และ “update”) เขียนครั้งเดียว; การเปลี่ยนวิธี load บางอย่างแตะไฟล์เดียว
• Cons: นี่คืออีกชั้นที่ route ต้องผ่าน และสำหรับ endpoint query เดียวที่ trivial จริง ๆ repository รู้สึกเหมือนพิธีการ FitTrack ยอมรับนั่น: consistency (ทุก resource มี repo) มีค่ามากกว่าการโกน class ออกจาก route ที่ง่ายสุด และ workout query ไม่ trivial เลยสักนิด

ติดตั้ง

1. app/schemas/exercise.py

# app/schemas/exercise.py — the API edges for an exercise.
from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field


class ExerciseCreate(BaseModel):
    """What a client may send to create an exercise — nothing server-owned."""

    name: str = Field(min_length=1, max_length=120)
    muscle_group: str = Field(min_length=1, max_length=60)
    is_public: bool = False


class ExerciseRead(BaseModel):
    """What the API returns. Built directly from a SQLAlchemy Exercise."""

    model_config = ConfigDict(from_attributes=True)

    id: UUID
    name: str
    muscle_group: str
    is_public: bool
    created_by: UUID | None
    created_at: datetime

2. app/schemas/workout.py

# app/schemas/workout.py — a workout is created with its sets nested inside.
from datetime import datetime
from decimal import Decimal
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field


class SetInput(BaseModel):
    """One set in a create payload. No set_index — the server assigns it."""

    exercise_id: UUID
    reps: int = Field(gt=0)
    weight_kg: Decimal = Field(ge=0, max_digits=6, decimal_places=2)


class WorkoutCreate(BaseModel):
    """Log a whole session in one payload; at least one set is required."""

    performed_at: datetime | None = None  # defaults to now() server-side
    notes: str | None = None
    sets: list[SetInput] = Field(min_length=1)


class SetRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    exercise_id: UUID
    set_index: int
    reps: int
    weight_kg: Decimal


class WorkoutRead(BaseModel):
    """A workout with its sets, returned to the client."""

    model_config = ConfigDict(from_attributes=True)

    id: UUID
    user_id: UUID
    performed_at: datetime
    notes: str | None
    created_at: datetime
    sets: list[SetRead]

3. app/repositories/exercise.py

# app/repositories/exercise.py — every exercise query, one place.
from uuid import UUID

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Exercise
from app.schemas.exercise import ExerciseCreate


class ExerciseRepo:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def list_visible(self, user_id: UUID) -> list[Exercise]:
        """The public catalog plus the caller's own exercises."""
        stmt = (
            select(Exercise)
            .where(or_(Exercise.is_public.is_(True), Exercise.created_by == user_id))
            .order_by(Exercise.name)
        )
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def get(self, exercise_id: UUID) -> Exercise | None:
        return await self.session.get(Exercise, exercise_id)

    async def create(self, data: ExerciseCreate, user_id: UUID) -> Exercise:
        exercise = Exercise(
            name=data.name,
            muscle_group=data.muscle_group,
            is_public=data.is_public,
            created_by=user_id,  # server-owned, never from the client
        )
        self.session.add(exercise)
        await self.session.commit()  # get_session doesn't commit; the write must
        await self.session.refresh(exercise)
        return exercise

4. app/repositories/workout.py

# app/repositories/workout.py — create a session + its sets, read history.
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models import Workout, WorkoutSet
from app.schemas.workout import WorkoutCreate


class WorkoutRepo:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def create(self, data: WorkoutCreate, user_id: UUID) -> Workout:
        workout = Workout(user_id=user_id, notes=data.notes)
        if data.performed_at is not None:
            workout.performed_at = data.performed_at

        # The client sends sets in order; the server numbers them.
        for index, item in enumerate(data.sets):
            workout.sets.append(
                WorkoutSet(
                    exercise_id=item.exercise_id,
                    set_index=index,
                    reps=item.reps,
                    weight_kg=item.weight_kg,
                )
            )

        self.session.add(workout)  # cascade adds the sets too
        await self.session.commit()  # one commit: parent + all sets, atomically
        # Re-read with sets eagerly loaded so WorkoutRead can serialize them.
        return await self.get(workout.id, user_id)

    async def get(self, workout_id: UUID, user_id: UUID) -> Workout | None:
        """One workout — but only if it belongs to this user."""
        stmt = (
            select(Workout)
            .where(Workout.id == workout_id, Workout.user_id == user_id)
            .options(selectinload(Workout.sets))
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def list_for_user(self, user_id: UUID) -> list[Workout]:
        """The caller's history, most recent first, sets loaded."""
        stmt = (
            select(Workout)
            .where(Workout.user_id == user_id)
            .order_by(Workout.performed_at.desc())
            .options(selectinload(Workout.sets))
        )
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

มีสองอย่างที่ต้องสังเกต query filter บน user_id (get และ list_for_user) ดังนั้น repository return ได้แค่ workout ของผู้เรียกเองเท่านั้น — กฎ ownership จาก /me โผล่ที่นี่อีกครั้งในรูป where clause และทุกการอ่านที่แตะ workout.sets ใช้ selectinload: ใน async SQLAlchemy ไม่มี lazy loading ดังนั้น collection ที่คุณตั้งใจจะ serialize ต้อง eager-load ไว้ก่อน ไม่งั้นเข้าถึงทีหลังจะ raise selectinload fetch set ใน query ที่สองที่วางแผนไว้

ตรวจสอบผล

repository ต้องการ database และ router เพื่อ exercise ครบ — นั่นคือ module ถัดไป สิ่งที่คุณ check ได้ standalone คือ schema layer ที่เป็น validation ล้วน: ต้องยอมรับ workout ที่ well-formed และ ปฏิเสธ ตัวที่ malformed รัน round-trip ผ่าน uv:

uv run python -c "
from app.schemas.workout import WorkoutCreate

good = WorkoutCreate.model_validate({
    'notes': 'leg day',
    'sets': [
        {'exercise_id': '11111111-1111-1111-1111-111111111111', 'reps': 5, 'weight_kg': '100.00'},
        {'exercise_id': '11111111-1111-1111-1111-111111111111', 'reps': 5, 'weight_kg': '102.50'},
    ],
})
print('parsed', len(good.sets), 'sets; first weight is', type(good.sets[0].weight_kg).__name__)
"

parsed 2 sets; first weight is Decimal

น้ำหนักกลับมาเป็น Decimal ไม่ใช่ float — เป๊ะกับที่คุณต้องการสำหรับ precision ของเงินและน้ำหนัก ตอนนี้พิสูจน์ว่า guardrail กัด: set list ที่ว่างเปล่าและ reps ที่ไม่เป็นบวกต้องถูกปฏิเสธทั้งคู่:

uv run python -c "
from pydantic import ValidationError
from app.schemas.workout import WorkoutCreate

for bad in ({'sets': []}, {'sets': [{'exercise_id': '11111111-1111-1111-1111-111111111111', 'reps': 0, 'weight_kg': '50'}]}):
    try:
        WorkoutCreate.model_validate(bad)
        print('ERROR: accepted invalid payload')
    except ValidationError as exc:
        print('rejected:', exc.errors()[0]['msg'])
"

rejected: List should have at least 1 item after validation, not 0
rejected: Input should be greater than 0

สองการปฏิเสธ ไม่มี crash — schema บังคับ contract ก่อนสิ่งใดในนี้ถึง database เหลือ import check สุดท้ายว่า repository ต่อสายเข้า model ถูกต้อง ยืนยันว่า module load ได้:

uv run python -c "import app.repositories.exercise, app.repositories.workout; print('repos import cleanly')"

repos import cleanly

ตรวจสอบความเข้าใจ:

• ExerciseCreate ละ id, created_by, และ created_at ส่วน ExerciseRead รวม field พวกนั้นไว้ อะไรจะพังถ้าคุณรับ schema permissive เดียวสำหรับทั้งสองทิศทาง?
• SetInput ไม่มี set_index แต่ set ที่ store แล้วมี index ค่านั้นมาจากไหน และทำไม client ไม่ควรเป็นคนส่งมา?
• WorkoutRepo.create add workout พร้อม set ทั้งหมด แล้วเรียก session.commit() เดียว ทำไมต้อง commit เดียวแทนที่จะ commit แต่ละ set ตอน add — และ workout ที่ log ไปครึ่งเดียวจะหน้าตายังไงถ้า commit ทีละ set?
• ทุกการอ่านที่ return workout.sets ใช้ selectinload อะไรพังถ้าคุณละ selectinload ไว้ใต้ async session และทำไม?

สรุป

domain layer เสร็จแล้ว Pydantic v2 schema (ExerciseCreate/ExerciseRead, WorkoutCreate ที่มี list SetInput ซ้อน, WorkoutRead) คือ boundary ที่ validate แล้ว — input schema expose เฉพาะ field ที่ client ตั้งได้, output schema สร้างจาก model ตรง ๆ ผ่าน from_attributes, และทั้ง workout-with-sets validate ในทีเดียว Async repository (ExerciseRepo, WorkoutRepo) รับ AsyncSession และเก็บทุก query ไว้ที่เดียว: ownership filter เก็บการอ่าน scope ไว้ที่ผู้เรียก, selectinload eager-load collection สำหรับ async, และ write method commit unit of work ของตัวเอง — ทั้ง workout-with-sets ใน commit เดียว — ส่วน get_session แค่แจก session แล้วปิดให้ เรา verify ว่า schema ยอมรับ input ที่ดีและปฏิเสธ set list ที่ว่างและ reps ที่ไม่เป็นบวก นั่นคือทุกอย่างที่ API ต้องการ: model ไว้ store, schema ไว้ validate, repository ไว้ query ต่อไป The Exercises API → เปลี่ยนชิ้นส่วนพวกนี้เป็น endpoint จริง — router บาง ๆ ที่ authenticate ด้วย get_current_user, รับ AsyncSession, และเรียกตรงเข้า ExerciseRepo