Pendahuluan

FastAPI adalah framework Python modern yang sangat cepat untuk membangun API. Namun, secara default, FastAPI development server (uvicorn) hanya menjalankan satu worker process, yang berarti hanya bisa menangani satu request pada satu waktu. Untuk production dan handling concurrent requests, kita memerlukan Gunicorn sebagai process manager.

Gunicorn (Green Unicorn) adalah HTTP server untuk Python yang dapat menjalankan multiple worker processes, memungkinkan aplikasi FastAPI kita menangani banyak request secara bersamaan.

Mengapa Gunicorn?

• Multiple worker processes untuk concurrent request handling
• Load balancing otomatis antar workers
• Graceful restart tanpa downtime
• Production-ready dan battle-tested
• Kompatibel dengan uvicorn workers untuk async support

Instalasi

Pertama, install dependencies yang diperlukan:

pip install fastapi uvicorn gunicorn

Atau tambahkan ke requirements.txt:

fastapi==0.109.0
uvicorn[standard]==0.27.0
gunicorn==21.2.0

Kemudian install:

pip install -r requirements.txt

Membuat Aplikasi FastAPI Sederhana

Buat file main.py:

from fastapi import FastAPI
import time
import os

app = FastAPI()

@app.get("/")
async def root():
    return {
        "message": "Hello World",
        "workerpid": os.getpid()

    }

@app.get("/slow")
async def slowendpoint():
    # Simulasi operasi yang memakan waktu
    time.sleep(5)
    return {
        "message": "This took 5 seconds",
        "workerpid": os.getpid()

    }

@app.get("/health")
async def healthcheck():
    return {"status": "healthy"}

Setup Gunicorn dengan Uvicorn Workers

Gunicorn sendiri adalah synchronous server. Untuk mendapatkan async benefits dari FastAPI, kita menggunakan Uvicorn workers.

Menjalankan dengan Command Line

gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

• main:app - module:application (main.py dengan variable app)
• --workers 4 - jumlah worker processes
• --worker-class uvicorn.workers.UvicornWorker - gunakan async uvicorn workers
• --bind 0.0.0.0:8000 - host dan port

Membuat Configuration File

Buat file gunicornconf.py untuk konfigurasi yang lebih terstruktur:

import multiprocessing
import os

Server Socket
bind = "0.0.0.0:8000"
backlog = 2048

Worker Processes
workers = int(os.getenv("WORKERS", multiprocessing.cpucount()  2 + 1))

workerclass = "uvicorn.workers.UvicornWorker"

workerconnections = 1000
maxrequests = 10000

maxrequestsjitter = 1000

timeout = 120
keepalive = 5

Logging
accesslog = "-"
errorlog = "-"
loglevel = "info"
accesslogformat = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'


Process Naming
procname = "fastapiapp"


Server Mechanics
daemon = False
pidfile = None
user = None
group = None
tmpuploaddir = None


Graceful Timeout
gracefultimeout = 30

gunicorn main:app -c gunicornconf.py

workers = (2  CPUCORES) + 1

• 2 CPU cores = 5 workers
• 4 CPU cores = 9 workers
• 8 CPU cores = 17 workers

• Untuk CPU-bound tasks: gunakan formula di atas
• Untuk I/O-bound tasks: bisa lebih banyak workers
• Monitor memory usage, jangan sampai kehabisan RAM
• Start dengan formula standar, lalu adjust berdasarkan monitoring

Testing Concurrent Requests

Test apakah concurrent invocation bekerja dengan benar.

Menggunakan Python

Buat file testconcurrent.py:

import asyncio
import aiohttp
import time

async def makerequest(session, url, requestid):

    start = time.time()
    async with session.get(url) as response:
        data = await response.json()
        duration = time.time() - start
        print(f"Request {requestid}: PID={data.get('workerpid')} Duration={duration:.2f}s")

        return data

async def testconcurrent():
    url = "http://localhost:8000/slow"

    async with aiohttp.ClientSession() as session:
        tasks = [makerequest(session, url, i) for i in range(10)]


        starttime = time.time()
        results = await asyncio.gather(*tasks)
        totaltime = time.time() - starttime

        print(f"\nTotal time for 10 concurrent requests: {totaltime:.2f}s")

        print(f"Without concurrency would take: ~50s")

        # Check different PIDs
        pids = set(r.get('workerpid') for r in results)
        print(f"\nRequests handled by {len(pids)} different workers")
        print(f"Worker PIDs: {pids}")

if name == "main":
    asyncio.run(testconcurrent())

pip install aiohttp
python testconcurrent.py

Request 0: PID=12345 Duration=5.02s
Request 1: PID=12346 Duration=5.03s
Request 2: PID=12347 Duration=5.01s
...
Total time for 10 concurrent requests: 5.10s
Without concurrency would take: ~50s

Requests handled by 4 different workers
Worker PIDs: {12345, 12346, 12347, 12348}

Menggunakan Apache Bench

ab -n 100 -c 10 http://localhost:8000/

• -n 100 - total 100 requests
• -c 10 - 10 concurrent requests

Environment Variables

Buat file .env untuk konfigurasi:

WORKERS=4
HOST=0.0.0.0
PORT=8000
LOGLEVEL=info

Update gunicornconf.py untuk membaca env vars:

import os
from dotenv import loaddotenv


loaddotenv()

bind = f"{os.getenv('HOST', '0.0.0.0')}:{os.getenv('PORT', '8000')}"
workers = int(os.getenv('WORKERS', 4))
loglevel = os.getenv('LOGLEVEL', 'info')

[Unit]
Description=FastAPI Application
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/app
Environment="PATH=/path/to/venv/bin"
ExecStart=/path/to/venv/bin/gunicorn main:app -c gunicornconf.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Enable dan start service:

sudo systemctl daemon-reload
sudo systemctl enable fastapi
sudo systemctl start fastapi
sudo systemctl status fastapi

Menggunakan Docker

Buat Dockerfile:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "main:app", "-c", "gunicornconf.py"]

docker build -t fastapi-app .
docker run -p 8000:8000 -e WORKERS=4 fastapi-app

version: '3.8'

services:
  api:
    build: .
    ports:
"8000:8000"
    environment:

WORKERS=4
LOGLEVEL=info
    restart: unless-stopped

    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

Jalankan:

docker-compose up -d

Best Practices

1. Worker Management

# gunicornconf.py

Restart workers after handling requests (prevent memory leaks)
maxrequests = 1000
maxrequestsjitter = 50  # Randomize restart untuk avoid thundering herd

Timeout untuk long-running requests
timeout = 120

Graceful shutdown
gracefultimeout = 30

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

@app.get("/health")
async def healthcheck():
    return JSONResponse(
        statuscode=status.HTTP200OK,

        content={
            "status": "healthy",
            "workerpid": os.getpid()
        }
    )

3. Monitoring

Gunakan middleware untuk tracking:

import time
from fastapi import Request

@app.middleware("http")
async def addprocesstimeheader(request: Request, callnext):
    starttime = time.time()

    response = await callnext(request)
    processtime = time.time() - starttime
    response.headers["X-Process-Time"] = str(processtime)

    response.headers["X-Worker-PID"] = str(os.getpid())
    return response

import logging
from fastapi import FastAPI

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(name)

app = FastAPI()

@app.onevent("startup")
async def startupevent():

    logger.info(f"Worker started with PID: {os.getpid()}")

@app.onevent("shutdown")
async def shutdownevent():

    logger.info(f"Worker shutting down with PID: {os.getpid()}")

timeout = 300  # 5 minutes

maxrequests = 1000
maxrequestsjitter = 100

Issue 3: Too Many Workers

workers = 4  # Start small, monitor, then adjust

Kesimpulan

Setup Gunicorn dengan FastAPI untuk concurrent invocation meliputi:

Dengan setup yang benar, aplikasi FastAPI Anda akan siap menangani ribuan concurrent requests dengan efisien dan reliable di production environment.

• Gunicorn + Uvicorn workers = async concurrency support
• Formula workers: (2 × CPU cores) + 1
• Always test concurrent behavior sebelum production
• Monitor dan adjust configuration berdasarkan real usage
• Implement proper logging dan health checks