DEV Community: Ayush Kumar

Building microservices python fastapi: design to production

Ayush Kumar — Tue, 14 Jul 2026 14:08:36 +0000

Why microservices python fastapi are a solid choice for production

If you need to spin up a set of loosely-coupled services that can evolve independently, microservices python fastapi is a pragmatic answer. FastAPI gives you async-first request handling, automatic OpenAPI docs, and type-safe validation with almost no boilerplate. Coupled with Python’s rich ecosystem for data, AI, and messaging, you can ship production-grade services in weeks instead of months.

In this post I’ll walk through the entire lifecycle: designing a FastAPI-based microservice, picking a communication pattern, wiring RabbitMQ or Kafka, containerizing with Docker Compose and Kubernetes, testing, monitoring, tracing, and locking down security. I’ll also sprinkle in the hard-earned lessons that kept me awake at 2 am.

How do I design a FastAPI-based microservice that stays maintainable?

The first step is to define a clear bounded context. Each service should own a single business capability and expose a minimal public API. I start with a folder layout that mirrors that intent:

myservice/
├── app/
│   ├── __init__.py
│   ├── api/
│   │   ├── v1/
│   │   │   └── endpoints.py
│   ├── core/
│   │   ├── config.py
│   │   └── di.py          # dependency injection helpers
│   ├── models/
│   │   └── orm.py
│   └── services/
│       └── business.py
├── tests/
│   └── test_endpoints.py
├── Dockerfile
└── pyproject.toml

api/v1/endpoints.py holds the router objects, nothing else.
services/business.py contains pure Python functions that implement the domain logic. No FastAPI imports.
core/di.py wires the dependencies (DB session, external clients) using FastAPI’s Depends.

Why separate business logic from the router? Because it lets you unit-test the core without starting an ASGI server and keeps the codebase friendly to future refactors (e.g., swapping FastAPI for another framework).

A minimal router looks like this:

# app/api/v1/endpoints.py
from fastapi import APIRouter, Depends, HTTPException, status
from ..services.business import greet_user
from ..core.di import get_db

router = APIRouter(prefix="/v1", tags=["greeting"])

@router.get("/hello/{name}", response_model=str)
async def hello(name: str, db=Depends(get_db)):
    try:
        return await greet_user(name, db)
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail=str(exc))

Notice the thin wrapper around greet_user. The service function can be called from a Celery task, a CLI script, or a test harness without any FastAPI baggage.

Lesson learned: I once let the router import the ORM models directly and ended up with circular imports that broke hot-reloading in development. Keeping the layers strictly separated avoided that mess.

Which communication pattern should I pick: REST, gRPC, or async messaging?

When is REST enough?

If the interaction is request-response and latency tolerance is in the hundreds of milliseconds, plain HTTP/JSON works fine. FastAPI’s automatic OpenAPI spec makes client generation painless.

Pros:

Human-readable payloads, easy debugging with curl.
No extra runtime (just uvicorn).

Cons:

Verbose JSON for high-frequency data.
No built-in streaming or binary support.

When does gRPC make sense?

For internal, high-throughput services that need strict contracts and binary payloads, gRPC shines. FastAPI can still expose a REST side-car while a separate gRPC server handles the heavy lifting.

Pros:

Protobuf enforces schema, reduces payload size.
Built-in code generation for many languages.

Cons:

Requires a separate server process or grpcio integration.
Debugging is less straightforward; you need grpcurl or similar.

When to use async messaging (Kafka, RabbitMQ)?

If you need eventual consistency, event sourcing, or fan-out to many consumers, an async broker is the answer. You can still expose a REST endpoint that publishes a message, but the heavy work happens downstream.

Trade-off: Async adds operational complexity (broker management, duplicate handling) and latency (typically seconds). Don’t use it for simple CRUD that needs immediate confirmation.

My rule of thumb:

Start with REST.
Add gRPC only if you hit protobuf-friendly performance limits.
Introduce a message queue when you need decoupling or replayability.

How do I implement message queues with RabbitMQ or Kafka?

FastAPI itself is agnostic; the integration lives in background workers. I prefer RabbitMQ for task queues (Celery) and Kafka for event streams.

RabbitMQ + Celery example

# app/core/celery_app.py
from celery import Celery

celery = Celery(
    "myservice",
    broker="amqp://guest:guest@rabbitmq:5672//",
    backend="redis://redis:6379/0",
)

@celery.task
def process_order(order_id: int):
    # heavy processing, DB writes, external calls
    ...

# app/api/v1/endpoints.py
from ..core.celery_app import process_order

@router.post("/orders")
async def create_order(order: OrderIn):
    # persist order synchronously
    db_order = await save_order(order)
    # fire‑and‑forget async work
    process_order.delay(db_order.id)
    return {"id": db_order.id}

Failure mode: If the Celery worker crashes after the DB commit but before the task is enqueued, you lose the async step. To mitigate, use the Outbox pattern: write an “outbox” table in the same transaction, and have a separate poller publish those rows to RabbitMQ.

Kafka producer example

# app/services/events.py
from aiokafka import AIOKafkaProducer
import json
import os

producer = AIOKafkaProducer(
    bootstrap_servers=os.getenv("KAFKA_BOOTSTRAP_SERVERS")
)

async def publish_user_created(user_id: int):
    await producer.start()
    try:
        await producer.send_and_wait(
            "user.created",
            json.dumps({"user_id": user_id}).encode("utf-8")
        )
    finally:
        await producer.stop()

Kafka guarantees ordering within a partition and retains messages for replay. The downside is higher operational overhead: you need Zookeeper/KRaft, topic configurations, and careful consumer offset management.

When NOT to use: If you only have a handful of services and can tolerate occasional coupling, adding Kafka is overkill. It also hurts latency for real-time UI updates; a WebSocket or SSE may be better.

How do I containerize and orchestrate a FastAPI microservice with Docker Compose & Kubernetes?

Dockerfile (multi-stage)

# syntax=docker/dockerfile:1
FROM python:3.12-slim AS builder
WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry export -f requirements.txt --output requirements.txt --without-hashes
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.12-slim AS runtime
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

The builder stage isolates pip install, keeping the final image tiny (≈80 MB).

Docker Compose for local dev

version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://postgres:postgres@db:5432/mydb
    depends_on:
      - db
      - rabbitmq
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
volumes:
  pgdata:

Compose gives me a one-command dev environment. I’ve been bitten by mismatched environment variables when moving to Kubernetes, so I keep a single source of truth in a .env file and reference it both locally and in the Helm chart.

Helm chart snippet for Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ include "myservice.fullname" . }}
spec:
  replicas: {{ .Values.replicaCount }}
  selector:
    matchLabels:
      app.kubernetes.io/name: {{ include "myservice.name" . }}
  template:
    metadata:
      labels:
        app.kubernetes.io/name: {{ include "myservice.name" . }}
    spec:
      containers:
        - name: api
          image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
          ports:
            - containerPort: 8000
          envFrom:
            - secretRef:
                name: myservice-secret
          readinessProbe:
            httpGet:
              path: /healthz
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10

Kubernetes gives you automated rollouts, pod health checks, and horizontal scaling. The only surprise I ran into was the default termination grace period (30 s) being too short for long-running DB migrations. I increased terminationGracePeriodSeconds to 60 s in the pod spec.

How should I test, monitor, and trace FastAPI microservices?

Unit & integration testing

I keep the business logic free of FastAPI, so a typical unit test looks like:

# tests/test_business.py
import pytest
from app.services.business import greet_user

@pytest.mark.asyncio
async def test_greet_user():
    result = await greet_user("Alice", db=None)  # db mock or None for pure logic
    assert result == "Hello, Alice!"

For endpoint tests I use httpx with FastAPI’s TestClient:

from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_hello_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        resp = await client.get("/v1/hello/Bob")
    assert resp.status_code == 200
    assert resp.json() == "Hello, Bob!"

I once faced a SQLAlchemy session leak that only manifested after hours of load. The fix is detailed in my FastAPI SQLAlchemy Session Leak Detection post, but the quick tip is: always use async with get_db() as db: in background tasks, and never store the session globally.

Monitoring & tracing

Prometheus + Grafana for metrics. FastAPI exposes /metrics through prometheus_fastapi_instrumentator.
OpenTelemetry for distributed tracing. I instrument the HTTP layer and the Kafka producer:

from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.aiokafka import AIOKafkaInstrumentor

FastAPIInstrumentor().instrument_app(app)
AIOKafkaInstrumentor().instrument()

Health checks: /healthz endpoint returns DB and broker status. Kubernetes uses it for liveness.

A common pitfall: forgetting to set OTEL_EXPORTER_OTLP_ENDPOINT in the container, which made tracing silently drop. Adding the env var to the Helm values fixed it.

Load testing

I run locust against the /v1/hello/{name} endpoint with 200 concurrent users. The latency stayed under 150 ms, well within my SLA. When I increased to 500 users, the CPU spiked to 90 % and the service started returning 502 from the ingress. The fix was to enable Uvicorn workers (uvicorn app.main:app --workers 4) and tune the Gunicorn timeout.

What are the security and authentication best practices for FastAPI microservices?

Prefer OAuth2 with JWT for inter-service auth. FastAPI’s OAuth2PasswordBearer combo works, but for service-to-service I use client credentials flow and validate the token with a shared public key.

from fastapi import Security, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt

bearer = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer)):
    try:
        payload = jwt.decode(credentials.credentials, PUBLIC_KEY, algorithms=["RS256"])
        return payload
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Scope checks: embed service name and allowed actions in the token claims, and verify them in each endpoint.
Rate limiting: Deploy Envoy as a sidecar with a token-bucket filter. It protects against accidental DoS from internal clients.
Input validation: FastAPI’s Pydantic models already reject malformed JSON, but never trust external libraries. I once let a third-party library deserialize raw JSON into a dict, opening a prototype pollution path. The fix was to keep all external data behind a Pydantic model.
Secrets management: Store DB passwords, API keys, and JWT signing keys in Kubernetes Secrets or HashiCorp Vault. Never hard-code them. In my CI pipeline I use kubectl create secret generic with --from-literal.
CORS: Only allow origins that belong to your front-end. app.add_middleware(CORSMiddleware, allow_origins=["https://myapp.com"], ...).
Patch dependencies: Run pip list --outdated in CI and automate security scans with GitHub Dependabot. A recent CVE in pyyaml forced me to upgrade across all services within a day.

FAQ

Q: Should I use async SQLAlchemy with FastAPI?

A: Yes, if your service does I/O-bound DB work. Async drivers (asyncpg) avoid thread-pool exhaustion. Just remember to close the session in a finally block or use a dependency that yields the session.

Q: When is it safe to run multiple FastAPI workers in the same pod?

A: When the service is CPU-bound or you need higher throughput. Use a process manager like Gunicorn with the uvicorn.workers.UvicornWorker. Avoid sharing in-memory caches between workers; use Redis instead.

Q: How do I avoid “QueuePool limit reached” errors?

A: Tune the SQLAlchemy pool_size and max_overflow settings, and make sure every request returns the DB connection to the pool. My write-up on fixing that error is Fixing “QueuePool limit reached”.

Q: Do I need OpenAPI docs in production?

A: They’re useful for internal debugging, but expose them only behind authentication or disable them with docs_url=None in production to reduce surface area.

Key Takeaways

Design first: isolate business logic from FastAPI routers to keep tests fast and code reusable.
Pick the right communication pattern: start with REST, add gRPC or async messaging only when the use case demands it.
Use RabbitMQ for tasks, Kafka for event streams; guard against outbox consistency issues.
Containerize with multi-stage Docker, run locally with Docker Compose, and ship to Kubernetes via Helm.
Test early, monitor with Prometheus/OpenTelemetry, and watch for session leaks (see my session-leak guide).
Secure everything: JWT verification, scope checks, rate limiting, secret management, and regular dependency audits.

Microservices python fastapi isn’t a silver bullet, but with the right patterns you can ship reliable, observable, and secure services that survive the chaos of production. Happy coding!

Why FastAPI Apps Slow Down Over Time (Low CPU, High Latency Explained)

Ayush Kumar — Tue, 14 Jul 2026 13:55:19 +0000

The "Silent Death" Scenario

You deploy your FastAPI application. For the first few hours, it feels blazing fast. Your P99 latency is sitting comfortably at 20ms. You high-five your team and go to sleep.

The next morning, your alerts are screaming.

Latency: Spiking to 3,000ms+ randomly.
Health Checks: Failing intermittently.
The Weird Part: You check your dashboards, and CPU usage is at 5%. Memory is flat. The database is responding instantly.

If your server isn't out of CPU and isn't out of memory, why is it crawling?

The answer lies in The Event Loop. When CPU is low but latency is high, it usually means your application isn't working, it is waiting.

This guide focuses on real-world scenarios where FastAPI apps slow down over time despite low CPU usage, a pattern that only appears under sustained load.

TL;DR: If your FastAPI app has low CPU but high latency, your event loop is blocked. The fix is not scaling CPUs, it’s fixing lifecycle, async misuse, and resource ownership.

Understanding the Single-Threaded Model

To fix the problem, you must first understand the machine.

FastAPI (running on Uvicorn) relies on Python's asyncio event loop. Think of the Event Loop as a single waiter in a busy restaurant.

Async: When a customer orders food (an API request), the waiter sends the order to the kitchen (Database/External API) and immediately goes to serve the next table. He doesn't wait in the kitchen. This is why FastAPI can handle thousands of requests with one thread.
Sync (Blocking): If the waiter decides to cook the food himself (Heavy Calculation) or stand in the kitchen staring at the chef (Synchronous I/O), he ignores all other tables.

If that waiter stops for just 1 second to chop an onion, every other customer waits 1 second longer. In a high-throughput system, these delays stack up exponentially.

**Mental Model**  
Think of the event loop as a single-lane highway:
- Async = cars exit quickly, highway stays free
- Blocking = one broken car stops everything behind it

Killer #1: The "Sync" Blocker (and the `def` Trap)

This is the most common mistake developers make when migrating from Flask or Django.

In FastAPI, the moment you define an endpoint with async def, you are promising FastAPI: "I will not block the main thread." FastAPI trusts you and runs that code directly on the main event loop.

If you put anything blocking inside an async def, you freeze the entire server.

The Mistake: Synchronous I/O in Async Path

Python

import time
import requests  # <--- The silent killer

@app.get("/external-data")
async def get_data():
    # BLOCKING!
    # This freezes the entire server. No other request can enter.
    response = requests.get("https://example.com/huge-data")
    time.sleep(1) # Simulating processing
    return response.json()

If 50 users hit this endpoint simultaneously, the 50th user doesn't wait 1 second. They wait 50 seconds (1s x 50). The CPU usage remains low because the CPU isn't computing - it's just idle, waiting for the network response.

The Nuance: `def` vs `async def`

FastAPI has a safety valve. If you define your endpoint with just def (no async), FastAPI assumes it might block and runs it in a separate thread pool (Starlette's run_in_threadpool).

Python

@app.get("/safe-sync")
def safe_sync():
    # Safe! FastAPI runs this in a thread pool.
    time.sleep(1)
    return {"status": "ok"}

However, you lose the benefits of async concurrency. The best solution is to keep async def but offload the blocking work.

💡
Checkout that you are using your wokers wisely, so it doesn’t slow down your requests.

The Fix: `asyncio.to_thread`

If you must use a blocking library (like requests or pandas) or perform a heavy calculation (like image processing), run it in a non-blocking way.

Python

import asyncio
import time

@app.get("/heavy-async")
async def heavy_async():
    # ✅ Runs in a separate thread, leaving the Event Loop free
    await asyncio.to_thread(time.sleep, 1)
    return {"status": "done"}

Red Flags That Your Event Loop Is Blocked

If you see any of these, suspect blocking immediately:

Health checks timing out while business endpoints are slow
Latency spikes without CPU spikes
Restarting pods “fixes” performance temporarily
Throughput collapses before memory or CPU exhaustion

Killer #2: The Hidden CPU Crushers (Password Hashing)

Sometimes, the blocker isn't I/O; it's pure CPU work.

A classic example is Password Hashing (bcrypt or argon2). Hashing a password is designed to be slow and CPU-intensive to thwart hackers.

The Mistake:

Python

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

@app.post("/login")
async def login(user: UserLogin):
    # BLOCKING! 
    # Verifying a hash takes ~300ms of pure CPU time.
    if not pwd_context.verify(user.password, user.hashed_password):
        raise HTTPException(...)
    return {"token": "..."}

If 10 users try to login at once, your event loop locks up for 3 seconds (300ms * 10). Your health check endpoint (which should take 1ms) will time out, causing Kubernetes to kill your pod.

Password hashing belongs to the request lifecycle, but not to the event loop.

The Fix: Even though this is CPU-bound, not I/O-bound, you must still move it off the main thread.

Python

@app.post("/login")
async def login(user: UserLogin):
    # Offload CPU work to a thread
    is_valid = await asyncio.to_thread(
        pwd_context.verify, user.password, user.hashed_password
    )
    if not is_valid:
        raise HTTPException(...)

Killer #3: The `httpx` Client Leak

If your API acts as a gateway or aggregator calling other services, you are likely using httpx or aiohttp. A common mistake is treating the async client like a standard variable.

HTTP clients should be treated like database engines: created once, owned by the application lifecycle, and cleaned up deterministically.

The Mistake: Creating a Client per Request

Python

@app.get("/proxy")
async def proxy_request():
    # Creates a new Connection Pool + SSL Context every time!
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://example.com")
    return resp.json()

Why this kills performance:

TCP Handshake: Every request performs a 3-way handshake.
SSL Handshake: Expensive crypto negotiation happens every time.
File Descriptors: You will eventually run out of sockets (OSError: [Errno 24] Too many open files).

The Fix: The Lifespan Pattern Create one client at startup and reuse it. The connection pool stays open, making subsequent requests lightning fast.

Python

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize one client for the whole app
    app.state.http_client = httpx.AsyncClient()
    yield
    # Cleanup on shutdown
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/proxy")
async def proxy_request(request: Request):
    client = request.state.http_client
    resp = await client.get("https://example.com")
    return resp.json()

Killer #4: The "Zombie" Background Task

FastAPI provides BackgroundTasks, which are great. But often, developers want "fire and forget" logic using standard AsyncIO.

The Mistake:

Python

@app.post("/process")
async def process():
    # DANGER! 
    # The Python Garbage Collector might destroy this task 
    # immediately because no variable references it.
    asyncio.create_task(long_running_job()) 
    return {"status": "started"}

Not only can this task fail silently, but if it hangs, you have no way to cancel it or monitor it.

The Fix: The Set-Reference Pattern You must maintain a "Hard Reference" to your running tasks.

Python

# A global set to hold references
active_tasks = set()

async def safe_process():
    task = asyncio.create_task(long_running_job())

    # 1. Add to set (keep it alive)
    active_tasks.add(task)

    # 2. Remove from set when done (prevent memory leak)
    task.add_done_callback(active_tasks.discard)

Why Adding More CPU or Pods Won’t Save You

Auto-scaling hides event loop bugs, it doesn’t fix them.

When the loop is blocked:

Each new pod reproduces the same problem
Load balancers distribute latency, not relief
Costs go up while performance stays bad

Until the event loop is unblocked, scaling is cosmetic.

How to Debug: Finding the Invisible Wall

You can't see these issues with standard access logs because the logs only show total time. They don't show waiting time.

You need a sampling profiler. The best tool for Python is py-spy. It reads memory directly, so it has almost zero overhead and works on production servers.

1. Install py-spy:

Bash

pip install py-spy

2. Peek at the running process: Find your uvicorn Process ID (PID) and run:

Bash

# --rate 100: samples 100 times per second
# --subprocesses: checks worker processes too
sudo py-spy top --pid <PID> --rate 100 --subprocesses

3. Interpret the Output:

If you see time.sleep, select, or requests at the top of the list: You are blocking the loop.
If you see _run_until_complete dominating without specific function calls: You might be starving the loop with too many tasks.

Conclusion

Low CPU + High Latency is almost always a symptom of a Blocked Event Loop.

FastAPI apps don’t decay randomly.

They slow down because something is blocking the event loop.

Once you understand resource ownership, what runs on the loop, what must not, and what belongs in the application lifespan performance issues stop being mysterious.

If you’re also seeing database pool exhaustion or connection timeouts, read my guide on Fixing QueuePool Overflow Errors in FastAPI, where lifecycle bugs surface again in a different form.

To keep your FastAPI app responsive:

Audit your async def endpoints: Ensure no requests, time.sleep, or heavy math exists there.
Use asyncio.to_thread: For anything blocking or CPU-intensive (like password hashing).
Reuse HTTP Clients: Treat httpx.AsyncClient like a database connection - create it once, use it everywhere.

If you are struggling with database connection pool limits (another common cause of latency), read my guide on Fixing QueuePool Overflow errors.

Welcome to LogicLoopTech

Ayush Kumar — Tue, 14 Jul 2026 13:55:13 +0000

LogicLoopTech is a backend engineering blog. The focus is practical, hands-on content for developers building real systems with Python, FastAPI, AI, and RAG.

What you'll find here

How AI tooling actually helps backend developers - not the marketing version
RAG architectures that work in production, not just in demos
FastAPI patterns for shipping reliable APIs
Debugging guides for the kinds of bugs that eat your afternoon
Honest takes on what's worth using and what isn't

Why this exists

Most backend content is either too shallow ("here's a hello world") or too academic. The goal here is the middle ground: enough depth to actually use, short enough to read in one sitting.

New articles drop every two days. Let's go.

Why Uvicorn Health Checks Fail Under Load (And How to Fix It Properly)

Ayush Kumar — Tue, 14 Jul 2026 13:54:35 +0000

You deploy your FastAPI app to Kubernetes (or ECS).

You configure a liveness probe on /health.

Everything looks green.

Then traffic spikes.

Pods start restarting.

CrashLoopBackOff.

You check logs:

No Python exceptions
No OOM
CPU high, but not maxed
Memory stable

So why did Kubernetes kill a healthy process?

Because your application was too busy working to answer one simple question:

“Are you alive?”

The Real Problem: Event Loop Starvation

Uvicorn runs your FastAPI app on a single-threaded event loop.

That means every request - including your health check - waits its turn.

Think of it this way:

You have one cashier.

Customer A → heavy DB query
Customer B → CPU-heavy image processing
Customer C → Kubernetes asking “Are you alive?”

If the cashier is busy, Customer C waits.

Kubernetes waits 1 second.

Then 2 seconds.

Timeout.

Kubernetes assumes the process is dead.

It kills the pod.

Your app wasn’t dead.

It was just starved.

This failure pattern is extremely common in async Python services running behind Kubernetes.

If your readiness probes start timing out randomly, it might not be Kubernetes - your connection pool could be exhausted by leaked sessions. Here’s how to detect and fix them.

Mistake #1: “Deep” Health Checks

This is the most common failure pattern.

@app.get("/health")
async def health_check(db: Session = Depends(get_db)):
    user = db.query(User).first()
    return {"status": "ok"}

Looks harmless.

It isn’t.

If your DB pool is exhausted or under pressure:

get_db() waits for a connection
request hangs
health check times out
Kubernetes kills your pod

Now your database hiccup becomes a full web server restart.

That’s not health monitoring. That’s self-sabotage.

The Correct Design: Split Probes

Kubernetes gives you two tools:

1- Liveness Probe

Question: “Is the process alive?”

It should:

return instantly
not touch the DB
not call Redis
not do logic

@app.get("/health/live")
async def liveness():
    return {"status": "alive"}

That’s it.

If this fails, your process is truly frozen.

2- Readiness Probe

Question: “Can this instance handle traffic right now?”

This is where you check:

DB connectivity
Redis
downstream services

@app.get("/health/ready")
async def readiness(db: Session = Depends(get_db)):
    try:
        db.execute(text("SELECT 1"))
        return {"status": "ready"}
    except Exception:
        raise HTTPException(status_code=503)

If readiness fails:

Kubernetes stops routing traffic
Pod stays alive
No restart

That’s the correct behavior.

Mistake #2: Aggressive Probe Settings

Many clusters default to:

timeoutSeconds: 1
failureThreshold: 3

That means:

3 seconds of event loop delay = pod death.

That’s too aggressive for Python.

A safer baseline:

livenessProbe:
  httpGet:
    path: /health/live
    port: 8000
  initialDelaySeconds: 10
  periodSeconds: 10
  timeoutSeconds: 5
  failureThreshold: 5

Give your app breathing room.

Python has:

GC pauses
I/O jitter
occasional event loop delays

You don’t want restarts because of micro-stalls.

If your event loop is getting blocked, I broke this down deeply here.

Mistake #3: Blocking the Event Loop

Even a perfect /health/live can fail if something else is blocking the loop.

Common culprits:

time.sleep() inside async def
heavy password hashing
large JSON parsing
CPU-bound work
sync HTTP calls (requests.get)

If the event loop is blocked for 5 seconds, your health check never runs.

The timeout is a symptom.

The disease is blocking code.

Optional Strategy: Use `def` Instead of `async def`

FastAPI runs normal def endpoints in a thread pool.

For health endpoints specifically:

@app.get("/health/live")
def liveness():
    return {"status": "alive"}

This can sometimes isolate it from event loop stalls.

But if your process is truly saturated, this won’t save you.

It’s a mitigation, not a cure.

Incorrect worker counts can amplify probe failures and here’s the math.

The Nuclear Option: Separate Health Port

In high-load systems, some teams run health checks on a separate port using a minimal HTTP server.

Why?

Even if Uvicorn is saturated, the OS can still answer on another thread.

This is rare and usually unnecessary, but in extreme environments, it removes probe-related restarts entirely.

What Causes Liveness Probe Failures in FastAPI?

Most probe failures are not crashes infact they are timeout failures caused by event loop starvation.

The Real Lesson

When pods restart under load:

It’s usually not a crash.

It’s starvation.

Your application was alive, it was just unable to respond fast enough.

Final Checklist

✔ Split /live and /ready

✔ Keep /live dumb and instant

✔ Don’t put DB logic in liveness

✔ Increase timeoutSeconds

✔ Remove blocking code

✔ Scale workers if needed

Health checks should detect death, they should not cause it.

Symptoms You’ll Notice Before Pods Restart

Random pod restarts during traffic spikes
No Python errors
Health endpoint timing out
CPU elevated but not maxed
Latency spikes before restart

Using the OpenAI Agents SDK for Python: A Practical Guide

Ayush Kumar — Tue, 14 Jul 2026 13:54:30 +0000

Quick answer: what is the OpenAI Agents SDK for Python and how do I get it running?

The OpenAI Agents SDK for Python is a thin wrapper that lets you create autonomous agents, give them tool-access, and keep state between calls without writing boilerplate HTTP code yourself. Install it with pip, configure your API key, and you can spin up a simple agent in a single script. Below is a step-by-step walk through from installation to production deployment.

Installing and configuring the OpenAI Agents SDK in a Python environment

How do I install the SDK and make sure it works with my existing virtual environment?

You install the package from PyPI and point it at your OpenAI key. A clean virtualenv or Conda env keeps dependencies isolated, which saves you from version clashes later on.

# Create an isolated environment (optional but recommended)
python -m venv .venv
source .venv/bin/activate   # on Windows use `.venv\Scripts\activate`

# Install the SDK and a couple of helpers
pip install openai agents-sdk python-dotenv

Create a .env file at the project root:

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

Load the key in your code:

from dotenv import load_dotenv
import os

load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")
assert openai_api_key, "Set OPENAI_API_KEY in .env"

The SDK reads the key from the environment automatically, so you can also skip the manual assignment and just rely on os.getenv.

What can bite you during installation?

On Alpine-based Docker images the default gcc and musl-dev packages are missing, causing the pydantic build to fail. Adding apk add --no-cache gcc musl-dev to your Dockerfile solves it. In production I’ve also hit rate-limit errors when the key isn’t scoped for the chat.completions endpoint; double-check the permissions in the OpenAI dashboard.

Building your first OpenAI agent with tool use

Can I make an agent that calls external APIs without writing HTTP code each time?

Yes. The SDK lets you declare tools – small Python callables that the LLM can invoke. Here’s a minimal “weather” agent:

from openai import OpenAI
from agents_sdk import Agent, Tool

client = OpenAI(api_key=openai_api_key)

def get_weather(city: str) -> str:
    """Return a fake weather report for the given city."""
    # In a real app you would call a weather API here.
    return f"The weather in {city} is sunny, 24°C."

weather_tool = Tool(
    name="get_weather",
    description="Fetches the current weather for a city.",
    func=get_weather,
    parameters={"city": {"type": "string", "description": "Name of the city"}}
)

agent = Agent(
    client=client,
    system_prompt="You are a helpful assistant that can look up weather.",
    tools=[weather_tool],
)

# One‑shot interaction
response = agent.run("What's the weather in Berlin?")
print(response.content)

When the model decides it needs the get_weather tool, the SDK automatically calls get_weather and injects the result back into the conversation. The user sees a seamless answer.

When does this break?

If the tool raises an exception, the SDK will return an error message to the LLM, which may try the tool again in a loop. Wrap your functions in a try/except block and return a friendly error string instead of bubbling up the exception.

Managing agent memory and state across interactions

Do agents remember previous turns automatically?

No. The SDK treats each run call as a fresh conversation unless you pass a memory object. The simplest pattern is to keep a list of messages in a FastAPI dependency or a Redis list.

from typing import List
from agents_sdk import Message

class SimpleMemory:
    def __init__(self):
        self.history: List[Message] = []

    def add(self, msg: Message):
        self.history.append(msg)
        # Keep only the last N messages to avoid token bloat.
        if len(self.history) > 20:
            self.history = self.history[-20:]

    def get(self) -> List[Message]:
        return self.history

memory = SimpleMemory()

When you call agent.run, feed the stored messages:

response = agent.run(
    "Tell me a joke about cats.",
    messages=memory.get()
)
memory.add(response)
print(response.content)

Trade-offs: Storing full history gives richer context but inflates token usage and latency. In high-throughput services I cap the window to the last 5–10 messages or summarise older turns with a summarisation tool.

What if I need persistence across restarts?

Serialize the message list to a database (PostgreSQL, DynamoDB, etc.) or a KV store like Redis. Remember to strip out any ToolResult objects that contain large payloads; keep only the textual representation.

Adding custom tools and functions to agents

How do I expose my own business logic to the LLM?

Define a Tool with a Python callable and a JSON schema for its parameters. The SDK validates the incoming arguments before calling the function, which protects you from malformed data.

def calculate_tax(income: float, state: str) -> float:
    """Calculate estimated tax based on income and US state."""
    # Very naive tax logic, just for demo.
    rates = {"CA": 0.09, "NY": 0.08, "TX": 0.0}
    rate = rates.get(state.upper(), 0.05)
    return round(income * rate, 2)

tax_tool = Tool(
    name="calculate_tax",
    description="Estimate state tax for a given income.",
    func=calculate_tax,
    parameters={
        "income": {"type": "number", "description": "Annual income in USD"},
        "state": {"type": "string", "description": "Two‑letter state code"},
    },
)
agent.tools.append(tax_tool)

Now the agent can answer queries like “How much tax would I owe on $120k in CA?” without you writing any prompt engineering.

Failure modes: If the schema is wrong (e.g., missing a required field), the SDK will reject the call before it reaches your code. Double-check the JSON schema against the function signature. Also watch out for long-running tools – the SDK has a default 30-second timeout. For heavy computation, offload to a background worker (Celery, RQ) and return a placeholder message that the user can poll later.

Deploying and scaling OpenAI agents with FastAPI or serverless platforms

Can I serve agents behind FastAPI and still keep low latency?

Absolutely. The SDK is async-compatible, so you can write an endpoint that reuses a single Agent instance per worker process. Here’s a minimal FastAPI app:

# app/main.py
import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from agents_sdk import Agent, Message
from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

app = FastAPI()
agent = Agent(
    client=client,
    system_prompt="You are a helpful assistant with access to custom tools.",
    tools=[],
)

class ChatRequest(BaseModel):
    user_message: str
    session_id: str | None = None

# Simple in‑memory session store (replace with Redis in prod)
_sessions: dict[str, list[Message]] = {}

def get_memory(session_id: str | None):
    if not session_id:
        return []
    return _sessions.get(session_id, [])

@app.post("/chat")
async def chat(req: ChatRequest):
    memory = get_memory(req.session_id)
    response = await agent.run(req.user_message, messages=memory)
    # Update memory
    if req.session_id:
        _sessions.setdefault(req.session_id, []).extend([Message(role="user", content=req.user_message), response])
    return {"reply": response.content}

Run with uvicorn app.main:app --workers 4. Each worker holds its own Agent instance, which reuses the underlying HTTP connection pool to OpenAI.

Scaling beyond a single VM?

Serverless (AWS Lambda, Cloud Functions): The SDK works fine, but cold starts add ~200 ms overhead. Keep the Agent object at module level so it survives warm invocations.
Kubernetes: Deploy the FastAPI container with a horizontal pod autoscaler (HPA) based on CPU or request latency. Remember to set OPENAI_API_KEY as a secret and limit outbound traffic to the OpenAI endpoint to avoid unexpected egress costs.

Cost considerations: Each LLM call charges per token. If you keep a large conversation history, token usage can balloon. Trim the history or use a summarisation tool after every N turns. Also watch the number of tool invocations – each call still counts as a separate LLM request.

When NOT to use the SDK: If you need ultra-low latency (<30 ms) or run in an environment without internet access, the SDK’s reliance on remote OpenAI models makes it unsuitable. In those cases you’d need an on-prem LLM or a distilled model served locally.

Frequently asked questions

How do I debug a tool that the LLM refuses to call?

Enable SDK logging by setting AGENTS_SDK_LOG_LEVEL=debug. The logs show the model’s thought process and the exact JSON payload it tried to send. Often the issue is a mismatched parameter name or a missing required field.

Can I use the SDK with multiple OpenAI models simultaneously?

Yes. Pass a different model argument when constructing each Agent. For example, Agent(..., model="gpt-4o-mini") for cheap tasks and Agent(..., model="gpt-4o") for high-quality responses.

What’s the best way to store session memory for a large user base?

Redis is the go-to solution: store a list of serialized Message objects under a key like session:{session_id}. Set an expiration (e.g., 24 h) to avoid unbounded growth. If you already use PostgreSQL for other data, a JSONB column works too, but Redis gives you sub-millisecond reads.

Does the SDK support streaming responses?

Yes. Call agent.run(..., stream=True) and iterate over the async generator. This is useful for chat UIs that want to display partial results. Remember to forward the stream chunks to the client without buffering the whole response.

Key Takeaways

The OpenAI Agents SDK for Python removes most of the boilerplate needed to build tool-aware agents.
Install with pip install agents-sdk, load your API key from .env, and you’re ready to write a single-function agent.
Use Tool objects to expose internal services; validate parameters with JSON schemas to avoid runtime errors.
Manage state with a lightweight memory class or a persistent store like Redis; prune history to control token costs.
Deploy with FastAPI for low-latency HTTP endpoints, or use serverless functions for bursty traffic; keep the Agent instance at module level to reuse connections.
Watch for hidden costs: long conversation histories, frequent tool calls, and cold-start latency in serverless environments.

By following the patterns above you can move from a notebook prototype to a production-grade AI assistant that lives alongside your existing FastAPI services, scales with Kubernetes, and stays within a predictable budget. Happy building!

Supercharge Your API: FastAPI asyncpg Integration for High‑Performance PostgreSQL Access

Ayush Kumar — Tue, 14 Jul 2026 13:53:52 +0000

Introduction

If you’re building a modern Python web service, FastAPI asyncpg is the duo you want to master. FastAPI gives you an ultra‑fast, type‑safe HTTP layer, while asyncpg delivers low‑overhead, fully asynchronous PostgreSQL access. Together they let you serve thousands of requests per second with minimal latency. In this post we’ll walk through setting up asyncpg inside a FastAPI project, manage connection pools correctly, hook asyncpg into SQLAlchemy 2.0’s async engine, handle transactions robustly, and finally squeeze out every last drop of performance for production deployments.

Note: This guide assumes Python 3.9+, FastAPI 0.100+, asyncpg 0.28+, and SQLAlchemy 2.0. If you’re still on the classic SQLAlchemy ORM, check out our article on Fix AsyncSession Errors in FastAPI (SQLAlchemy 2.0 Guide).

Setting up asyncpg in a FastAPI project

1. Install the dependencies

pip install fastapi uvicorn asyncpg

If you plan to use SQLAlchemy later, also install:

pip install sqlalchemy[asyncio] psycopg2-binary

2. Minimal FastAPI app using asyncpg

Create app/main.py:

from fastapi import FastAPI, HTTPException
import asyncpg
import os

app = FastAPI()

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://postgres:password@localhost:5432/mydb"
)

# We'll fill the pool in the startup event (see next section)
db_pool: asyncpg.Pool | None = None


@app.on_event("startup")
async def startup():
    global db_pool
    db_pool = await asyncpg.create_pool(DATABASE_URL, min_size=5, max_size=20)
    # Test connection
    async with db_pool.acquire() as conn:
        await conn.execute("SELECT 1")


@app.on_event("shutdown")
async def shutdown():
    global db_pool
    if db_pool:
        await db_pool.close()


@app.get("/users/{user_id}")
async def read_user(user_id: int):
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, username, email FROM users WHERE id = $1", user_id
        )
        if not row:
            raise HTTPException(status_code=404, detail="User not found")
        return dict(row)

Run it:

uvicorn app.main:app --reload

You now have a FastAPI asyncpg endpoint that pulls a user record without ever blocking the event loop.

Why asyncpg?

Native async/await – built on libpq’s asynchronous API, no thread‑pool shim.
Prepared statements – automatic caching of query plans.
Fast serialization – returns Record objects that map cleanly to dicts.

Connection pooling and lifecycle management with FastAPI

Creating a new PostgreSQL connection per request is a serious performance killer. asyncpg’s built‑in connection pool solves this, but you need to tie its lifecycle to FastAPI’s startup/shutdown events.

Pool configuration best practices

Parameter	Typical value	Reason
`min_size`	5	Keep a few warm connections ready for bursts.
`max_size`	20‑50 (depends on DB limits)	Prevent exhausting DB resources under load.
`max_queries`	5000	Re‑prepare statements after this many queries.
`max_inactive_connection_lifetime`	300	Recycle idle connections to avoid stale sockets.

db_pool = await asyncpg.create_pool(
    DATABASE_URL,
    min_size=5,
    max_size=30,
    max_queries=5000,
    max_inactive_connection_lifetime=300,
    command_timeout=60,
)

Using a Dependency for Per‑request Access

FastAPI’s dependency system lets you inject a connection automatically:

from fastapi import Depends

async def get_connection() -> asyncpg.Connection:
    async with db_pool.acquire() as conn:
        yield conn

@app.get("/posts")
async def list_posts(conn: asyncpg.Connection = Depends(get_connection)):
    rows = await conn.fetch("SELECT id, title FROM posts ORDER BY created_at DESC LIMIT 20")
    return [dict(r) for r in rows]

The async with inside the dependency guarantees the connection is released back to the pool even if the endpoint raises an exception.

Detecting leaks

If you notice “QueuePool limit reached” errors, you probably have connections not returning to the pool. The article Fixing “QueuePool limit reached”: Debugging DB Connection Leaks in FastAPI explains how to instrument asyncpg’s pool statistics to spot leaks early.

Using asyncpg with SQLAlchemy 2.0 async engine

SQLAlchemy 2.0 introduced a first‑class async API that can sit on top of asyncpg. This gives you the expressive ORM while retaining raw‑SQL performance where needed.

1. Install the async driver

SQLAlchemy will automatically use asyncpg when you specify the postgresql+asyncpg dialect.

pip install sqlalchemy[asyncio] asyncpg

2. Configure the async engine

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL_ASYNC = (
    "postgresql+asyncpg://postgres:password@localhost:5432/mydb"
)

engine = create_async_engine(
    DATABASE_URL_ASYNC,
    pool_size=20,
    max_overflow=10,
    future=True,  # enables 2.0 style usage
)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
Base = declarative_base()

3. Define a model

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)

4. Dependency that yields an async session

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

5. Mixing raw asyncpg and ORM

You can still run raw asyncpg queries inside the same request:

@app.get("/users/{user_id}/stats")
async def user_stats(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    conn: asyncpg.Connection = Depends(get_connection),  # from previous section
):
    # ORM part
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Raw asyncpg part – heavy analytics query
    stats = await conn.fetchrow(
        """
        SELECT count(*) AS post_count,
               avg(like_count) AS avg_likes
        FROM posts
        WHERE author_id = $1
        """,
        user_id,
    )
    return {
        "user": {"id": user.id, "username": user.username},
        "stats": dict(stats),
    }

This hybrid approach keeps the codebase flexible: use the ORM for everyday CRUD, and asyncpg for high‑performance analytics.

Transaction handling and error management

When you work with asyncpg directly, you must manage transactions explicitly. SQLAlchemy’s AsyncSession does it for you, but you still need to be aware of proper patterns.

Using asyncpg transaction objects

@app.post("/transfer")
async def transfer_funds(
    from_user: int,
    to_user: int,
    amount: float,
    conn: asyncpg.Connection = Depends(get_connection),
):
    async with conn.transaction():
        # Debit
        await conn.execute(
            """
            UPDATE accounts
            SET balance = balance - $1
            WHERE user_id = $2 AND balance >= $1
            """,
            amount,
            from_user,
        )
        # Credit
        await conn.execute(
            """
            UPDATE accounts
            SET balance = balance + $1
            WHERE user_id = $2
            """,
            amount,
            to_user,
        )
    return {"status": "success"}

If any statement raises an exception, the async with conn.transaction() block rolls back automatically.

Nesting and savepoints

PostgreSQL supports savepoints, which asyncpg exposes via transaction() with nested=True:

async with conn.transaction():
    await conn.execute("INSERT INTO orders ...")
    async with conn.transaction(nested=True):
        # This can be rolled back independently
        await conn.execute("INSERT INTO order_items ...")
        # raise some error -> only inner transaction rolls back

SQLAlchemy async transaction pattern

@app.post("/orders")
async def create_order(
    order_data: OrderCreate,
    db: AsyncSession = Depends(get_db),
):
    async with db.begin():
        new_order = Order(**order_data.dict())
        db.add(new_order)
        # Flush to get the primary key
        await db.flush()
        # Add line items
        for item in order_data.items:
            db.add(OrderItem(order_id=new_order.id, **item.dict()))
    # Commit happens automatically when exiting the `begin` block
    return {"order_id": new_order.id}

If any exception bubbles out, the transaction is rolled back.

Centralized error handling

FastAPI lets you register exception handlers. For DB‑related errors you can translate asyncpg.PostgresError into clean HTTP responses:

from fastapi import Request
from fastapi.responses import JSONResponse
import asyncpg

@app.exception_handler(asyncpg.PostgresError)
async def pg_error_handler(request: Request, exc: asyncpg.PostgresError):
    return JSONResponse(
        status_code=400,
        content={"detail": f"Database error: {exc.message}"}
    )

Now any uncaught asyncpg error becomes a 400 response rather than a 500.

Performance tuning and best practices for production deployments

1. Use prepared statements wisely

asyncpg automatically prepares statements after the first execution. For hot paths you can pre‑prepare them during startup:

async def preload_statements():
    async with db_pool.acquire() as conn:
        await conn.prepare(
            "SELECT id, username FROM users WHERE id = $1"
        )
        await conn.prepare(
            "INSERT INTO logs (level, message) VALUES ($1, $2)"
        )

Call await preload_statements() inside the startup event.

2. Leverage `COPY` for bulk loads

If you need to ingest large CSVs, asyncpg’s copy_records_to_table is dramatically faster than row‑by‑row inserts.

import csv
from pathlib import Path

async def bulk_insert_users(csv_path: Path):
    async with db_pool.acquire() as conn:
        with csv_path.open() as f:
            reader = csv.DictReader(f)
            await conn.copy_records_to_table(
                "users",
                records=[(r["username"], r["email"]) for r in reader],
                columns=("username", "email")
            )

3. Monitor pool metrics

Add a simple endpoint to expose pool statistics (useful with Prometheus):

@app.get("/db/pool-stats")
async def pool_stats():
    return {
        "size": db_pool.get_size(),
        "used": db_pool.get_used_size(),
        "free": db_pool.get_free_size(),
        "max_size": db_pool.max_size,
    }

4. Tune PostgreSQL parameters

max_connections should be at least pool.max_size + some headroom.
shared_buffers = 25‑30 % of RAM.
effective_cache_size = 50‑75 % of RAM.

5. Deploy with an ASGI server that respects async I/O

Use uvicorn with the --workers flag only when you have CPU‑bound work; for pure async I/O a single worker is often enough. Example Dockerfile snippet:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Scale horizontally with a container orchestrator (Kubernetes, Docker Swarm). Each replica gets its own pool, so keep max_connections in mind.

6. Use connection‑level statement timeouts

await conn.execute(
    "SET statement_timeout = 5000"  # 5 seconds
)

This prevents runaway queries from hogging a connection.

7. Periodic health checks

Kubernetes liveness probes can run a lightweight SELECT 1 against the pool:

livenessProbe:
  exec:
    command: ["python", "-c", "import asyncpg, asyncio, os; asyncio.run(asyncpg.connect(os.getenv('DATABASE_URL')).execute('SELECT 1'))"]
  initialDelaySeconds: 10
  periodSeconds: 30

Key Takeaways

FastAPI asyncpg gives you a truly non‑blocking stack for PostgreSQL.
Create a single asyncpg pool at startup, release connections via async with, and expose it through FastAPI dependencies.
When you need an ORM, plug asyncpg into SQLAlchemy 2.0’s async engine (postgresql+asyncpg). You can mix raw asyncpg queries with ORM models seamlessly.
Manage transactions explicitly with conn.transaction() or AsyncSession.begin(). Use nested transactions for fine‑grained rollbacks.
Tune the pool (min_size, max_size, max_queries), pre‑prepare hot statements, and monitor pool metrics to avoid leaks.
Align PostgreSQL server settings (max_connections, shared_buffers) with your pool size, and use container orchestration health checks for resiliency.

By following these patterns, your FastAPI services will handle PostgreSQL traffic at lightning speed while staying robust, maintainable, and ready for production scale. Happy coding!

How to Setup a Firebase Account for FCM Notifications: A Step-by-Step Guide

Ayush Kumar — Tue, 14 Jul 2026 13:53:46 +0000

Before you can send a single push notification to a user's device, you need infrastructure. You need a service that can act as the post office, accepting messages from your backend and delivering them to Android, iOS, or Web apps.

That service is Firebase Cloud Messaging (FCM).

For backend developers, setting up Firebase is the non-negotiable first step. It’s where you get the credentials - like the Project ID and Service Account Key - that your Python, Node.js, or Go backend will use to authenticate against Google's servers.

In this guide, I will walk you through the exact steps to create a Firebase project and get the critical configuration files you need to start building your notification engine.

Prerequisites

A Google Account (Gmail, Workspace, etc.).

🧐 The Basics: What is Firebase & FCM?

If you are new to cloud development, these acronyms might sound confusing. Let's break them down.

1. What is Firebase? Think of Firebase as a "Backend-as-a-Service" (BaaS) platform acquired by Google. It provides a suite of tools - like databases, authentication, hosting, and analytics - that let you build apps faster without managing physical servers. instead of building a login system from scratch, you just "plug in" Firebase Auth.

2. What is FCM (Firebase Cloud Messaging)? FCM is the specific tool within the Firebase suite used for Notifications.

Imagine you have a messaging app. When User A sends a message to User B, how does User B's phone know to buzz? You can't just keep the phone's screen on forever waiting for data. FCM acts as a global post office. Your backend server hands the message to FCM, and FCM handles the complex job of waking up User B's device (whether it's an iPhone, Android, or Web Browser) and delivering the notification efficiently.

Why do developers use it?

It's Free: You can send unlimited messages at no cost.
It's Cross-Platform: You write one backend code, and FCM handles the differences between Apple's system (APNs) and Android's system.
It's Reliable: Google processes hundreds of billions of messages per day through this infrastructure.

Step 1: Create a New Firebase Project

First, we need to create a container in Google's cloud for your application's data and services.

Navigate to the Firebase Console.
Click on the big "Create a project" (or "Add project") card.

Enter a project name. This can be anything, like LogicLoop-Demo or FCM-Test-App. Firebase will automatically generate a unique unique identifier below it. Accept the terms and click Continue.
Google Analytics: For a pure backend notification setup, you can disable Google Analytics for now to keep things simple. Click Create project.
Wait a few seconds for the setup to finish, then click Continue. You will be taken to your new project's overview dashboard.

Step 2: Navigate to Project Settings

Right now, you have a project, but no credentials. As a backend developer, you don't necessarily need to register an iOS or Android app yet. You just need the server-side keys.

On the left-hand sidebar menu, click the gear icon (⚙️) next to "Project Overview".
Select Project settings from the popup menu.

This page contains all the meta-data about your project. Make a note of your Project ID, as you will often need it for SDK integration.

Step 3: Get the FCM Service Account Key (The Important Part)

This is the most critical step. To send notifications from a backend server (like a Python FastAPI service), you need secure credentials.

Historically, Firebase used a "Server Key". This is now considered the legacy API. The modern, secure way is to use the Firebase Admin SDK with a Service Account.

In the Project settings menu, click on the Service accounts tab.
You will see a code snippet showing how to use the Admin SDK in various languages (Node.js, Java, Python, etc.).
Below that, look for a button labeled Generate new private key.

A warning window will popup. Read it - it tells you to keep this key safe! Click Generate key.
A .json file will automatically download to your computer.

⚠️ STOP AND SECURE THIS FILE ⚠️

This JSON file contains your project's private key. Never commit this file to a public GitHub repository. Anyone who has this file can send notifications on your behalf and control your Firebase project. Store it securely on your local machine or use a secrets manager in production.

Conclusion

Congratulations! You have successfully laid the foundation.

✅ You created a Firebase Project.
✅ You navigated the console to find project settings.
✅ You generated and downloaded the secure Service Account JSON key.

Your backend now has the "passport" it needs to talk to Google's servers. The next step is to write the code that uses this key to actually send a message. We will cover that in the next tutorial.

Serverless Python: Deploying FastAPI to Google Cloud Run with Docker

Ayush Kumar — Tue, 14 Jul 2026 13:53:09 +0000

Introduction

In my previous article, we built a high-performance Async API using FastAPI and SQLAlchemy 2.0.

But code that only lives on localhost provides zero value.

Today, we are moving that API to production. We will containerize our application using Docker and deploy it to Google Cloud Run - a fully managed serverless platform that scales automatically.

Why Cloud Run?

Scale to Zero: You don't pay for servers when no one is using your API.
Concurrency: Unlike AWS Lambda (which handles 1 request per instance), Cloud Run can handle 80+ requests per container concurrently.
Portability: Since it's just a Docker container, you can move it to AWS or Azure later if you want.

Step 1: The Dockerfile

The first step is packaging our Python environment. Create a file named Dockerfile (no extension) in your project root.

We will use a multi-stage build or a slim image to keep our container lightweight.

Dockerfile

# 1. Use an official lightweight Python image
FROM python:3.11-slim

# 2. Set environment variables
# PYTHONDONTWRITEBYTECODE: Prevents Python from writing .pyc files
# PYTHONUNBUFFERED: Ensures logs are flushed directly to terminal (critical for Cloud Logging)
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# 3. Set the working directory
WORKDIR /app

# 4. Install dependencies
# We copy requirements.txt first to leverage Docker cache
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 5. Copy the application code
COPY . .

# 6. Expose the port (Cloud Run defaults to 8080)
EXPOSE 8080

# 7. Run the application
# Host 0.0.0.0 is required for the container to be accessible from outside
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

Pro Tip: Make sure you have a requirements.txt file. You can generate one with pip freeze > requirements.txt.

Step 2: Build & Test Locally

Before pushing to the cloud, always test the container locally.

Bash

# Build the image
docker build -t logicloop-api .

# Run the container
docker run -p 8080:8080 logicloop-api

Go to http://localhost:8080/docs. If you see your Swagger UI, your container is valid.

Step 3: Push to Google Artifact Registry

Cloud Run pulls images from a registry. We will use Google's Artifact Registry (the successor to Container Registry).

1. Create a Repository (One time setup):

Bash

gcloud artifacts repositories create my-repo \
    --repository-format=docker \
    --location=us-central1 \
    --description="Docker repository"

2. Tag your image: Replace PROJECT_ID with your actual Google Cloud Project ID.

Bash

docker tag logicloop-api us-central1-docker.pkg.dev/PROJECT_ID/my-repo/logicloop-api:v1

3. Push the image:

Bash

docker push us-central1-docker.pkg.dev/PROJECT_ID/my-repo/logicloop-api:v1

Step 4: Deploy to Cloud Run

Now for the magic command. This tells Google to take that image and turn it into a live HTTPS endpoint.

Bash

gcloud run deploy logicloop-backend \
    --image us-central1-docker.pkg.dev/PROJECT_ID/my-repo/logicloop-api:v1 \
    --platform managed \
    --region us-central1 \
    --allow-unauthenticated

--allow-unauthenticated: Makes the API public (great for testing). For internal APIs, you would remove this.

The Result: Google will spit out a URL that looks like: https://logicloop-backend-ue542.a.run.app

Click it, add /docs, and congratulations - your API is live on the internet, secured with HTTPS automatically.

Important Architecture Note: The Database

If you followed my previous guide using SQLite, you need to know a critical Cloud Run limitation: The filesystem is ephemeral.

If Cloud Run scales down or restarts your container (which happens often), your SQLite file is deleted and your data resets.

For Production: You must connect this service to a managed database like Google Cloud SQL (Postgres). You would pass the connection string as an Environment Variable:

Bash

gcloud run deploy logicloop-backend \
    --set-env-vars DATABASE_URL="postgresql+asyncpg://user:pass@IP:5432/dbname" ...

This keeps your application "stateless" (the code) while your data lives safely in a persistent "stateful" layer (the database).

What's Next?

We have Code and Deployment. But running docker build manually every time you change a line of code is painful.

In the next article, we will automate this pipeline using GitHub Actions for a full CI/CD workflow.

Serverless Automation: Scheduling Python Scripts with Google Cloud

Ayush Kumar — Tue, 14 Jul 2026 13:53:03 +0000

In Part 1, we set up the credentials. In Part 2, we wrote the Python code to send a notification.

But there is a problem: The code only runs when you manually type python main.py on your laptop.

If you want to send a "Good Morning" notification to your users at 9:00 AM every day, you can't rely on your laptop being open. You need a server.

But buying a VPS (Virtual Private Server) just to run a 10-line script once a day is a waste of money. This is where Serverless shines.

In this final guide, we will deploy our Python logic to Google Cloud Functions and trigger it with Cloud Scheduler.

The Architecture

We are building a "Cron Job" in the cloud:

Cloud Scheduler: The Alarm Clock (Triggers at 9:00 AM).
Pub/Sub: The Messenger (Passes the signal).
Cloud Function: The Worker (Executes our Python code).

Step 1: Prepare the Code for the Cloud

Cloud Func tions work slightly differently than local scripts. They don't use if __name__ == "__main__":. Instead, they wait for an event.

We need two files: main.py and requirements.txt.

1. requirements.txt This tells Google which libraries to install.

Plaintext

firebase-admin==6.2.0
functions-framework==3.0.0

2. main.py Notice how we removed the local certificate file path. In Google Cloud, the environment authenticates automatically!

Python

import firebase_admin
from firebase_admin import credentials, messaging

# Initialize Firebase without a manual JSON file
# Google Cloud Functions authenticates automatically!
if not firebase_admin._apps:
    firebase_admin.initialize_app()

def send_daily_notification(event, context):
    """
    Triggered from a message on a Cloud Pub/Sub topic.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    print(f"Function triggered by event: {event}")

    # Define the message
    message = messaging.Message(
        notification=messaging.Notification(
            title='Rise and Shine! ☀️',
            body='This is your automated daily update.',
        ),
        topic='news', # Target the 'news' topic
    )

    # Send it
    try:
        response = messaging.send(message)
        print(f"Successfully sent message: {response}")
        return "Done"
    except Exception as e:
        print(f"Error: {e}")
        return "Failed"

Step 2: Deploy to Cloud Functions

Go to the Google Cloud Console and search for "Cloud Functions".
Click Create Function.
Basics:

* [**Envir**](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)**onmen**[**t:** 1st](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3) [](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)Gen (simpler for thi[s tuto](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)rial)[.](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)

* [**F**](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)**unction N**[**ame:** `d`](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)`aily-notification-sender`.

* **Reg**[**ion:** C](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)h[oose o](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)ne close to you (e.g., `us-centra`[`l1` or](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3) `asia-south1`).

Trigger:

* [Sele](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)ct **Clo**[**ud Pub**](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)**/Sub**.

* Click **Crea**[**te a t**](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)[**o**](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)**pic** an[d name](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3) it `daily-trigger`.

* [Cli](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)ck **Save**[.](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)

[IMAGE: Screenshot of the Trigger selection screen in GCP Console]

Runtime:

* [Clic](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)k **Next**[.](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)

* [S](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)elect **Pyt**[**hon 3.**](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)**10** (o[r 3.9)](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3) as th[e runt](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)ime.

* C[opy an](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)d p[aste o](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)ur [`main.py`](http://main.py) and `requirements.txt` [code i](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)nto the inline editor.

* **Entry Poin**[**t:** Typ](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)e `send_daily_notification` (this [must m](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)atch your python function name).

Click Deploy.

It will take about 2 minutes. Once the green checkmark appears, your code is live on Google's infrastructure!

Step 3: Set the Schedule (The Alarm Clock)

Now we nee d something to "poke" that Pub/Sub topic every day.

Search for "Cloud Scheduler" in the console.
Click Create Job.
Configuration:

* **Name:** `mor`[`n`](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)[`ing-tr`](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)`igger`.

* **Freq**[**uency:**](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3) [`0 9 *`](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3) `* *` ([This i](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)s Cron syntax for "E[very d](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)ay at 9:00 AM").

* **Timezone:** [Select](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3) your local timezone.

Target:

* [**Targe**](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)**t Typ**[**e:** Pub](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)/Sub.

* **Topic:** Sel[ect th](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)[e](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3) `daily`[`-trigg`](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)`er` topic we created [earli](https://www.google.com/search?q=https://logiclooptech.dev/send-fcm-notifications-python&authuser=3)er.

* **Message Bo**[**dy:** Ty](https://www.google.com/search?q=https://logiclooptech.dev/setup-firebase-fcm-account&authuser=3)pe anything (e.g., "Wake up").

Click Create.

Step 4: Test It

You don't have to wait until 9 AM to test it.

In the Cloud Scheduler list, find your job.
Click the "Force Run" (or "Run Now") button.
Check your phone (if you have an app connected) or check the Cloud Functions Logs. You should see "Successfully sent message."

Conclusion

You have just built a completely Serverless Notification Engine.

Cost: $0.00 (until you hit millions of invocations).
Maintenance: None. No OS updates, no server restarts.
Scalability: Google handles it automatically.

This is the power of modern Backend Engineering. You moved from a script on a laptop to a production-grade cloud architecture.

Fix AsyncSession Errors in FastAPI (SQLAlchemy 2.0 Guide)

Ayush Kumar — Tue, 14 Jul 2026 13:52:25 +0000

Introduction

Building async APIs with FastAPI and SQLAlchemy 2.0 looks straightforward in tutorials, until you deploy to production.

Suddenly you start seeing issues like random MissingGreenlet errors, confusing async session behavior, blocked event loops, or database calls that are technically “async” but still slow under load. These problems usually appear when teams migrate from synchronous Flask or Django applications to FastAPI without fully understanding how async architecture actually works.

This article is not a beginner’s FastAPI tutorial.

It is a practical, production-focused guide to building high-performance async backend APIs using FastAPI and SQLAlchemy 2.0, covering real-world concerns such as async engine configuration, session lifecycle management, lifespan events, connection pooling, and common failure modes.

Most AsyncSession errors come from incorrect lifespan handling. Here’s the production-safe setup used in high-load FastAPI systems.

If you are already using FastAPI (or planning a migration from Flask) and want an async architecture that scales cleanly beyond toy examples, this guide is written for you.

The Setup

First, let's grab our dependencies. Notice we need an async driver (aiosqlite) because standard drivers like psycopg2 or sqlite3 are synchronous and will block your loop.

Bash

pip install fastapi uvicorn sqlalchemy aiosqlite pydantic

1. The Database Engine (`database.py`)

The most critical part of an async setup is the AsyncEngine. If you initialize this wrong, your whole app runs synchronously.

Python

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

# 1. Connection String (Note the +aiosqlite driver)
# For Postgres, use: postgresql+asyncpg://user:pass@localhost/dbname
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///./test.db"

# 2. Create the Async Engine
engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL,
    echo=True, # Logs SQL queries to console (Great for debugging)
)

# 3. Create the Session Factory
# This is what generates new database sessions for each request
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# 4. Base Class for Models
class Base(DeclarativeBase):
    pass

# 5. Dependency Injection
# We use this in our FastAPI routes to get a DB session
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

💡
getting an AttributeError: 'AsyncSession' object has no attribute 'query'? Read my [fix for migrating legacy queries to SQLAlchemy 2.0].

2. The Models (`models.py`)

SQLAlchemy 2.0 introduced a beautiful new way to define models using Python type hints (Mapped). No more vague Column(Integer, ...) syntax.

Python

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer, Boolean
from database import Base

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(50), index=True)
    description: Mapped[str] = mapped_column(String(255), nullable=True)
    is_completed: Mapped[bool] = mapped_column(default=False)

3. The Schemas (`schemas.py`)

Pydantic handles our data validation. We keep our "Create" logic separate from our "Response" logic.

Python

from pydantic import BaseModel, ConfigDict

class TaskCreate(BaseModel):
    title: str
    description: str | None = None

class TaskResponse(TaskCreate):
    id: int
    is_completed: bool

    # Pydantic V2 Config to read from ORM models
    model_config = ConfigDict(from_attributes=True)

4. The API Endpoints (`main.py`)

Here is where the magic happens. Notice two key things:

async def: The endpoints are asynchronous.
await session.execute(select(...)): We use the new SQLAlchemy 2.0 selection style, not the old session.query().

Python

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from contextlib import asynccontextmanager

import models, schemas
from database import engine, get_db

# Lifespan event to create tables on startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(models.Base.metadata.create_all)
    yield

app = FastAPI(lifespan=lifespan)

# CREATE
@app.post("/tasks/", response_model=schemas.TaskResponse)
async def create_task(task: schemas.TaskCreate, db: AsyncSession = Depends(get_db)):
    new_task = models.Task(**task.model_dump())
    db.add(new_task)
    await db.commit()
    await db.refresh(new_task)
    return new_task

# READ (Async Select)
@app.get("/tasks/", response_model=list[schemas.TaskResponse])
async def read_tasks(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
    # The Modern 2.0 Syntax
    query = select(models.Task).offset(skip).limit(limit)
    result = await db.execute(query)
    return result.scalars().all()

# UPDATE
@app.patch("/tasks/{task_id}", response_model=schemas.TaskResponse)
async def update_task(task_id: int, completed: bool, db: AsyncSession = Depends(get_db)):
    query = select(models.Task).where(models.Task.id == task_id)
    result = await db.execute(query)
    task = result.scalar_one_or_none()

    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")

    task.is_completed = completed
    await db.commit()
    await db.refresh(task)
    return task

Async SQLAlchemy Engine and Session Lifecycle in FastAPI

In production FastAPI applications, the async SQLAlchemy engine should be created once at application startup and reused across requests. Creating engines or sessions per request is a common mistake that leads to connection exhaustion and unpredictable performance.

FastAPI’s lifespan context is the recommended place to initialize the async engine and session factory, ensuring clean startup and shutdown behavior while avoiding hidden global state.

Note- SQLAlchemy 2.0 removed legacy query patterns, which is why AsyncSession no longer exposes .query().”

Reference Implementation

A minimal, production-focused reference implementation for this architecture is available on GitHub:

https://github.com/AyushKaushik-BD/fastapi-sqlalchemy-async-patterns

Why This Matters

In the synchronous world, if the database takes 200ms to fetch those tasks, your entire server thread is blocked for 200ms. It can do nothing else.

In this Async version, while the database is fetching data (await db.execute), Python releases the control loop. Your API can accept 50 other requests during that 200ms "wait" time.

This is how you scale to thousands of users on a single server.

Next Step: Deploying this

Now that you have a high-performance backend, how do you deploy it? You can't just use python main.py in production. In the next article, I will show you how to containerize this with Docker and deploy it to Google Cloud Run.

Mastering FastAPI Error Handling Best Practices – A Hands‑On Guide

Ayush Kumar — Tue, 14 Jul 2026 13:52:20 +0000

Introduction

When you build APIs with FastAPI, you quickly realize that graceful error handling is as important as the business logic itself. Poorly handled exceptions leak stack traces, break client contracts, and make debugging a nightmare. This article dives deep into fastapi error handling best practices - from understanding the framework’s default responses to crafting custom handlers, logging with middleware, shaping validation errors, and testing everything with TestClient. By the end you’ll have a ready‑to‑use error‑management toolkit that keeps your API predictable, secure, and developer‑friendly.

Understanding FastAPI's Default Error Responses

FastAPI inherits its request‑handling core from Starlette, which already provides a set of sensible HTTP error responses:

Status	Trigger	Default JSON Body
404	Route not found	`{"detail":"Not Found"}`
405	Method not allowed	`{"detail":"Method Not Allowed"}`
422	Validation error (Pydantic)	`{"detail":[{"loc":["body","name"],"msg":"field required","type":"value_error.missing"}]}`
500	Unhandled exception	`{"detail":"Internal Server Error"}`

These defaults are useful for quick prototypes, but they expose internal details (e.g., raw detail messages) and lack consistency across your API. For production‑grade services you’ll want to:

Standardize the error schema (e.g., always return {code, message, details}).
Hide implementation details from callers (no stack traces in production).
Log the full traceback for operators while sending a clean response to clients.

Understanding what FastAPI does out of the box helps you know what you need to override.

Creating Custom Exception Handlers

FastAPI lets you register handlers for any subclass of Exception. The most common pattern is to define a base API error and then derive specific cases.

# errors.py
from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse
from typing import Any, Dict

class APIError(HTTPException):
    """Base class for all custom API errors."""
    def __init__(self, status_code: int, message: str, payload: Dict[str, Any] | None = None):
        super().__init__(status_code=status_code, detail=message)
        self.payload = payload or {}

def register_error_handlers(app):
    @app.exception_handler(APIError)
    async def api_error_handler(request: Request, exc: APIError):
        # Log the error (could be sent to Sentry, etc.)
        # Here we just print for demonstration
        print(f"❗️ APIError: {exc.detail} – Path: {request.url.path}")

        body = {
            "code": exc.status_code,
            "message": exc.detail,
            "details": exc.payload,
        }
        return JSONResponse(status_code=exc.status_code, content=body)

    # Optional: catch any unhandled exception and wrap it
    @app.exception_handler(Exception)
    async def generic_exception_handler(request: Request, exc: Exception):
        print(f"⚠️ Unexpected error: {exc!r}")
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"code": 500, "message": "Internal server error"},
        )

Using the custom error in routes

# main.py
from fastapi import FastAPI, Depends
from errors import APIError, register_error_handlers

app = FastAPI()
register_error_handlers(app)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        raise APIError(
            status_code=404,
            message="Item not found",
            payload={"item_id": item_id},
        )
    return {"item_id": item_id, "name": f"Item {item_id}"}

Why this is a best practice:

All error responses now share the same JSON shape.
You can enrich the payload with contextual data without breaking the contract.
Centralized logging ensures every error is captured in one place.

Using Middleware for Error Logging

Exception handlers are perfect for shaping the response, but they’re invoked after the request has been processed. If you need to log every request/response pair - including the time taken and any exception that bubbles up - middleware is the right tool.

# middleware.py
import time
import traceback
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class ErrorLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        try:
            response: Response = await call_next(request)
        except Exception as exc:
            # Log full traceback for ops
            tb = traceback.format_exc()
            print(f"🚨 Unhandled exception on {request.url.path}:\n{tb}")

            # Re‑raise so FastAPI's exception handlers can format the response
            raise exc
        finally:
            process_time = (time.time() - start) * 1000
            print(
                f"🕒 {request.method} {request.url.path} completed in {process_time:.2f}ms "
                f"with status {response.status_code}"
            )
        return response

Add it to your app:

# main.py (continued)
from middleware import ErrorLoggingMiddleware

app.add_middleware(ErrorLoggingMiddleware)

Benefits:

Guarantees that every request is accounted for, even those that raise before hitting a route (e.g., authentication failures).
Keeps logging logic separate from business logic, adhering to the single‑responsibility principle.
Works nicely with external observability stacks (Prometheus, Grafana, Sentry) because you can replace the print statements with structured logs.

Handling Validation Errors and Custom Response Models

FastAPI automatically validates path, query, and body parameters using Pydantic. Validation failures raise a RequestValidationError, which FastAPI turns into a 422 response. While useful, the default schema can be verbose for clients.

Overriding the 422 handler

# validation.py
from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

def register_validation_handler(app):
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        # Extract a cleaner list of errors
        errors = [
            {"field": ".".join(map(str, err["loc"])), "message": err["msg"]}
            for err in exc.errors()
        ]
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={
                "code": 422,
                "message": "Input validation failed",
                "details": errors,
            },
        )

Add it alongside your other handlers:

# main.py (continued)
from validation import register_validation_handler

register_validation_handler(app)

Using a custom response model for errors

Define a reusable Pydantic model that describes your error payload:

# schemas.py
from pydantic import BaseModel
from typing import Any, List, Optional

class ErrorDetail(BaseModel):
    field: str
    message: str

class APIErrorResponse(BaseModel):
    code: int
    message: str
    details: Optional[Any] = None

Now you can annotate your endpoint to explicitly state the possible error response:

# main.py (continued)
from fastapi import status
from fastapi.responses import JSONResponse
from schemas import APIErrorResponse

@app.post(
    "/users/",
    responses={422: {"model": APIErrorResponse}},
)
async def create_user(name: str):
    # Imagine some validation beyond Pydantic here
    if len(name) < 3:
        raise APIError(
            status_code=422,
            message="Name too short",
            payload={"field": "name", "min_length": 3},
        )
    return {"id": 1, "name": name}

Result: Clients receive a consistent error object, and OpenAPI docs now display the exact schema - great for front‑end teams and API consumers.

Testing Error Handling with FastAPI's TestClient

Automated tests should verify that your error handling behaves exactly as documented. FastAPI’s TestClient (built on requests) makes this straightforward.

# test_error_handling.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_item_not_found():
    response = client.get("/items/0")
    assert response.status_code == 404
    json = response.json()
    assert json["code"] == 404
    assert json["message"] == "Item not found"
    assert json["details"]["item_id"] == 0

def test_validation_error():
    # Omit required query param 'name' to trigger 422
    response = client.get("/items/5")
    # Suppose we have a route that expects a query param; demonstration only
    assert response.status_code == 422
    json = response.json()
    assert json["code"] == 422
    assert json["message"] == "Input validation failed"
    # Ensure our custom error detail shape appears
    assert any(err["field"] == "query.name" for err in json["details"])

def test_unexpected_exception_is_wrapped():
    # Create a route that raises a ZeroDivisionError for testing
    @app.get("/explode")
    async def explode():
        1 / 0

    response = client.get("/explode")
    assert response.status_code == 500
    json = response.json()
    assert json["code"] == 500
    assert json["message"] == "Internal server error"

Tips for robust testing

Parametrize common error scenarios (e.g., missing fields, wrong types).
Use fixtures to spin up a clean app instance per test to avoid state leakage.
Integrate these tests into your CI pipeline; a failing test will immediately surface a regression in your error handling logic.

Putting It All Together – A Minimal Production‑Ready Template

Below is a compact skeleton you can copy into a new project. It incorporates all the practices discussed, plus two internal links for deeper dives:

# app/main.py
from fastapi import FastAPI
from errors import register_error_handlers
from validation import register_validation_handler
from middleware import ErrorLoggingMiddleware

app = FastAPI(
    title="FastAPI Error Handling Best Practices",
    version="0.1.0",
)

# Register core components
app.add_middleware(ErrorLoggingMiddleware)
register_error_handlers(app)
register_validation_handler(app)

# Example endpoint
@app.get("/ping")
async def ping():
    return {"msg": "pong"}

# For async DB work, see our guide on
# [Fix AsyncSession Errors in FastAPI (SQLAlchemy 2.0 Guide)](https://www.logiclooptech.dev/modern-backend-building-high-performance-async-apis-with-fastapi-and-sqlalchemy-20/)
# and for deployment considerations, read
# [FastAPI in Production: The Complete Deployment Guide](https://www.logiclooptech.dev/fastapi-in-production-the-complete-deployment-guide-docker-workers-scaling-and-best-practices/)

Running uvicorn app.main:app --reload now gives you an API with:

Uniform error JSON (code, message, details).
Centralized logging of request latency and stack traces.
Clean OpenAPI docs that expose error schemas.
Testable error paths via TestClient.

Key Takeaways

FastAPI’s defaults are a starting point, not a final solution; customize them to enforce a single error contract.
Create a base APIError and register a global exception handler to shape every error response.
Middleware is the ideal place for request‑wide logging, timing, and catching unexpected exceptions before they escape.
Override the 422 handler to return a concise, client‑friendly validation payload and document it with a Pydantic model.
Test rigorously using TestClient to guarantee that both happy‑path and error‑path responses match the documented schema.
Integrate these patterns early; they pay off when you scale the service, add authentication, or move to production environments.

By following the fastapi error handling best practices outlined here, you’ll deliver APIs that are easier to consume, simpler to debug, and ready for the demands of modern production workloads. Happy coding!

Mastering FastAPI Deployment on Render – From Docker to CI/CD, Scaling & Cost‑Effective Tips

Ayush Kumar — Tue, 14 Jul 2026 13:51:42 +0000

Introduction

If you’re looking for a fastapi deployment on render that balances performance, developer experience, and cost, you’ve landed in the right place. Render’s managed platform lets you spin up web services, background workers, and databases with just a few clicks, while still giving you the flexibility of Docker and full CI/CD pipelines. In this post we’ll walk through the entire lifecycle: creating a Render web service for FastAPI, crafting an optimal Dockerfile, wiring environment variables and databases, automating releases with GitHub Actions, and finally scaling and monitoring your API in production. By the end you’ll have a production‑ready FastAPI app that can handle traffic spikes without blowing the budget.

Why Render?

Render abstracts away the operational overhead of provisioning VMs, configuring load balancers, and managing TLS certificates. It also offers a generous free tier for hobby projects, making it an attractive choice for startups and solo developers alike.

Creating a Render web service for FastAPI

Sign up and create a new service
- Log into your Render dashboard.
- Click New → Web Service.
- Connect your GitHub repository that contains the FastAPI code.
Select the runtime
- Choose Docker as the environment (Render will build the image from your Dockerfile).
- Set the Instance Type – start with the free “Starter” instance (512 MiB RAM) and upgrade later as needed.
Set the start command

Render automatically detects the CMD or ENTRYPOINT from the Dockerfile, but you can override it. For a typical Uvicorn launch:

   gunicorn -k uvicorn.workers.UvicornWorker app.main:app --workers 2 --bind 0.0.0.0:${PORT}

Render injects the ${PORT} environment variable, so your container always binds to the correct port.

Deploy Click Create Web Service. Render will clone the repo, run the Docker build, and spin up the container. In a few minutes you’ll have a publicly accessible URL like https://fastapi-demo.onrender.com.

Writing an optimal Dockerfile for Render

A lean Docker image reduces build time, memory footprint, and cost. Below is a production‑ready Dockerfile that follows best practices:

# ---- Base image ------------------------------------------------------------
FROM python:3.12-slim-bullseye AS builder

# Install build dependencies (gcc, libpq-dev) for compiled wheels
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential gcc libpq-dev && rm -rf /var/lib/apt/lists/*

# Set a non‑root user
ARG UID=1000
ARG GID=1000
RUN groupadd -g ${GID} appgroup && \
    useradd -m -u ${UID} -g ${GID} -s /bin/bash appuser

# Create a virtual environment
ENV VIRTUAL_ENV=/opt/venv
RUN python -m venv $VIRTUAL_ENV
ENV PATH="$VIRTUAL_ENV/bin:$PATH"

# Install production dependencies only
WORKDIR /app
COPY requirements.txt .
RUN pip install --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# ---- Runtime image --------------------------------------------------------
FROM python:3.12-slim-bullseye AS runtime

# Copy the virtual environment from the builder stage
COPY --from=builder /opt/venv $VIRTUAL_ENV
ENV PATH="$VIRTUAL_ENV/bin:$PATH"

# Create a non‑root user (same UID/GID as builder)
ARG UID=1000
ARG GID=1000
RUN groupadd -g ${GID} appgroup && \
    useradd -m -u ${UID} -g ${GID} -s /bin/bash appuser

# Copy application code
WORKDIR /app
COPY . /app

# Ensure the app runs as non‑root
USER appuser

# Expose the port Render will assign (default 10000, but use $PORT at runtime)
EXPOSE 10000

# Default command – Render will override if you set a custom start command
CMD ["gunicorn", "-k", "uvicorn.workers.UvicornWorker", "app.main:app", "--workers", "2", "--bind", "0.0.0.0:${PORT}"]

Why this Dockerfile works well on Render

Feature	Benefit
Multi‑stage build	Keeps the final image tiny (≈80 MB) because build‑time packages are discarded.
Non‑root user	Aligns with Render’s security recommendations; prevents privilege escalation.
Pinned Python version	Guarantees reproducible builds across environments.
`requirements.txt` only	Avoids copying the whole repo before installing deps, which speeds up caching.

If you need extra OS packages (e.g., ffmpeg for media processing), add them to the first stage’s apt-get install line.

Configuring environment variables and databases on Render

1. Adding secret environment variables

Render’s UI offers a Secrets tab for each service. Click Add Secret and enter key/value pairs:

Key	Example Value
`DATABASE_URL`	`postgresql://user:password@mydb.internal:5432/mydb`
`REDIS_URL`	`redis://default:password@myredis.internal:6379`
`APP_ENV`	`production`
`JWT_SECRET_KEY`	`super‑secret‑key`

All secrets are injected as environment variables at container start, and they are never exposed in logs.

2. Provisioning a PostgreSQL database

From Render dashboard, click New → Database → PostgreSQL.
Choose a plan (the free tier provides 256 MiB, sufficient for dev or low‑traffic apps).
Once created, copy the auto‑generated DATABASE_URL secret into your web service’s secret list.

3. Accessing the DB from FastAPI

# app/config.py
import os
from pydantic import BaseSettings, PostgresDsn

class Settings(BaseSettings):
    database_url: PostgresDsn = os.getenv("DATABASE_URL")
    redis_url: str = os.getenv("REDIS_URL", "")
    environment: str = os.getenv("APP_ENV", "development")

    class Config:
        env_file = ".env"

settings = Settings()

# app/db.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from .config import settings

engine = create_async_engine(settings.database_url, echo=False, future=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Tip: For deeper diagnostics on long‑running DB sessions, see our guide on FastAPI SQLAlchemy Session Leak Detection.

4. Using Redis for background tasks

Render also supports Redis as a managed service. After provisioning, add REDIS_URL to your secrets and use it with aioredis or redis-py:

# app/worker.py
import aioredis
from .config import settings

redis = aioredis.from_url(settings.redis_url)

async def enqueue_task(name: str, payload: dict):
    await redis.rpush(name, json.dumps(payload))

Setting up automated CI/CD with GitHub Actions for Render

Render can automatically redeploy on every push, but coupling it with GitHub Actions gives you linting, tests, and image scanning before the build hits Render.

1. Create a workflow file

.github/workflows/render-deploy.yml

name: CI & Deploy to Render

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
      - name: Run pytest
        run: pytest -q

  build-and-deploy:
    needs: test
    runs-on: ubuntu-latest
    permissions:
      contents: read
      id-token: write   # needed for Render Deploy Hook
    steps:
      - uses: actions/checkout@v4
      - name: Build Docker image (optional pre‑check)
        run: |
          docker build -t fastapi-app:${{ github.sha }} .
      - name: Trigger Render Deploy Hook
        env:
          RENDER_DEPLOY_HOOK: ${{ secrets.RENDER_DEPLOY_HOOK }}
        run: |
          curl -X POST "$RENDER_DEPLOY_HOOK"

2. Configure the Deploy Hook in Render

In your Render service, go to Settings → Deploy Hooks.
Click Create Deploy Hook, give it a name, and copy the generated URL.
Store this URL as a secret named RENDER_DEPLOY_HOOK in your GitHub repository (Settings → Secrets and variables).

Now every push to main runs unit tests, builds the Docker image locally (to catch syntax errors early), and finally notifies Render to pull the latest commit and redeploy.

3. Adding a Linting Stage

If you want static analysis, add a step using ruff or flake8:

- name: Lint with ruff
  run: |
    pip install ruff
    ruff check .

Result: A fast, automated pipeline that guarantees only healthy code reaches production.

Scaling, monitoring, and cost‑effective tips on Render

Horizontal scaling

Render lets you scale a web service horizontally by increasing the Instance Count. Each additional instance runs the same Docker image behind a built‑in load balancer.

When to scale: Monitor request latency and CPU usage. If average CPU > 70% for 5 minutes, add an instance.
How to automate: Use Render’s Autoscaling (beta) – set a target CPU threshold, and Render will spin up/down instances automatically.

Vertical scaling

If your API is CPU‑bound (e.g., heavy data processing), upgrade to a larger instance type (e.g., “Standard” with 2 vCPU, 4 GiB RAM). Render’s pricing is linear, so you can experiment on a staging branch before committing.

Monitoring

Built‑in metrics – Render Dashboard shows CPU, memory, and request latency per service.
Prometheus + Grafana – Render supports custom metrics via the /metrics endpoint. Add prometheus_fastapi_instrumentator:

   # app/main.py
   from fastapi import FastAPI
   from prometheus_fastapi_instrumentator import Instrumentator

   app = FastAPI()
   Instrumentator().instrument(app).expose(app)

Then point Grafana to https://your-service.onrender.com/metrics.

Error tracking – Integrate Sentry (add SENTRY_DSN secret). Render will forward unhandled exceptions automatically.

Cost‑effective tricks

Tip	Why it saves money
Use the free tier for dev	Render’s free Starter instance is enough for CI testing and low‑traffic preview URLs.
Turn off unused workers	If you have background workers (e.g., Celery) that are idle, set them to “Paused” in the UI.
Leverage Render’s static site hosting	For documentation or static assets, host them on Render’s static site service (free) instead of bundling them in the API container.
Cache Docker layers	The multi‑stage Dockerfile above caches pip installs, reducing build time and Render’s build minutes consumption.

Example: Autoscaling configuration

# render.yaml (optional, for declarative infra)
services:
  - type: web
    name: fastapi-api
    env: docker
    plan: starter
    autoDeploy: true
    autoscaling:
      minInstances: 1
      maxInstances: 5
      targetCpuUtilization: 0.65

Commit this file to the repo root, and Render will sync the config on next deploy.

Full Minimal FastAPI Application

Here’s a concise, production‑ready FastAPI app that works out‑of‑the‑box with the Dockerfile and Render configuration above.

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from . import db, models, schemas, crud
from .config import settings

app = FastAPI(
    title="Render‑Ready FastAPI",
    version="0.1.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# Dependency injection for DB session
async def get_db_session() -> AsyncSession:
    async for session in db.get_db():
        yield session

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

@app.post("/users/", response_model=schemas.UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(user_in: schemas.UserCreate, db: AsyncSession = Depends(get_db_session)):
    existing = await crud.get_user_by_email(db, email=user_in.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    user = await crud.create_user(db, obj_in=user_in)
    return user

@app.get("/users/{user_id}", response_model=schemas.UserRead)
async def read_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
    user = await crud.get_user(db, user_id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# app/schemas.py
from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = None

class UserCreate(UserBase):
    password: str

class UserRead(UserBase):
    id: int

    class Config:
        orm_mode = True

# app/crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(models.User).where(models.User.id == user_id))
    return result.scalars().first()

async def get_user_by_email(db: AsyncSession, email: str):
    result = await db.execute(select(models.User).where(models.User.email == email))
    return result.scalars().first()

async def create_user(db: AsyncSession, obj_in: schemas.UserCreate):
    user = models.User(email=obj_in.email, full_name=obj_in.full_name)
    user.set_password(obj_in.password)  # assumes a helper method on model
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

# app/models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
import bcrypt

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    full_name = Column(String, nullable=True)
    hashed_password = Column(String, nullable=False)

    def set_password(self, raw_password: str):
        self.hashed_password = bcrypt.hashpw(
            raw_password.encode("utf-8"), bcrypt.gensalt()
        ).decode("utf-8")

Run migrations (Render will execute them on start if you add a render.yaml hook or a docker-compose step). For a quick start, you can use alembic or simply:

python -c "from app.db import engine; import app.models as m; m.Base.metadata.create_all(bind=engine.sync_engine)"

Key Takeaways

Fastapi deployment on render is straightforward: a Dockerfile, a Render web service, and a handful of secrets get you online in minutes.
Use a multi‑stage Dockerfile with a non‑root user to keep images small and secure.
Store database URLs, API keys, and other secrets in Render’s secret manager; never hard‑code them.
GitHub Actions provide a safety net - run tests, lint, and trigger Render’s Deploy Hook automatically.
Scaling can be vertical, horizontal, or autoscaled; start on the free tier and monitor CPU/memory to decide when to upgrade.
Leverage Render’s built‑in metrics, logs, and TLS; add Prometheus instrumentation for deeper insights.
Follow cost‑saving best practices: pause idle workers, use static site hosting for assets, and cache Docker layers.

With these steps you now have a robust pipeline that takes a FastAPI codebase from local development to a scalable, monitored production service on Render - all while keeping operational costs under control. Happy deploying!

DEV Community: Ayush Kumar

Building microservices python fastapi: design to production

Why microservices python fastapi are a solid choice for production

How do I design a FastAPI-based microservice that stays maintainable?

Which communication pattern should I pick: REST, gRPC, or async messaging?

When is REST enough?

When does gRPC make sense?

When to use async messaging (Kafka, RabbitMQ)?

How do I implement message queues with RabbitMQ or Kafka?

RabbitMQ + Celery example

Kafka producer example

How do I containerize and orchestrate a FastAPI microservice with Docker Compose & Kubernetes?

Dockerfile (multi-stage)

Docker Compose for local dev

Helm chart snippet for Kubernetes

How should I test, monitor, and trace FastAPI microservices?

Unit & integration testing

Monitoring & tracing

Load testing

What are the security and authentication best practices for FastAPI microservices?

FAQ

Key Takeaways

Why FastAPI Apps Slow Down Over Time (Low CPU, High Latency Explained)

The "Silent Death" Scenario

Understanding the Single-Threaded Model

Killer #1: The "Sync" Blocker (and the def Trap)

The Mistake: Synchronous I/O in Async Path

The Nuance: def vs async def

The Fix: asyncio.to_thread

Red Flags That Your Event Loop Is Blocked

Killer #2: The Hidden CPU Crushers (Password Hashing)

Killer #3: The httpx Client Leak

Killer #4: The "Zombie" Background Task

Why Adding More CPU or Pods Won’t Save You

How to Debug: Finding the Invisible Wall

Conclusion

Welcome to LogicLoopTech

What you'll find here

Why this exists

Why Uvicorn Health Checks Fail Under Load (And How to Fix It Properly)

The Real Problem: Event Loop Starvation

Mistake #1: “Deep” Health Checks

The Correct Design: Split Probes

1- Liveness Probe

2- Readiness Probe

Mistake #2: Aggressive Probe Settings

Mistake #3: Blocking the Event Loop

Optional Strategy: Use def Instead of async def

The Nuclear Option: Separate Health Port

What Causes Liveness Probe Failures in FastAPI?

The Real Lesson

Final Checklist

Symptoms You’ll Notice Before Pods Restart

Using the OpenAI Agents SDK for Python: A Practical Guide

Quick answer: what is the OpenAI Agents SDK for Python and how do I get it running?

Installing and configuring the OpenAI Agents SDK in a Python environment

Building your first OpenAI agent with tool use

Managing agent memory and state across interactions

Adding custom tools and functions to agents

Deploying and scaling OpenAI agents with FastAPI or serverless platforms

Frequently asked questions

How do I debug a tool that the LLM refuses to call?

Can I use the SDK with multiple OpenAI models simultaneously?

What’s the best way to store session memory for a large user base?

Does the SDK support streaming responses?

Key Takeaways

Supercharge Your API: FastAPI asyncpg Integration for High‑Performance PostgreSQL Access

Introduction

Setting up asyncpg in a FastAPI project

1. Install the dependencies

2. Minimal FastAPI app using asyncpg

Why asyncpg?

Connection pooling and lifecycle management with FastAPI

Pool configuration best practices

Using a Dependency for Per‑request Access

Detecting leaks

Using asyncpg with SQLAlchemy 2.0 async engine

1. Install the async driver

2. Configure the async engine

3. Define a model

Killer #1: The "Sync" Blocker (and the `def` Trap)

The Nuance: `def` vs `async def`

The Fix: `asyncio.to_thread`

Killer #3: The `httpx` Client Leak

Optional Strategy: Use `def` Instead of `async def`

Supercharge Your API: FastAPI asyncpg Integration for High‑Performance PostgreSQL Access

Using asyncpg with SQLAlchemy 2.0 async engine

2. Leverage `COPY` for bulk loads

1. The Database Engine (`database.py`)

2. The Models (`models.py`)

3. The Schemas (`schemas.py`)

4. The API Endpoints (`main.py`)