Bagian 3: Persistensi Database

Penyimpanan tugas_store dari Bagian 2 hidup hanya di memori proses: begitu server mati atau restart, seluruh tugas lenyap. Bagian ini memindahkan penyimpanan ke SQLite sehingga data bertahan antar-proses, memakai module sqlite3 dari standard library persis seperti chapter SQLite dengan Python dan CRUD dengan SQLite. Tak ada ORM — kita menulis SQL langsung dengan parameterized queries dari chapter Parameterized Queries.

Struktur proyek bertambah app/database.py; app/__init__.py dari Bagian 1 tetap ada dan tetap kosong:

text
api-manajemen-tugas/
├── pyproject.toml
└── app/
    ├── __init__.py
    ├── database.py
    ├── main.py
    └── models.py

Koneksi per Request sebagai Dependency #

Buat berkas app/database.py:

python
import sqlite3
from collections.abc import Iterator

DB_PATH = "tugas.db"

SKEMA = """
CREATE TABLE IF NOT EXISTS tugas (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    judul TEXT NOT NULL,
    selesai INTEGER NOT NULL DEFAULT 0
)
"""


def dapatkan_koneksi() -> Iterator[sqlite3.Connection]:
    koneksi = sqlite3.connect(DB_PATH)
    koneksi.row_factory = sqlite3.Row
    koneksi.execute(SKEMA)
    try:
        yield koneksi
    finally:
        koneksi.close()

Mari kita bedah bagiannya:

DB_PATH = "tugas.db": nama berkas database, relatif terhadap direktori kerja. Konstanta ini satu-satunya tempat yang perlu diganti saat menguji (Bagian 5) dan saat konfigurasi (Bagian 6).
SKEMA: CREATE TABLE IF NOT EXISTS membuat tabel hanya bila belum ada, sehingga menjalankan ulang aplikasi tidak pernah merusak data. Kolom selesai bertipe INTEGER karena SQLite tidak punya tipe BOOLEAN — nilai 0 dan 1 yang kita konversi bolak-balik.
def dapatkan_koneksi() -> Iterator[...]: ini generator dengan yield, sehingga FastAPI memperlakukannya sebagai dependency dengan siklus hidup — pola yield dari chapter Dependency Injection. Koneksi dibuat sebelum request, diserahkan ke fungsi rute, lalu ditutup setelahnya.
koneksi.row_factory = sqlite3.Row: membuat hasil query bisa diakses dengan nama kolom (baris["judul"]), bukan indeks angka yang rapuh.
try ... finally: finally menjamin koneksi.close() terpanggil meskipun fungsi rute melempar exception — disiplin penutupan resource dari chapter Context Manager dan with.

CRUD di Atas SQLite #

Ganti seluruh isi app/main.py:

python
import sqlite3

from fastapi import Depends, FastAPI, status

from app.database import dapatkan_koneksi
from app.models import TugasBuat, TugasTampil

app = FastAPI(title="API Manajemen Tugas", version="0.3.0")


def baris_ke_tugas(baris: sqlite3.Row) -> dict:
    return {
        "id": baris["id"],
        "judul": baris["judul"],
        "selesai": bool(baris["selesai"]),
    }


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/tugas", response_model=list[TugasTampil])
def daftar_tugas(koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)) -> list[dict]:
    baris_semua = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas ORDER BY id"
    ).fetchall()
    return [baris_ke_tugas(baris) for baris in baris_semua]


@app.post("/tugas", response_model=TugasTampil, status_code=status.HTTP_201_CREATED)
def buat_tugas(
    payload: TugasBuat, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> dict:
    with koneksi:
        kursor = koneksi.execute(
            "INSERT INTO tugas (judul, selesai) VALUES (?, ?)",
            (payload.judul, 0),
        )
    baris = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas WHERE id = ?",
        (kursor.lastrowid,),
    ).fetchone()
    return baris_ke_tugas(baris)


@app.get("/tugas/{tugas_id}", response_model=TugasTampil)
def ambil_tugas(
    tugas_id: int, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> dict:
    baris = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas WHERE id = ?",
        (tugas_id,),
    ).fetchone()
    if baris is None:
        raise LookupError(f"tugas dengan id {tugas_id} tidak ditemukan")
    return baris_ke_tugas(baris)


@app.delete("/tugas/{tugas_id}", status_code=status.HTTP_204_NO_CONTENT)
def hapus_tugas(
    tugas_id: int, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> None:
    with koneksi:
        kursor = koneksi.execute(
            "DELETE FROM tugas WHERE id = ?",
            (tugas_id,),
        )
    if kursor.rowcount == 0:
        raise LookupError(f"tugas dengan id {tugas_id} tidak ditemukan")

Ada lima hal baru yang perlu diurai:

koneksi: sqlite3.Connection = Depends(dapatkan_koneksi): setiap fungsi rute meminta koneksi dari dependency — FastAPI memanggil dapatkan_koneksi() sekali per request dan menutupnya setelah fungsi rute selesai.
baris_ke_tugas: satu tempat konversi dari baris database ke dictionary, termasuk bool(baris["selesai"]) yang mengubah 0/1 menjadi False/True.
with koneksi:: block with pada objek koneksi sqlite3 mengelola transaksi — perubahan di-commit saat block selesai, di-rollback bila terjadi exception. Koneksi tidak tertutup di sini; penutupan tetap tugas dependency.
kursor.lastrowid: id yang diberikan SQLite untuk baris yang baru di-INSERT, dipakai untuk membaca kembali tugas tersebut.
kursor.rowcount: jumlah baris yang terkena DELETE — nol berarti id tidak ada, dan kita tetap memakai LookupError sampai Bagian 4 menggantinya dengan 404.

Perhatikan juga setiap SQL memakai tanda ? untuk nilai dari klien — nilai tidak pernah dirangkai ke string SQL, sehingga pintu masuk SQL injection tertutup sejak awal.

Berkas app/models.py tidak berubah dari Bagian 2 dan ditampilkan kembali agar tahap ini utuh:

python
from pydantic import BaseModel, Field


class TugasBuat(BaseModel):
    judul: str = Field(min_length=1, examples=["Belajar Python"])


class TugasTampil(BaseModel):
    id: int
    judul: str
    selesai: bool

Bukti Data Bertahan #

Bukti terbaik adalah dua proses Python yang berbeda. Proses pertama mengisi dua tugas lewat skrip sementara cek_api.py:

python
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

respons = client.post("/tugas", json={"judul": "Belajar Python"})
print(respons.status_code, respons.json())

respons = client.post("/tugas", json={"judul": "Belajar REST API"})
print(respons.status_code, respons.json())

respons = client.get("/tugas")
print(respons.status_code, respons.json())

Output yang Dihasilkan:

text
{'id': 1, 'judul': 'Belajar Python', 'selesai': False}
{'id': 2, 'judul': 'Belajar REST API', 'selesai': False}
[{'id': 1, 'judul': 'Belajar Python', 'selesai': False}, {'id': 2, 'judul': 'Belajar REST API', 'selesai': False}]

Ganti isi cek_api.py menjadi pembacaan saja, lalu jalankan sebagai proses baru — tanpa mengisi ulang apa pun:

python
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

respons = client.get("/tugas")
print(respons.status_code, respons.json())

Output yang Dihasilkan:

text

200 [{'id': 1, 'judul': 'Belajar Python', 'selesai': False}, {'id': 2, 'judul': 'Belajar REST API', 'selesai': False}]

Kedua tugas masih ada. Di Bagian 2, proses kedua pasti menjumpai list kosong karena tugas_store hanya ada di memori; kini status aplikasi dan status database tidak lagi menyatu — restart aplikasi tidak menghapus apa pun di tugas.db. Sebagai bukti terakhir, intip isi berkas tersebut secara langsung dari luar API:

python
import sqlite3

koneksi = sqlite3.connect("tugas.db")
baris_semua = koneksi.execute("SELECT id, judul, selesai FROM tugas").fetchall()
for baris in baris_semua:
    print(baris)
koneksi.close()

Output yang Dihasilkan:

text

(1, 'Belajar Python', 0)
(2, 'Belajar REST API', 0)

Di dalam berkas, selesai tersimpan sebagai 0 dan 1 — hanya fungsi baris_ke_tugas di lapisan aplikasi yang mengubahnya menjadi False dan True bagi klien.

Data kini bertahan, tetapi API masih menjawab 500 untuk id yang tidak ada dan belum bisa memperbarui tugas sebagian. Kode lengkap tahap ini tersedia pada bagian Berkas Sumber di bawah halaman. Bagian 4 menuntaskan keduanya.

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

python
import sqlite3
from collections.abc import Iterator

DB_PATH = "tugas.db"

SKEMA = """
CREATE TABLE IF NOT EXISTS tugas (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    judul TEXT NOT NULL,
    selesai INTEGER NOT NULL DEFAULT 0
)
"""


def dapatkan_koneksi() -> Iterator[sqlite3.Connection]:
    koneksi = sqlite3.connect(DB_PATH)
    koneksi.row_factory = sqlite3.Row
    koneksi.execute(SKEMA)
    try:
        yield koneksi
    finally:
        koneksi.close()

python
import sqlite3

from fastapi import Depends, FastAPI, status

from app.database import dapatkan_koneksi
from app.models import TugasBuat, TugasTampil

app = FastAPI(title="API Manajemen Tugas", version="0.3.0")


def baris_ke_tugas(baris: sqlite3.Row) -> dict:
    return {
        "id": baris["id"],
        "judul": baris["judul"],
        "selesai": bool(baris["selesai"]),
    }


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/tugas", response_model=list[TugasTampil])
def daftar_tugas(koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)) -> list[dict]:
    baris_semua = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas ORDER BY id"
    ).fetchall()
    return [baris_ke_tugas(baris) for baris in baris_semua]


@app.post("/tugas", response_model=TugasTampil, status_code=status.HTTP_201_CREATED)
def buat_tugas(
    payload: TugasBuat, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> dict:
    with koneksi:
        kursor = koneksi.execute(
            "INSERT INTO tugas (judul, selesai) VALUES (?, ?)",
            (payload.judul, 0),
        )
    baris = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas WHERE id = ?",
        (kursor.lastrowid,),
    ).fetchone()
    return baris_ke_tugas(baris)


@app.get("/tugas/{tugas_id}", response_model=TugasTampil)
def ambil_tugas(
    tugas_id: int, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> dict:
    baris = koneksi.execute(
        "SELECT id, judul, selesai FROM tugas WHERE id = ?",
        (tugas_id,),
    ).fetchone()
    if baris is None:
        raise LookupError(f"tugas dengan id {tugas_id} tidak ditemukan")
    return baris_ke_tugas(baris)


@app.delete("/tugas/{tugas_id}", status_code=status.HTTP_204_NO_CONTENT)
def hapus_tugas(
    tugas_id: int, koneksi: sqlite3.Connection = Depends(dapatkan_koneksi)
) -> None:
    with koneksi:
        kursor = koneksi.execute(
            "DELETE FROM tugas WHERE id = ?",
            (tugas_id,),
        )
    if kursor.rowcount == 0:
        raise LookupError(f"tugas dengan id {tugas_id} tidak ditemukan")

python
from pydantic import BaseModel, Field


class TugasBuat(BaseModel):
    judul: str = Field(min_length=1, examples=["Belajar Python"])


class TugasTampil(BaseModel):
    id: int
    judul: str
    selesai: bool