Python FastAPI E-Learning Part 2: Async ORM & Type Safety – SQLAlchemy 2.0 Async, Pydantic v2 & Pencegahan Masalah N+1

Tantangan Persistensi Asinkron di Ekosistem Python

Ketika bertransisi dari backend monolitik berbasis PHP atau Ruby ke Python, pengembang sering kali terjebak dalam kesalahpahaman fatal: "Membungkus query database di dalam fungsi async def otomatis membuatnya berjalan asinkron."

Kenyataannya, jika Anda menggunakan driver database sinkron konvensional (seperti psycopg2 standar atau pola lama SQLAlchemy 1.4), setiap query SQL akan memblokir (blocking I/O) event loop asyncio yang bersifat single-threaded. Bayangkan saat 200 siswa serentak memuat soal ujian CBT; seluruh coroutine akan antre secara sekuensial menunggu socket I/O selesai, menyebabkan latensi melonjak drastis dan throughput server anjlok.

Untuk mencapai konkurensi non-blocking murni pada proyek mfarim/fastapi-elearning-react, kami membangun fondasi persistensi dan validasi data berdasarkan tiga pilar utama:

1. SQLAlchemy 2.0 Async Engine dengan asyncpg: Connection pooling PostgreSQL asinkron native berkecepatan tinggi.
2. Pola Eager Loading Deterministik: Eliminasi total jebakan query N+1 dan error MissingGreenlet menggunakan selectinload.
3. Pipeline Validasi Pydantic v2: Parsing payload JSON dan serialisasi sub-milidetik yang ditenagai oleh mesin Rust (pydantic-core).

1. Arsitektur Async Engine & Connection Pool

Pada berkas backend/app/core/database.py, engine database dikonfigurasi untuk menangani beban konkurensi tinggi menggunakan driver biner asinkron PostgreSQL asyncpg:

# backend/app/core/database.py
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings

# 1. Connection Pool Asinkron
engine = create_async_engine(
    settings.DATABASE_URL,       # postgresql+asyncpg://user:pass@host:5432/elearning
    echo=False,                  # Matikan di production untuk menghindari overhead logging
    future=True,                 # Wajibkan standar sintaks SQLAlchemy 2.0
    pool_pre_ping=True,          # Tes koneksi sebelum checkout (cegah broken pipe socket)
    pool_size=10,                # Koneksi persisten per proses worker
    max_overflow=20,             # Kapasitas burst saat lonjakan ujian serentak
)

# 2. Asynchronous Session Factory
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,      # KRUSIAL: Mencegah error MissingGreenlet pasca-commit
    class_=AsyncSession,
)

class Base(DeclarativeBase):
    """DeclarativeBase modern menggantikan fungsi legacy declarative_base()"""
    pass

Mengapa expire_on_commit=False Bersifat Wajib?

Pada SQLAlchemy sinkron tradisional, pemanggilan session.commit() secara otomatis mengosongkan status memori (expire) dari seluruh atribut objek model. Ketika Anda kemudian membaca exam.title atau exam.questions, SQLAlchemy secara diam-diam menjalankan query SELECT baru di balik layar (Lazy Loading).

Namun, di lingkungan asinkron, implicit I/O semacam itu dilarang keras oleh arsitektur Python karena akses properti biasa seperti user.roles tidak dapat diawali dengan kata kunci await. Akibatnya, sistem akan melempar error fatal:

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call a greenlet-waiting function in a different greenlet.

Dengan menyetel expire_on_commit=False, objek model mempertahankan seluruh atribut yang telah dimuat di memori setelah commit(). Hal ini memungkinkan endpoint langsung mengembalikan model ke Pydantic untuk diserialisasi tanpa memicu query tersembunyi ataupun tabrakan greenlet.

2. Deklarasi Model 2.0 Modern dengan Mapped dan mapped_column

Model SQLAlchemy gaya lama kerap menggunakan sintaks longgar (Column(Integer, primary_key=True)), yang membingungkan linter IDE dan menyembunyikan potensi bug ketidaksesuaian tipe data.

Di berkas backend/app/models/exam.py, seluruh model memanfaatkan anotasi tipe data PEP 484 Python modern via Mapped[T] dan mapped_column():

# backend/app/models/exam.py
from datetime import datetime, date
from typing import Optional, List, Any
from sqlalchemy import BigInteger, String, Boolean, Integer, Date, DateTime, ForeignKey, Text, JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base

class Examination(Base):
    __tablename__ = "examinations"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    teacher_id: Mapped[int] = mapped_column(
        BigInteger, ForeignKey("teachers.id", ondelete="CASCADE"), nullable=False, index=True
    )
    subject_id: Mapped[int] = mapped_column(
        BigInteger, ForeignKey("subjects.id", ondelete="CASCADE"), nullable=False
    )
    classroom_id: Mapped[int] = mapped_column(
        BigInteger, ForeignKey("classrooms.id", ondelete="CASCADE"), nullable=False, index=True
    )
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    duration_minutes: Mapped[int] = mapped_column(Integer, default=60, nullable=False)
    passing_score: Mapped[int] = mapped_column(Integer, default=75, nullable=False)
    status: Mapped[str] = mapped_column(String(20), default="draft", nullable=False)
    
    # Pemetaan relasi bertipe statis dengan strategi pemuatan eksplisit
    teacher: Mapped["Teacher"] = relationship("Teacher", lazy="selectin")
    subject: Mapped["Subject"] = relationship("Subject", lazy="selectin")
    classroom: Mapped["Classroom"] = relationship("Classroom", lazy="selectin")
    questions: Mapped[List["Question"]] = relationship(
        "Question", back_populates="examination", cascade="all, delete-orphan"
    )
    attempts: Mapped[List["ExamAttempt"]] = relationship(
        "ExamAttempt", back_populates="examination", cascade="all, delete-orphan"
    )

Keuntungan pengetikan statis ini:

• id: Mapped[int] menjamin bahwa exam.id selalu bertipe integer murni.
• description: Mapped[Optional[str]] mendokumentasikan secara presisi kolom teks yang dapat bernilai None.
• questions: Mapped[List["Question"]] menghadirkan auto-complete komprehensif pada struktur data relasi di seluruh codebase.

3. Menghindari Jebakan Query N+1: selectinload vs. joinedload

Masalah query N+1 adalah musuh utama performa database pada aplikasi berbasis web.

Misalkan kita memuat daftar 20 ujian sekolah. Jika backend membaca ke-20 ujian tersebut lalu melakukan iterasi exam.questions, ORM naif akan mengeksekusi 1 query awal ditambah 20 query susulan:

$$\text{Total Query} = 1 + N = 21 \text{ query}$$

Di dalam event loop asinkron, query sekuensial beruntun ini melumpuhkan throughput. Di SQLAlchemy 2.0, kita menyelesaikannya secara elegan:

Opsi A: joinedload (SQL Outer Join)

Mengeksekusi satu query tunggal dengan LEFT OUTER JOIN:

stmt = select(Examination).options(joinedload(Examination.questions))

• Kelemahan: Untuk relasi one-to-many (misal 50 soal per ujian), joinedload menghasilkan duplikasi baris Cartesian product yang masif di jaringan dan membebani alokasi memori deserialisasi Python.

Opsi B: selectinload (Batch IN Query) — Pendekatan Ideal

Mengeksekusi tepat dua query terindeks yang sangat efisien, berapa pun banyaknya data:

# Query 1: Ambil data seluruh ujian
# SELECT * FROM examinations WHERE status = 'published';

# Query 2: Ambil seluruh soal sekaligus menggunakan klausa IN
# SELECT * FROM questions WHERE examination_id IN (1, 2, 3, ..., 20);

Pada berkas backend/app/api/v1/exams.py, kita mengimplementasikannya secara berantai (chained eager loading):

stmt = (
    select(Examination)
    .options(
        selectinload(Examination.teacher),
        selectinload(Examination.subject),
        selectinload(Examination.classroom),
        selectinload(Examination.questions).selectinload(Question.options),
    )
    .where(Examination.id == exam_id, Examination.deleted_at.is_(None))
)
result = await db.execute(stmt)
exam = result.scalars().first()

Dengan merangkai .selectinload(Examination.questions).selectinload(...), seluruh pohon relasi hierarkis (Ujian $\rightarrow$ Soal $\rightarrow$ Opsi Jawaban) diambil hanya dalam 3 query terindeks berkecepatan tinggi, sepenuhnya mengeliminasi beban N+1 query.

4. Siklus Hidup Transaksi & Dependency Injection Bersih

Sistem Dependency Injection FastAPI menjadikan pengelolaan batas transaksi database sangat bersih. Di backend/app/core/database.py:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Manfaat Arsitektural:

1. Auto-Commit Otomatis: Handler endpoint tidak perlu mengulang kode boilerplate await db.commit() untuk mutasi standar; keluar dari fungsi handler secara normal akan langsung melakukan commit.
2. Fail-Safe Rollback Terpadu: Jika terjadi unhandled exception atau error validasi di tengah jalan, session langsung dibatalkan (rollback), mencegah data korup atau transaksi gantung.
3. Pelepasan Socket Terjamin: Blok finally: memastikan koneksi database selalu dikembalikan ke pool, bahkan jika klien menutup koneksi browser secara mendadak.

5. Pydantic v2: Validasi Berbasis Rust & Envelope Seragam

Pada FastAPI, Data Transfer Object (DTO) berfungsi sebagai benteng validasi antara payload HTTP eksternal dan entitas ORM internal. Di backend/app/schemas/exam.py:

# backend/app/schemas/exam.py
from datetime import datetime
from typing import Optional, List, Any
from pydantic import BaseModel, Field, ConfigDict

class QuestionOptionDto(BaseModel):
    key: str = Field(..., description="Label pilihan: A, B, C, D, atau E")
    text: str = Field(..., min_length=1, description="Teks pilihan jawaban")

class QuestionResponse(BaseModel):
    id: int
    question_text: str
    question_type: str
    options: Optional[Any] = None
    points: int = 1
    difficulty: str = "medium"

    model_config = ConfigDict(from_attributes=True)  # Menggantikan orm_mode = True legacy

class ExamStartResponse(BaseModel):
    attempt_id: int
    examination_id: int
    duration_minutes: int
    remaining_seconds: int
    started_at: datetime
    questions: List[QuestionResponse]

    model_config = ConfigDict(from_attributes=True)

Standarisasi Envelope API (ApiResponse[T])

Agar frontend (React, Next.js, atau aplikasi mobile) menerima struktur respons JSON yang konsisten di semua situasi, kami mengimplementasikan envelope generik di backend/app/common/response.py:

# backend/app/common/response.py
from datetime import datetime
from typing import Generic, TypeVar, Optional, Any
from pydantic import BaseModel, Field

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "Success"
    data: Optional[T] = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

    @classmethod
    def ok(cls, data: Optional[T] = None, message: str = "Success") -> "ApiResponse[T]":
        return cls(success=True, message=message, data=data)

    @classmethod
    def error(cls, message: str = "Error", data: Optional[Any] = None) -> "ApiResponse[Any]":
        return cls(success=False, message=message, data=data)

Ketika digunakan pada endpoint:

@router.get("/{id}", response_model=ApiResponse[ExamStartResponse])
async def get_exam(id: int, db: AsyncSession = Depends(get_db)):
    exam = await exam_service.get_by_id(db, id)
    return ApiResponse.ok(data=exam, message="Data ujian berhasil diambil")

Hasil JSON yang dikirimkan terstruktur rapi, terdefinisi otomatis pada dokumentasi Swagger UI (/docs), dan seragam di seluruh modul:

{
  "success": true,
  "message": "Data ujian berhasil diambil",
  "data": {
    "attempt_id": 142,
    "examination_id": 12,
    "duration_minutes": 90,
    "remaining_seconds": 5400,
    "started_at": "2026-10-05T08:00:00Z",
    "questions": [
      {
        "id": 101,
        "question_text": "Apa keunggulan utama selectinload pada SQLAlchemy 2.0?",
        "question_type": "multiple_choice",
        "points": 5,
        "difficulty": "medium"
      }
    ]
  },
  "timestamp": "2026-10-05T08:00:01.124Z"
}

Uji Komparasi: ORM Sinkron vs. SQLAlchemy 2.0 Async

Hasil benchmark pengujian stres beban (stress testing) menggunakan Apache Bench (ab -n 5000 -c 100) pada server VPS 4-Core saat membaca payload ujian:

Ringkasan Rekomendasi Arsitektural

1. Wajib pasang expire_on_commit=False pada session factory asinkron untuk mencegah error MissingGreenlet saat serialisasi pasca-commit.
2. Prioritaskan selectinload untuk relasi one-to-many: Menghindari ledakan duplikasi baris Cartesian product sekaligus menghentikan masalah N+1.
3. Migrasi ke Mapped[T] dan mapped_column: Memberikan jaminan keamanan tipe data statis dan auto-complete sempurna di seluruh lapisan kode.
4. Gunakan ConfigDict(from_attributes=True) di Pydantic v2: Membaca atribut model ORM secara langsung secepat eksekusi biner Rust.

Pada Part 3, kita akan membedah jantung real-time sistem ini: Engine Ujian CBT Presisi Tinggi & Telemetri Pengawasan Native WebSocket, meliputi sinkronisasi timer server, pencegahan kecurangan, dan siaran telemetri pengawas via Redis Pub/Sub.