Belajar FastAPI - Background Tasks & Async DB

#Pendahuluan

Setelah di episode 13 API kalian bisa berkomunikasi realtime, sekarang kita bahas dua hal yang membuat aplikasi responsif dan skala besar: background tasks dan database async. Keduanya menjawab pertanyaan yang sama dari sudut berbeda — bagaimana supaya klien tidak menunggu pekerjaan yang sebenarnya tidak perlu ditunggunya?

Mengapa episode ini penting? Di produksi, endpoint yang mengirim email, memproses file, atau menghitung laporan tidak boleh membuat klien menunggu. Dan database yang diakses secara sinkron di dalam handler async akan membunuh performa begitu traffic naik. Episode ini memberi kalian kedua senjatanya.

#Background Tasks: Setelah Response

BackgroundTasks menjalankan fungsi setelah response dikirim ke klien. Klien tidak menunggu — server menyelesaikan tugas di belakang:

Background task sederhana

import time
 
from fastapi import BackgroundTasks, FastAPI
 
app = FastAPI()
 
 
def send_welcome_email(email: str) -> None:
    time.sleep(2)  # simulasi kirim email
    print(f"email terkirim ke {email}")
 
 
@app.post("/users/")
def create_user(
    email: str,
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    background_tasks.add_task(send_welcome_email, email)
    return {"status": "user dibuat"}

background_tasks.add_task(func, *args) menjadwalkan send_welcome_email dieksekusi setelah response. Kalian bisa menambahkan banyak tugas — semuanya dijalankan dalam urutan yang sama.

#Background Task dengan Fungsi Async

Background tasks juga mendukung fungsi async def — namun penting dipahami: keduanya dijalankan bersamaan (concurrent), bukan berurutan:

Background task async

import asyncio
 
from fastapi import BackgroundTasks, FastAPI
 
app = FastAPI()
 
 
async def notify_async(user_id: int) -> None:
    await asyncio.sleep(2)
    print(f"notify user {user_id}")
 
 
@app.post("/orders/")
async def create_order(
    order_id: int,
    background_tasks: BackgroundTasks,
) -> dict[str, int]:
    background_tasks.add_task(notify_async, order_id)
    return {"order_id": order_id}

Pendekatan ini sempurna untuk mengirim notifikasi push atau WebSocket broadcast (episode 13) tanpa memblokir response.

#Model Concurrency FastAPI

Sebelum masuk database async, pahami dulu bagaimana FastAPI menjalankan handler — ini kunci memahami kapan harus def dan kapan async def:

Kesalahan paling umum: menulis async def tapi memanggil operasi blocking di dalamnya — seperti query database sync. Akibatnya event loop diblokir, dan semua request lain ikut tersendat. Aturan praktis:

#Async SQLAlchemy + asyncpg

Untuk PostgreSQL async, gunakan driver asyncpg dengan SQLAlchemy 2.0 async API:

pip install "sqlalchemy[asyncio]>=2.0" asyncpg

app/database_async.py

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
 
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/appdb"
 
engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)
 
 
async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

Perbedaan dari versi sync (episode 9): create_async_engine, AsyncSession, dan session dipakai sebagai async context manager. Dependency get_async_db berbentuk async generator — pola yield dari episode 8 tetap berlaku.

#CRUD Async Penuh

Handler sekarang async def dan semua operasi DB memakai await:

app/main_async.py

from typing import Annotated
 
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
 
from app import models, schemas
from app.database_async import get_async_db
 
app = FastAPI()
 
DB = Annotated[AsyncSession, Depends(get_async_db)]
 
 
@app.post("/items/", response_model=schemas.ItemOut, status_code=201)
async def create_item(item: schemas.ItemCreate, db: DB) -> models.Item:
    db_item = models.Item(**item.model_dump())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item
 
 
@app.get("/items/{item_id}", response_model=schemas.ItemOut)
async def read_item(item_id: int, db: DB) -> models.Item:
    result = await db.execute(
        select(models.Item).where(models.Item.id == item_id)
    )
    item = result.scalar_one_or_none()
    if item is None:
        raise HTTPException(status_code=404, detail="Item tidak ditemukan")
    return item

Pola yang berubah: db.commit() → await db.commit(), dan query memakai await db.execute(select(...)) lalu result.scalar_one_or_none(). Sisanya — schema, response_model, error handling — tetap sama.

#Pooling di Async Engine

create_async_engine sudah memakai connection pool secara default. Konfigurasi penting untuk produksi:

Pool configuration

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=1800,
)

• pool_size + max_overflow membatasi jumlah koneksi maksimal — hindari menghabiskan koneksi database (batas default PostgreSQL).
• pool_recycle memutus koneksi yang sudah tua, mencegah koneksi mati di balik proxy/load balancer.

Kita akan mengukur dampak pooling ini saat benchmarking di episode 25.

#Common Pitfalls

#Penutup

Inti yang harus dibawa pulang:

• BackgroundTasks menjalankan pekerjaan setelah response — untuk tugas singkat non-kritis.
• def = thread pool; async def = event loop. Jangan campur operasi blocking di handler async.
• SQLAlchemy async + asyncpg: create_async_engine, AsyncSession, semua operasi await.
• Configure pool_size, max_overflow, pool_pre_ping untuk produksi.

Di episode 15 selanjutnya kita akan membahas upload file & static files — menerima file dari klien dengan UploadFile, memvalidasi tipe dan ukuran, menyimpan file dengan aman, serta melayani static files dan media. Siapkan beberapa file contoh untuk diunggah!