#fastapi

Welcome back, coding enthusiasts long time no see! Today we’re doing a quick refresh on the basics of authentication. We’ll walk through building a FastAPI backend with signup, login, email verification, JWT tokens, and Google OAuth login.

What is FastAPI & Why Not Just Python?

FastAPI is a modern Python web framework that makes building APIs super fast (pun intended 😎). You could technically write your authentication system in plain Python using Flask or even raw sockets, but FastAPI gives you:

Automatic request validation with Pydantic models. No more manually checking every field.

Automatic docs with Swagger UI so your API docs are basically built for free.

Async support, which means faster response times if you need to handle lots of users at once.

In short, FastAPI helps you focus on the logic, not the plumbing. And trust me, authentication has enough plumbing already.

The Core Idea: How Authentication Works

Authentication is about proving who a user is. Here’s how our setup works:

Sign Up – Users register with email + password. Password is hashed (never stored in plain text).

Email Verification – Users must confirm their email via a unique link. This ensures they own the email.

Login – Users can log in with email/password or via Google OAuth.

JWT Tokens – After logging in, a JWT token is issued, proving the user is authenticated for protected routes.

Everything is tied together so that once a user signs up, they cannot sign up again with the same email. We check the database first.

Password Hashing: Because Plain Text is a No-No

We use bcrypt to hash passwords. Here’s the flow:

When a user signs up, we hash their password with bcrypt.gensalt() and store the hash in the database.

When they log in, we use bcrypt.checkpw() to verify the password against the hash.

This way, even if the database leaks, passwords remain safe.

JWT Tokens: Your API Passport

JWT stands for JSON Web Token. Think of it like a passport for your API:

When a user logs in, we generate a JWT with their email embedded as sub.

The token has an expiry, usually 7 days in our case.

Every time the user calls a protected route, we check the token , they’re only allowed in if this token is valid.

Fast, stateless, and secure. No sessions to manage on the server.

Email Verification: Making Sure It’s Really You

Email verification is a key step:

When a user signs up, we generate a random token and store it in a verification_tokens table.

We send them an email with a verification link pointing to /verify-email?token=XYZ.

When they click it, our backend verifies the token and marks the user as verified.

Bonus: we don’t delete the token immediatelalloy because it will allow safe re-use and auditing.

Sending the email is handled via smtplib in Python. You can customize the HTML and plain-text versions.

Google OAuth: Login With One Click

We also support Google login:

The frontend sends a Google ID token to /api/auth/google.

We decode it and check if the user exists in our database via google_id or email.

If not, we create a new user and mark them verified automatically.

Then, we issue a JWT just like regular login.

This allows users to bypass password setup entirely if they want.

How It All Comes Together

app.py/main.py is the brain: it wires FastAPI routes, database connections, and services.

auth.py contains all the logic: password hashing, JWT creation/verification, Google token verification, email token handling.

email_service.py handles sending emails: verification emails and welcome emails.

database.py manages connections and table setup.

models.py defines request and response schemas for FastAPI to validate and serialize.

The architecture is clean: logic, models, services, and API are all separated, making it easy to maintain and scale.

And there you have it! That’s the full FastAPI authentication system we just dissected.

Next steps: implement your frontend to talk to these endpoints, test everything in Postman, and make sure your emails send correctly.

Optimizing Local RAG Pipelines: Advanced Context Injection and Automated Memory Management with SQLite-vec and Gemma 3

The Challenge of Infinite Memory in Local LLMs One of the primary limitations of running Large Language Models (LLMs) locally is the strict constraint...

#AI Development #Database Architecture #FastAPI #Linux Infrastructure #Python

FastAPI Observability : Correlation IDs & ContextVars (2026)

Untangling Asynchronous Logs: The Foundation of Request Tracing in Python Microservices

Debugging highly concurrent applications can feel like navigating a maze blindfolded. Imagine a scenario: your Python microservice, built with a framework like FastAPI, is humming along. Suddenly, production alerts blare at 3 AM. You check the logs, and a flood of Error 500: Database timeout while fetching user messages scrolls by. Ten thousand lines per minute, five different users hitting the same endpoint concurrently. Without a clear way to link these scattered log entries back to a specific user's request, you're not debugging; you're guessing.

This challenge highlights a fundamental need in modern distributed systems: observability. It's about giving your application a nervous system, allowing you to understand its internal state from external outputs. The first step in achieving this clarity is establishing a robust request tracing mechanism.

1. The Request Identifier: Your Debugging Compass

In traditional, synchronous applications, logs often appear in a somewhat sequential order, making it simpler to follow a single request's journey. However, asynchronous frameworks like FastAPI operate differently. While one request might be waiting for an I/O operation (like a database query), the event loop efficiently switches to process other requests. This interleaving of operations means your log files become a jumbled tapestry, with entries from multiple concurrent requests interwoven.

To cut through this noise, every single log entry related to a specific HTTP request needs a unique identifier. This is commonly known as a Trace ID or Correlation ID. It acts as a consistent tag, allowing you to group all related events, regardless of when or where they occurred in the log stream.

Attempting to manually pass this trace_id string through every function call – from your API endpoint down through service layers, repositories, and database adapters – is a significant anti-pattern. It clutters your clean code with boilerplate, making it harder to read and maintain.

The Right Tool for Asynchronous Context

Simply relying on global variables for a request ID is a recipe for disaster in concurrent environments, as they're shared across all requests. Similarly, threading.local() isn't suitable for asynchronous Python, where tasks might switch threads or run on the same thread but need independent context.

The correct approach for managing context in modern asynchronous Python applications is contextvars. This module provides a way to store and retrieve context-specific data that is local to an asynchronous task, ensuring isolation and preventing data leakage between concurrent operations.

Think of it like this: when a package enters a complex sorting facility (your application), it's immediately assigned a unique tracking number (the Correlation ID). No matter how many different conveyors (functions) it passes through, how many times it's paused or rerouted, any scanner (log entry) can read that tracking number and know exactly which shipment (request) it belongs to.

While you can implement contextvars from scratch, integrating it within a web framework often involves middleware. Here's how you might build an interceptor for FastAPI:

import uuid from contextvars import ContextVar from fastapi import FastAPI, Request, Response # 1. Define the Context Variable. This is safe across async boundaries! # It holds the trace_id for the current asynchronous task. trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="-") app = FastAPI() @app.middleware("http") async def inject_observability(request: Request, call_next): # Attempt to retrieve a trace ID from an incoming header (e.g., from an API Gateway) # If not present, generate a new unique ID for this request. request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())) # Attach the generated/received ID to the current Asyncio Task's context. # The 'token' is crucial for resetting the context later. token = trace_id_ctx.set(request_id) try: response = await call_next(request) # Add the trace ID to the response headers, allowing clients to report it. response.headers["X-Trace-ID"] = request_id return response finally: # CRITICAL: Reset the context variable for the current task. # This prevents memory leaks and ensures context isolation for subsequent tasks. trace_id_ctx.reset(token)

This middleware ensures that every incoming request either reuses an existing X-Request-ID or gets a new, unique trace_id. This ID is then made available throughout the request's lifecycle via trace_id_ctx, and finally, it's returned in the response header for client-side correlation.

2. Measuring Performance: Timing the Operations

Knowing what happened is crucial, but understanding how long it took is equally vital for performance analysis and identifying bottlenecks. Before you can push metrics to advanced dashboards, you need to capture the raw duration of operations.

For precise timing in Python, time.perf_counter() is your go-to function. Unlike time.time(), which can be affected by system clock adjustments, perf_counter() provides a high-resolution, monotonic timer, making it ideal for measuring short durations accurately.

Logs might tell you that an error occurred, but true observability reveals the full story: which specific request triggered it, the exact sequence of operations leading up to the failure, and precisely how long each step took. This level of detail transforms reactive debugging into proactive system health management.

Project: Building a Structured JSON Logger

Parsing unstructured text logs is inefficient and error-prone. For effective log aggregation and analysis, your logs should be structured, ideally in JSON format. This allows tools to easily ingest, filter, and query your application's output.

Your task: Create a custom Python logger that automatically formats all log entries into JSON, including the trace_id from your ContextVar and the duration of the request.

Custom Formatter: Implement a subclass of logging.Formatter.

Contextual Data: Within its format() method, construct a dictionary containing essential fields like timestamp, level, message, and crucially, the trace_id fetched from your trace_id_ctx ContextVar.

JSON Output: Use json.dumps() to convert this dictionary into a JSON string, which will be the final log entry.

Duration Logging: Enhance your middleware to calculate the total request duration (using time.perf_counter()) and include it in a structured log entry when the request finishes (e.g., {"message": "Request processed", "duration_ms": 123.45, "trace_id": "..."}).

Next Steps: Beyond Local Logs

While structured JSON logs and correlation IDs are powerful for a single application instance, the real challenge arises with distributed systems involving many services and multiple instances. Manual log inspection quickly becomes unscalable. The next step in building a truly observable system involves integrating dedicated tools for metrics collection, log aggregation, and visualization.

FastAPI Graceful Shutdown: Handling SIGTERM in Kubernetes

Ensuring Smooth Exits: Implementing Graceful Shutdowns in Kubernetes

Imagine a critical deployment. You push an update, expecting a seamless transition. Instead, your monitoring dashboard lights up: thousands of in-flight operations fail, active user sessions drop, and background tasks vanish mid-processing. The culprit? Your server didn't gracefully step aside; it was abruptly terminated. While container orchestrators like Kubernetes manage pod lifecycles, they don't inherently guarantee a "graceful" exit for your applications. That responsibility falls to you, the application developer.

The Pitfalls of Abrupt Termination

When Kubernetes decides to terminate a pod—perhaps during a rolling update, scale-down, or node drain—it sends a SIGTERM signal to the primary process within the container. Many developers, especially in Python, might rely on application framework-specific shutdown hooks, like FastAPI's lifespan events or on_event("shutdown").

However, these framework-level events often trigger too late in the termination sequence. By the time your application code receives the shutdown notification, the underlying server might have already stopped accepting new connections, or even worse, severed existing ones (like WebSocket connections). Any tasks queued, in progress, or users awaiting a final response are immediately impacted. To prevent this, your application needs to intercept the SIGTERM signal at a lower level, closer to the operating system, and initiate a controlled shutdown before the framework itself begins to unravel.

Why Just Closing the Database Isn't Enough

A truly graceful shutdown isn't just about cleaning up internal resources like database connections or file handles. The paramount concern is traffic draining. If your application doesn't signal its impending termination to the load balancer or service mesh before it stops processing requests, new traffic will continue to be routed to a dying instance for several seconds. This creates a race condition where users encounter errors from a server that's already in its final moments.

The Analogy: A Ship Abandonment Plan

Consider your server as a ship. A SIGTERM is the order to abandon ship. A poorly managed ship captain might immediately jump overboard, leaving passengers (active tasks and connections) to fend for themselves. A responsible captain, however, would first announce that no new passengers can board (stop accepting new traffic), then ensure all current passengers are safely offloaded into lifeboats (allow existing tasks to complete) before finally leaving the ship themselves. This is the essence of a graceful shutdown.

The Readiness Flag Pattern

A robust approach to graceful shutdowns centers around a simple, global boolean flag. Let's call it SHOULD_ACCEPT_TRAFFIC. Initially, this flag is True, and your application's /healthz or /readiness endpoint returns a 200 OK status.

The moment your application receives the SIGTERM signal, you immediately flip SHOULD_ACCEPT_TRAFFIC to False. Consequently, your /healthz endpoint now returns a 503 Service Unavailable status.

Kubernetes' readinessProbe continuously monitors this endpoint. Upon seeing the 503 status, and after its configured failureThreshold is met, Kubernetes will stop routing new traffic to that specific pod. This initiates a "quiet period," allowing existing connections and in-progress tasks to complete their work without being interrupted by new requests.

Implementing the Shutdown Guard in Python (FastAPI)

Here's how you can implement this pattern using Python with FastAPI, intercepting the SIGTERM signal directly:

import signal import asyncio from fastapi import FastAPI, Response, status app = FastAPI() # Global flag to control traffic acceptance SHOULD_ACCEPT_TRAFFIC = True ACTIVE_TASKS = 0 # Optional: for more advanced draining def handle_termination_signal(*_): """ Callback function for SIGTERM signal. Immediately sets the flag to stop accepting new traffic. """ global SHOULD_ACCEPT_TRAFFIC SHOULD_ACCEPT_TRAFFIC = False print("SIGTERM received. Initiating traffic draining...") # Register the OS signal handler immediately upon application start signal.signal(signal.SIGTERM, handle_termination_signal) @app.get("/healthz", status_code=status.HTTP_200_OK) async def readiness_probe(): """ Kubernetes readiness probe endpoint. Returns 503 if the application is shutting down. """ if not SHOULD_ACCEPT_TRAFFIC: return Response(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, content="Server is shutting down.") return {"status": "ok"} @app.post("/process-data") async def process_data_endpoint(): """ Example endpoint for processing tasks. Checks the traffic flag to reject new requests during shutdown. """ global ACTIVE_TASKS if not SHOULD_ACCEPT_TRAFFIC: # Reject new requests if the server is draining return Response(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, content="Server terminating. No new tasks accepted.") # Increment active tasks counter (for more advanced draining) ACTIVE_TASKS += 1 try: # Simulate some asynchronous work await asyncio.sleep(5) print("Task processed.") return {"message": "Data processed successfully."} finally: # Decrement active tasks counter ACTIVE_TASKS -= 1 # Optional: A more robust shutdown hook using FastAPI's lifespan # This would run AFTER the readiness probe starts returning 503 @app.on_event("shutdown") async def app_shutdown(): print("FastAPI shutdown event triggered.") # Wait for active tasks to complete before truly exiting while ACTIVE_TASKS > 0: print(f"Waiting for {ACTIVE_TASKS} active tasks to finish...") await asyncio.sleep(1) print("All active tasks completed. Application shutting down.")

Essential Kubernetes Configuration

Implementing the code is only half the solution. Your Kubernetes deployment must be configured to leverage this pattern effectively:

terminationGracePeriodSeconds: Set a sufficient grace period in your deployment.yaml. This value dictates how long Kubernetes will wait after sending SIGTERM before forcibly killing the pod. A common value is 30 or 60 seconds, allowing ample time for tasks to drain.

apiVersion: apps/v1 kind: Deployment metadata: name: my-app spec: template: spec: terminationGracePeriodSeconds: 60 # Give the app 60 seconds to shut down containers: - name: my-app-container image: my-app-image:latest # ... other container settings

readinessProbe: Configure a readinessProbe that points to your /healthz endpoint. Adjust periodSeconds and failureThreshold to control how quickly Kubernetes detects the 503 status and stops sending traffic.

apiVersion: apps/v1 kind: Deployment metadata: name: my-app spec: template: spec: containers: - name: my-app-container image: my-app-image:latest readinessProbe: httpGet: path: /healthz port: 8000 initialDelaySeconds: 5 # Wait 5s before first probe periodSeconds: 5 # Check every 5 seconds failureThreshold: 3 # After 3 consecutive failures (503s), mark as NotReady # ... other container settings

By combining the in-application readiness flag with appropriate Kubernetes probe and termination settings, you build a robust mechanism for controlled, graceful server shutdowns. This ensures that your application can politely decline new work, finish existing tasks, and exit without causing service disruptions or data loss.

FastAPI WebSockets: Async Connections, Scaling, The Multi-Worker Nightmare (2026)

FastAPI WebSockets: Navigating State, Authentication, and Multi-Worker Scaling

FastAPI's WebSocket implementation often appears straightforward, mirroring the ease of building standard HTTP endpoints. This apparent simplicity, however, frequently conceals the underlying complexities of developing robust, scalable real-time applications. A common pitfall involves a WebSocket service functioning perfectly in a single-worker development environment, only to exhibit silent failures—like messages failing to broadcast—when deployed across multiple worker processes in production. This article explores critical architectural considerations to move beyond basic WebSocket examples and build truly production-ready, distributed real-time systems.

The Deceptive Simplicity of Basic WebSocket Implementations

FastAPI's WebSocket capabilities, leveraging Starlette, offer a clean, async/await syntax that feels familiar to anyone building HTTP APIs. This ease of use, however, can be misleading. Unlike the stateless nature of HTTP, where each request is independent, WebSockets maintain a persistent, stateful TCP connection. Failing to actively manage this long-lived connection's lifecycle can lead to resource leaks, event loop blockages, and unexpected server crashes. Many introductory examples overlook the critical exception handling necessary to gracefully manage client disconnections, such as when a user closes their browser tab or loses network connectivity.

The core misunderstanding often lies in treating WebSockets as merely extended HTTP requests. Production-grade WebSocket services demand meticulous state management, comprehensive error handling, and a solid grasp of the Python asyncio event loop. A single blocking operation within a WebSocket's message processing loop can halt all other concurrent connections on that worker process.

Consider an HTTP request as a quick transaction: you send a query, get a response, and the interaction concludes. A WebSocket, by contrast, is an ongoing conversation. The server must continuously monitor the connection. If the client abruptly ends the conversation without proper signaling, the server needs mechanisms to detect this and release the associated resources, preventing a 'phantom' connection from consuming memory indefinitely.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect import logging logger = logging.getLogger(__name__) app = FastAPI() # NEVER skip the try/except block. A dropped connection WILL crash the route. @app.websocket("/ws/echo") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() client_id = f"{websocket.client.host}:{websocket.client.port}" logger.info(f"Client {client_id} connected.") try: while True: # This awaits indefinitely until a message arrives data = await websocket.receive_text() await websocket.send_text(f"Server Echo: {data}") except WebSocketDisconnect as e: # This is expected behavior when a client leaves. Handle it cleanly. logger.info(f"Client {client_id} disconnected gracefully. Code: {e.code}") except Exception as e: # Catch everything else to prevent the worker thread from dying logger.error(f"Unexpected error with client {client_id}: {e}") finally: # Ensure cleanup happens even if the loop breaks unexpectedly logger.debug(f"Cleanup complete for {client_id}.")

Securing WebSocket Connections: Beyond Standard HTTP Headers

A common hurdle for backend engineers transitioning to WebSockets is authentication. The familiar pattern of using an Authorization: Bearer header for HTTP requests doesn't directly translate. Browser-based WebSocket APIs explicitly prevent custom headers during the initial handshake. This means attempting to pass a bearer token in the header of a client-initiated WebSocket request will fail, necessitating alternative, secure authentication strategies.

Avoid workarounds that compromise security. Embedding long-lived JSON Web Tokens (JWTs) directly in URL query parameters is highly insecure, as URLs are frequently logged by proxies, web servers, and browser history. If query parameters are unavoidable, implement a 'ticket' system: issue a short-lived, single-use token via a secure HTTP endpoint, then immediately consume it to establish the WebSocket connection. For browser-based single-page applications, HttpOnly cookies offer a robust solution, as the browser automatically includes domain-scoped cookies during the WebSocket handshake (which starts as an HTTP Upgrade request). For public APIs or mobile clients where cookies are less practical, the "First-Message Authentication" pattern provides a secure and flexible alternative.

Picture a private club: anyone can approach the entrance (connect the socket), but access to the main area is granted only after a valid password is whispered to the bouncer (sending an authentication payload as the very first message). Failure to provide the correct credentials, or a delay in doing so, results in immediate denial of entry (socket closure).

import asyncio from fastapi import status async def verify_token(token: str) -> bool: # Implementation details... return token == "valid-secret-token" @app.websocket("/ws/secure") async def secure_endpoint(websocket: WebSocket): await websocket.accept() try: # CRITICAL: Do not wait forever. If they don't auth fast, kill it. auth_msg = await asyncio.wait_for( websocket.receive_json(), timeout=5.0 ) token = auth_msg.get("token") if not token or not await verify_token(token): # Custom 4000+ close codes signify application-level errors await websocket.close(code=4001, reason="Unauthorized: Invalid Token") return except asyncio.TimeoutError: # They connected but didn't send the password fast enough await websocket.close(code=4002, reason="Auth Timeout") return except Exception: await websocket.close(code=status.WS_1008_POLICY_VIOLATION) return # If we reach here, the connection is authenticated. # We can now enter the main message loop. await websocket.send_json({"status": "authenticated"}) try: while True: data = await websocket.receive_text() # Process secure messages... except WebSocketDisconnect: pass

Scaling WebSockets: The Challenge of Distributed State

The most critical lesson for scalable WebSocket applications is this: in-memory connection managers are fundamentally incompatible with distributed deployments. While a simple ConnectionManager class storing active WebSocket objects works perfectly with a single Uvicorn process, production environments rarely operate this way. Deployments often involve multiple Uvicorn worker processes managed by Gunicorn, or numerous pods orchestrated by Kubernetes. These processes operate in isolation; they do not share memory. Consequently, if client A connects to worker 1 and client B connects to worker 3, worker 1 has no record of client B. Any attempt by client A to send a message intended for client B will fail silently, as worker 1 cannot route the message to a connection it doesn't manage.

FastAPI provides the transport layer for WebSockets, but it doesn't inherently offer a publish/subscribe (pub/sub) system. As soon as you scale beyond a single worker process or deploy across multiple server nodes, your WebSocket architecture transitions from a purely Python-centric challenge to a distributed systems problem. An external message broker becomes essential for synchronizing state and messages across all workers. Redis, with its robust Pub/Sub capabilities, is a widely adopted and practical solution for this.

Consider a network of independent call centers (your workers). If a customer calls center A and needs to relay information to another customer who called center C, center A cannot directly connect them. A central communication hub is required. Redis acts as this hub: when center A receives a message for a customer, it broadcasts it to the central hub. The hub then relays this message to all call centers. Only center C, which manages the target customer's connection, will pick up the message and deliver it.

import redis.asyncio as redis import json import asyncio from typing import Dict from fastapi import WebSocket class RedisPubSubManager: def __init__(self, redis_url: str = "redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.pubsub = self.redis.pubsub() # Local state for THIS specific worker process only self.active_connections: Dict[str, WebSocket] = {} async def connect(self, websocket: WebSocket, user_id: str): await websocket.accept() self.active_connections[user_id] = websocket # Worker subscribes to a global channel upon first connection await self.pubsub.subscribe("global_chat") def disconnect(self, user_id: str): if user_id in self.active_connections: del self.active_connections[user_id] async def publish_message(self, message: dict): # PUSH message to Redis. We don't send to local clients directly here. await self.redis.publish("global_chat", json.dumps(message)) async def listen_to_redis(self): # Background task that listens to Redis and broadcasts to LOCAL clients async for message in self.pubsub.listen(): if message["type"] == "message": payload = json.loads(message["data"].decode()) # Broadcast to all connections managed by THIS worker dead_connections = [] for uid, conn in self.active_connections.items(): try: await conn.send_json(payload) except Exception: # Catch dead sockets during broadcast to prevent loop crashing dead_connections.append(uid) # Cleanup dead connections for uid in dead_connections: self.disconnect(uid) manager = RedisPubSubManager() # You MUST start the Redis listener task when the app starts @app.on_event("startup") async def startup_event(): asyncio.create_task(manager.listen_to_redis())

This architecture ensures that each worker publishes messages to a shared message bus (Redis) and simultaneously subscribes to that same bus. When a message arrives on the bus, every worker receives it and then forwards it to any relevant clients connected to that specific worker. This design enables seamless horizontal scaling across numerous processes and nodes, preventing message loss in distributed environments.