Dependencies
Dependency injection di FastAPI: satu mekanisme untuk konfigurasi, koneksi, autentikasi, dan testing.
Intisari
Depends()menjalankan sebuah fungsi sebelum endpoint-mu dan menyuntikkan hasilnya.- Dependency dengan
yieldmemberi setup dan teardown โ pola yang tepat untuk session DB. - Dependency bisa bertingkat: satu dependency boleh bergantung pada dependency lain.
dependency_overridesmembuat testing mudah โ ganti apa pun tanpa monkeypatch.- Bungkus dengan
Annotated[...]supaya bisa dipakai ulang tanpa menulisDepends()berulang.
Bentuk dasarnya
from typing import Annotated
from fastapi import Depends
def parameter_umum(q: str | None = None, limit: int = 20) -> dict:
return {"q": q, "limit": limit}
@app.get("/dokumen")
async def daftar(p: Annotated[dict, Depends(parameter_umum)]):
return cari(**p)
FastAPI memanggil parameter_umum, mengambil query parameter dari request,
lalu menyerahkan hasilnya ke endpoint-mu.
Alias yang bisa dipakai ulang
from typing import Annotated
from fastapi import Depends
SettingsDep = Annotated[Settings, Depends(get_settings)]
ClientDep = Annotated[AsyncAnthropic, Depends(get_client)]
DbDep = Annotated[AsyncSession, Depends(get_db)]
@app.post("/chat")
async def chat(req: ChatRequest, client: ClientDep, db: DbDep) -> ChatResponse:
...
Pola alias Annotated ini layak dipakai sejak awal. Ia membuat tanda tangan
endpoint tetap pendek, memberi type hint yang benar untuk mypy dan editor, dan menghilangkan
pengulangan = Depends(...) di setiap endpoint.
Dependency dengan yield
async def get_db() -> AsyncIterator[AsyncSession]:
async with async_session() as session:
try:
yield session # โ diserahkan ke endpoint
await session.commit()
except Exception:
await session.rollback()
raise
Kode sebelum yield berjalan sebelum endpoint; kode setelahnya berjalan setelah
respons dikirim. Ini pola yang tepat untuk apa pun yang perlu dibersihkan: session database,
file terbuka, lock.
Menyuntik client LLM
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import anthropic
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.llm = anthropic.AsyncAnthropic()
yield
await app.state.llm.close()
app = FastAPI(lifespan=lifespan)
def get_client(request: Request) -> anthropic.AsyncAnthropic:
return request.app.state.llm
ClientDep = Annotated[anthropic.AsyncAnthropic, Depends(get_client)]
@app.post("/chat")
async def chat(req: ChatRequest, client: ClientDep) -> ChatResponse:
r = await client.messages.create(...)
return ChatResponse(...)
Client dibuat sekali saat startup, bukan per request. Membuat
AsyncAnthropic() di dalam endpoint akan membangun connection pool baru tiap kali โ
handshake TLS berulang, memori terbuang, dan latensi bertambah.
Dependency bertingkat
def get_settings() -> Settings:
return Settings()
def get_client(s: Annotated[Settings, Depends(get_settings)]) -> AsyncAnthropic:
return AsyncAnthropic(api_key=s.anthropic_api_key.get_secret_value())
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: DbDep,
) -> User:
user = await db.cari_by_token(token)
if user is None:
raise HTTPException(401, "token tidak valid")
return user
FastAPI menyelesaikan seluruh grafnya dan meng-cache tiap dependency per request โ
jadi get_settings hanya berjalan sekali meski dipakai tiga dependency berbeda.
Autentikasi
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def user_terautentikasi(
kredensial: Annotated[HTTPAuthorizationCredentials, Depends(security)],
) -> User:
user = await verifikasi(kredensial.credentials)
if user is None:
raise HTTPException(401, "tidak terautentikasi")
return user
UserDep = Annotated[User, Depends(user_terautentikasi)]
@app.post("/chat")
async def chat(req: ChatRequest, user: UserDep) -> ChatResponse:
...
Dependency tanpa nilai kembali
async def cek_kuota(user: UserDep) -> None:
if await pemakaian_hari_ini(user.id) > user.kuota:
raise HTTPException(429, "kuota harian habis")
@app.post("/chat", dependencies=[Depends(cek_kuota)])
async def chat(req: ChatRequest) -> ChatResponse:
...
Pakai parameter dependencies=[...] kalau kamu hanya butuh efek sampingnya (pengecekan), bukan nilainya.
Di seluruh router atau aplikasi
router = APIRouter(prefix="/admin", dependencies=[Depends(harus_admin)])
app = FastAPI(dependencies=[Depends(rate_limit)])
Override untuk testing
from fastapi.testclient import TestClient
def client_palsu() -> FakeAnthropic:
return FakeAnthropic(balasan="jawaban tetap untuk test")
app.dependency_overrides[get_client] = client_palsu
def test_chat():
with TestClient(app) as c:
r = c.post("/chat", json={"pesan": "halo", "session_id": "s1"})
assert r.status_code == 200
assert r.json()["balasan"] == "jawaban tetap untuk test"
app.dependency_overrides.clear()
Ini alasan terbesar memakai dependency injection. Tanpa DI, mengganti client LLM di test
berarti monkeypatch modul global โ rapuh dan mudah bocor antar test. Dengan
dependency_overrides, kamu mengganti simpul mana pun di graf dependency secara
eksplisit, tanpa menyentuh kode aplikasi.
Sebagai fixture pytest
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
app.dependency_overrides[get_client] = client_palsu
app.dependency_overrides[get_db] = db_uji
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
def test_chat(client):
r = client.post("/chat", json={"pesan": "halo", "session_id": "s1"})
assert r.status_code == 200
Ringkasan pola
| Kebutuhan | Pola |
|---|---|
| Konfigurasi | Depends(get_settings) + lru_cache |
| Client LLM / HTTP | Buat di lifespan, ambil dari app.state |
| Session database | Dependency dengan yield |
| Autentikasi | Dependency yang mengembalikan User |
| Otorisasi / rate limit | dependencies=[...], tanpa nilai kembali |
| Parameter berulang | Fungsi dependency biasa |
| Testing | app.dependency_overrides |
Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.