Fase 5 · Bangun API dengan FastAPI

Lifespan Events

Lifespan adalah tempat membuat resource yang berumur selama aplikasi hidup: client, koneksi pool, model yang dimuat.

Sumber asli fastapi.tiangolo.com Resmi Rangkuman ~4 menit baca

Intisari

Kode sebelum yield berjalan sekali saat startup; kode sesudahnya saat shutdown.
Tempat yang benar untuk: client LLM, HTTP client, connection pool DB, model embedding.
Simpan di app.state, ambil lewat dependency.
Menggantikan @app.on_event("startup") yang sudah usang.
Kalau setup gagal, aplikasi gagal start — dan itu memang yang kamu mau.

Bentuknya

from contextlib import asynccontextmanager
from fastapi import FastAPI
import anthropic, httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── Startup ──────────────────────────────────────
    app.state.llm = anthropic.AsyncAnthropic()
    app.state.http = httpx.AsyncClient(timeout=30.0)
    app.state.vector = await ChromaClient.connect(settings.chroma_url)
    logger.info("aplikasi siap")

    yield                                  # ← aplikasi melayani request di sini

    # ── Shutdown ─────────────────────────────────────
    await app.state.http.aclose()
    await app.state.llm.close()
    await app.state.vector.close()
    logger.info("aplikasi berhenti")


app = FastAPI(lifespan=lifespan)

@app.on_event("startup") dan ("shutdown") sudah usang. Kalau kamu menemukannya di tutorial, itu tandanya tutorial-nya lama. Pakai lifespan — satu fungsi untuk keduanya, dan setup/teardown yang berpasangan jadi berdekatan di kode.

Kenapa ini penting

# ❌ client baru untuk SETIAP request
@app.post("/chat")
async def chat(req: ChatRequest):
    client = anthropic.AsyncAnthropic()      # pool baru, handshake TLS baru
    return await client.messages.create(...)

# ✅ satu client untuk seumur hidup aplikasi
@app.post("/chat")
async def chat(req: ChatRequest, client: ClientDep):
    return await client.messages.create(...)

Biayanya nyata: handshake TLS memakan 100–300 ms. Pada 1.000 request per menit, membuat client baru tiap kali berarti membuang beberapa menit CPU dan latensi per jam — untuk pekerjaan yang seharusnya nol.

Mengaksesnya lewat dependency

from typing import Annotated
from fastapi import Depends, Request

def get_llm(request: Request) -> anthropic.AsyncAnthropic:
    return request.app.state.llm

def get_http(request: Request) -> httpx.AsyncClient:
    return request.app.state.http

LlmDep = Annotated[anthropic.AsyncAnthropic, Depends(get_llm)]
HttpDep = Annotated[httpx.AsyncClient, Depends(get_http)]


@app.post("/chat")
async def chat(req: ChatRequest, llm: LlmDep) -> ChatResponse:
    r = await llm.messages.create(...)
    return ChatResponse(...)

Membungkusnya dalam dependency (bukan mengakses app.state langsung) membuatnya bisa di-override saat testing.

Apa yang layak masuk lifespan

Resource	Kenapa
Client LLM	Connection pool, konfigurasi retry
HTTP client	Keep-alive, connection pooling
Connection pool DB	Membuat koneksi itu mahal
Client vector store	Koneksi persisten
Model embedding lokal	Memuat ke memori butuh detik hingga menit
Client cache (Redis)	Connection pool
Latar belakang berkala	Task yang harus dimulai dan dihentikan bersama aplikasi

Gagal saat startup itu bagus

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.settings = Settings()          # gagal kalau env var wajib kosong

    app.state.vector = await ChromaClient.connect(settings.chroma_url)
    if not await app.state.vector.ping():
        raise RuntimeError("vector store tidak bisa dihubungi")

    yield
    await app.state.vector.close()

Aplikasi yang gagal start lebih baik daripada aplikasi yang start lalu mengembalikan 500. Orkestrator container (Docker, ECS, Kubernetes) akan melihat kegagalan itu dan tidak mengalihkan trafik ke instance yang rusak. Deploy-nya gagal dengan jelas, bukan diam-diam melayani error.

Task latar belakang berkala

import asyncio

async def segarkan_index(app: FastAPI) -> None:
    while True:
        try:
            await asyncio.sleep(3600)
            await app.state.vector.reindex()
        except asyncio.CancelledError:
            raise                     # jangan telan — ini sinyal shutdown
        except Exception:
            logger.exception("gagal reindex")


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.vector = await ChromaClient.connect(...)
    task = asyncio.create_task(segarkan_index(app))

    yield

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    await app.state.vector.close()

Lifespan saat testing

from fastapi.testclient import TestClient

# ❌ lifespan TIDAK berjalan
client = TestClient(app)
client.get("/")

# ✅ lifespan berjalan
with TestClient(app) as client:
    client.get("/")

Ini sumber kebingungan yang umum. Tanpa blok with, lifespan tidak berjalan dan app.state.llm tidak ada — testmu akan gagal dengan AttributeError yang membingungkan. Selalu pakai TestClient sebagai context manager.

Health check

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

@app.get("/ready")
async def ready(request: Request) -> dict[str, bool]:
    return {
        "vector_store": await request.app.state.vector.ping(),
        "database": await request.app.state.db.ping(),
    }

Bedanya: /health menjawab "proses ini hidup" (liveness), /ready menjawab "aplikasi ini siap melayani trafik" (readiness). Orkestrator memakai keduanya untuk keputusan yang berbeda.

Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.