Belajar FastAPI - Microservices & Async Architecture

#Pendahuluan

Setelah 21 episode membangun satu aplikasi FastAPI yang solid, sekarang kita perluas cakrawala: microservices. Aplikasi yang makin besar sering dipecah menjadi beberapa service yang lebih kecil — masing-masing FastAPI, masing-masing dengan tanggung jawab dan database sendiri.

Mengapa episode ini penting? Microservices menambah kompleksitas jaringan — kalian menukar "semua dalam satu proses" dengan "banyak proses yang saling bicara". Tanpa pola komunikasi yang benar, arsitektur ini berubah menjadi benang kusut. Episode ini memberi kalian pola yang terbukti: kapan harus pecah, bagaimana bicara antar service, dan kapan harus memakai event broker.

#Kapan Memecah Aplikasi

Microservices bukan jawaban untuk semua masalah. Pola yang sehat biasanya mulai dari monolith modular — satu aplikasi FastAPI dengan router dan folder per domain — lalu memecah saat ada alasan konkret:

100%

Perhatikan perubahan penting: database ikut terpecah. Setiap service memiliki database sendiri — bukan hanya memindahkan kode.

#Komunikasi Antar Service: HTTP Sync

Cara paling langsung: service A memanggil endpoint service B via HTTP. Di FastAPI, gunakan httpx (bukan requests) agar bisa dipakai dari handler async:

Panggil service lain

import httpx
 
USERS_SERVICE = "http://users-service:8001"
 
 
async def get_user_profile(user_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{USERS_SERVICE}/users/{user_id}"
        )
    response.raise_for_status()
    return response.json()
 
 
@app.get("/orders/{order_id}/customer/")
async def get_order_with_customer(
    order_id: int,
    db: AsyncSession = Depends(get_async_db),
) -> dict:
    order = await db.get(models.Order, order_id)
    customer = await get_user_profile(order.customer_id)
    return {"order": order, "customer": customer}

Poin penting: await client.get(...) — sambil menunggu response service lain, event loop tetap melayani request lain (pola episode 14). Tambahkan timeout dan retry untuk menghadapi service yang sedang naik-turun.

#Event-Driven: Message Broker

Pola yang lebih tangguh untuk integrasi lintas domain: event-driven. Service mempublikasikan peristiwa ("order.created") ke broker, dan service lain yang peduli mendengarnya. Ini memutus dependensi langsung:

Publish event (aio-pika)

import json
 
import aio_pika
 
RABBITMQ_URL = "amqp://guest:guest@rabbitmq:5672/"
 
 
async def publish_event(routing_key: str, payload: dict) -> None:
    connection = await aio_pika.connect_robust(RABBITMQ_URL)
    async with connection:
        channel = await connection.channel()
        await channel.default_exchange.publish(
            aio_pika.Message(
                body=json.dumps(payload).encode(),
                content_type="application/json",
            ),
            routing_key=routing_key,
        )
 
 
@app.post("/orders/", status_code=201)
async def create_order(order: schemas.OrderCreate) -> dict:
    order_id = 1
    await publish_event(
        "order.created",
        {"order_id": order_id, "customer_id": order.customer_id},
    )
    return {"order_id": order_id}

order.created dipublikasikan ke RabbitMQ, lalu consumer (misal payments-service atau notifications-service) bereaksi — masing-masing berjalan independen.

#Consumer Async dengan FastAPI

Consumer bisa hidup di service yang sama — misalnya FastAPI yang menangani REST dan WebSocket sekaligus men-consume event untuk broadcast realtime (menghubungkan episode 13 dan 22):

Consume event di aplikasi

import asyncio
import json
from contextlib import asynccontextmanager
 
import aio_pika
from fastapi import FastAPI
 
 
async def consume_events() -> None:
    connection = await aio_pika.connect_robust(RABBITMQ_URL)
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue("order.created", durable=True)
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    payload = json.loads(message.body)
                    await manager.broadcast(
                        f"order baru: {payload['order_id']}"
                    )
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(consume_events())
    yield
    task.cancel()
 
 
app = FastAPI(lifespan=lifespan)

lifespan (Starlette) menjalankan consumer saat app start dan membersihkannya saat shutdown — pola modern FastAPI untuk kode yang harus hidup di balik layar.

#Pattern Inter-Service: Pilih Sesuai Kebutuhan

#Common Pitfalls

#Penutup

Inti yang harus dibawa pulang:

• Mulai dari monolith modular; pecah saat tim/skala/batas benar-benar menuntut.
• Setiap service punya database sendiri.
• Komunikasi: HTTP sync untuk data real-time, event broker (RabbitMQ/Kafka) untuk integrasi lintas domain.
• httpx async + lifespan consumer adalah toolkit dasar service FastAPI.

Di episode 23 selanjutnya kita akan membahas AI/ML integration & streaming — integrasi LLM dengan tool-calling berbasis schema Pydantic, streaming response token-by-token, dan serving model — panggung utama FastAPI di 2026!