DEV Community: Ayush Kumar

FastAPI testing with pytest: practical patterns for reliable code

Ayush Kumar — Thu, 20 Aug 2026 09:48:52 +0000

FastAPI testing with pytest is straightforward once you have the right scaffolding. In a few minutes you can spin up a TestClient, hit both sync and async endpoints, and get deterministic results without touching a real database. Below I’ll show you exactly how I set it up in production, the shortcuts that bite me, and the trade-offs you should keep in mind.

How do I set up pytest for a FastAPI project?

The first step is to make pytest aware of your application package and any fixtures you need. I keep a tests/ directory at the repo root and add a tiny conftest.py that creates the FastAPI app instance and a shared TestClient.

# conftest.py
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from myapp.main import create_app  # factory that builds the FastAPI instance

@pytest.fixture(scope="session")
def app() -> FastAPI:
    """Create a single FastAPI app for the whole test session."""
    return create_app()

@pytest.fixture(scope="session")
def client(app: FastAPI) -> TestClient:
    """Wrap the app in a TestClient."""
    return TestClient(app)

A factory function (create_app) is a must-have. It lets you inject different settings (e.g., a test config) without polluting the production startup path. I keep my settings in Pydantic models and switch to a TestSettings class in CI.

# settings.py
from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://prod:password@db/prod"

class TestSettings(Settings):
    database_url: str = "sqlite+aiosqlite:///./test.db"

When create_app receives a settings object you can hand it TestSettings in the fixture. This isolates the test environment from production credentials – a mistake that has cost me a few hard-to-track bugs.

Can I use TestClient for both sync and async endpoints?

Yes, TestClient works for both. Under the hood it runs the ASGI app in a thread-pooled httpx client, so you can call async routes just like sync ones.

def test_sync_hello(client: TestClient):
    response = client.get("/hello")
    assert response.status_code == 200
    assert response.json() == {"msg": "hello"}

@pytest.mark.asyncio
async def test_async_echo(client: TestClient):
    payload = {"msg": "ping"}
    response = client.post("/echo", json=payload)
    assert response.status_code == 200
    assert response.json() == payload

Notice the @pytest.mark.asyncio decorator. It tells pytest to run the test in an event loop. If you forget it, the test will hang because the coroutine never gets scheduled. I once ran a whole suite without the decorator and spent an hour chasing a silent timeout.

How do I override dependencies in tests?

FastAPI’s dependency_overrides dict is a clean way to replace heavy dependencies (like external APIs or auth services) with lightweight mocks.

# tests/conftest.py (continued)
from myapp.dependencies import get_current_user, get_redis

@pytest.fixture(autouse=True)
def override_deps(app: FastAPI):
    async def fake_user():
        class User:
            id = 1
            username = "test_user"
        return User()

    async def fake_redis():
        class DummyRedis:
            async def get(self, key): return None
            async def set(self, key, value, ex=None): pass
        return DummyRedis()

    app.dependency_overrides[get_current_user] = fake_user
    app.dependency_overrides[get_redis] = fake_redis
    yield
    app.dependency_overrides.clear()

With autouse=True the overrides apply to every test automatically. This saves you from sprinkling client.app.dependency_overrides[...] = ... in each test file. The pattern works for both sync and async dependencies. Just remember to clear the overrides after the test session; otherwise you’ll leak state into later runs.

What’s the best way to test authentication and security schemes?

FastAPI ships with OAuth2 password flow, API keys, and custom schemes. The trick is to generate a valid token once and reuse it across tests. I store the token in a module-level fixture.

# tests/auth_fixtures.py
import pytest
from httpx import AsyncClient

@pytest.fixture(scope="session")
async def auth_token(app: FastAPI):
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/token", data={"username": "alice", "password": "secret"})
        return response.json()["access_token"]

Now any endpoint that requires Depends(oauth2_scheme) can be exercised with the header:

def test_protected_route(client: TestClient, auth_token: str):
    headers = {"Authorization": f"Bearer {auth_token}"}
    response = client.get("/protected", headers=headers)
    assert response.status_code == 200
    assert response.json()["user"] == "alice"

If you’re using a third-party SSO provider, mock the introspection endpoint with responses or httpx-mock. Don’t hit the real provider during CI – it adds latency and can get you rate-limited.

How do I manage database fixtures and transaction rollbacks?

Testing with a real DB gives confidence, but you need isolation. I prefer a per-test transaction that rolls back automatically. With SQLAlchemy 2.0 async support the pattern looks like this:

# tests/db_fixtures.py
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from myapp.db import Base, get_db
from myapp.settings import TestSettings

engine = create_async_engine(TestSettings().database_url, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@pytest.fixture(scope="function")
async def db_session():
    async with AsyncSessionLocal() as session:
        # start a nested transaction (SAVEPOINT)
        async with session.begin_nested():
            yield session
        # rollback happens automatically when the context exits

Then override the get_db dependency to return this session:

# conftest.py (continued)
from myapp.dependencies import get_db
from .db_fixtures import db_session

@pytest.fixture(autouse=True)
def override_db(app: FastAPI, db_session):
    async def _get_test_db():
        yield db_session
    app.dependency_overrides[get_db] = _get_test_db
    yield
    app.dependency_overrides.clear()

Each test runs inside a SAVEPOINT, so even if you commit inside the test the outer transaction rolls back when the fixture tears down. This keeps the test DB pristine without recreating the schema each run. The downside: it only works with databases that support SAVEPOINT (Postgres, MySQL). SQLite in memory works, but you lose some concurrency semantics. If you need full isolation across processes, spin up a fresh Docker container per test suite – heavy, but sometimes necessary.

When should I avoid the TestClient approach?

TestClient is perfect for unit-style API tests, but it doesn’t exercise the full ASGI server stack. If you rely on middleware that interacts with the network (e.g., HTTP/2, custom TLS termination) you’ll need an integration test that runs the app with a real server like uvicorn. Also, heavy load tests belong to a separate suite that uses locust or hey; TestClient can’t simulate concurrent users accurately.

Real-world pitfalls I’ve hit

MissingGreenlet errors – When I first switched to async SQLAlchemy I forgot to install greenlet. The error surfaced only in tests because the sync test client spawns a thread. The fix is documented in my post about Fixing SQLAlchemy MissingGreenlet Error in FastAPI (Async Explained).
Environment leakage – Global state in a module (e.g., a singleton cache) survived between tests. I solved it by resetting the cache in a fixture’s teardown step.
CI timeouts – Running the full DB fixture suite on GitHub Actions took >10 minutes. I introduced a --fast marker that skips heavy integration tests on PR builds and only runs them on the nightly pipeline. The trade-off is slower feedback for complex scenarios, but the overall CI cost dropped dramatically.

Putting it all together

Here’s a minimal end-to-end test file that touches every piece we discussed:

# tests/test_user_flow.py
import pytest
from fastapi.testclient import TestClient

def test_register_and_login(client: TestClient):
    # Register a new user
    reg = client.post("/register", json={"username": "bob", "password": "pwd123"})
    assert reg.status_code == 201

    # Login to get a token
    token_resp = client.post("/token", data={"username": "bob", "password": "pwd123"})
    assert token_resp.status_code == 200
    token = token_resp.json()["access_token"]

    # Access a protected endpoint
    headers = {"Authorization": f"Bearer {token}"}
    protected = client.get("/me", headers=headers)
    assert protected.status_code == 200
    assert protected.json()["username"] == "bob"

@pytest.mark.asyncio
async def test_async_data_flow(client: TestClient):
    # Assume an async endpoint that writes to DB
    payload = {"title": "Test note", "content": "Hello"}
    resp = client.post("/notes", json=payload)
    assert resp.status_code == 201
    note_id = resp.json()["id"]

    # Fetch the note back
    get_resp = client.get(f"/notes/{note_id}")
    assert get_resp.status_code == 200
    data = get_resp.json()
    assert data["title"] == payload["title"]
    assert data["content"] == payload["content"]

Running pytest -q now executes both sync and async paths, uses the overridden dependencies, rolls back DB changes, and never touches production credentials. The same suite runs unchanged when I deploy to Google Cloud Run – see my article on Serverless Python: Deploying FastAPI to Google Cloud Run with Docker for the CI/CD wiring.

FAQ

How do I test websockets with pytest?

Create an AsyncClient from httpx and use client.ws_connect("/ws"). Remember to mark the test with @pytest.mark.asyncio and close the connection after assertions.

Can I use pytest-asyncio with the built-in TestClient?

Yes. The TestClient works in an async test as long as you wrap calls in await client.get(...) via client.__call__ which returns a coroutine. Most people prefer httpx.AsyncClient for pure async tests.

What’s the fastest way to spin up a test DB?

For PostgreSQL, use docker compose up -d db-test and point TestSettings.database_url to that container. Keep the container alive across the test session to avoid repeated image pulls.

Do I need to test OpenAPI schema generation?

Usually not. FastAPI guarantees schema correctness if your Pydantic models are valid. Focus on request/response behavior instead.

Key Takeaways

Use a factory (create_app) and a TestSettings class to keep production and test configurations separate.
TestClient handles both sync and async routes; just add @pytest.mark.asyncio for the latter.
Override heavy dependencies via app.dependency_overrides in a session-wide fixture.
Authenticate once per session and reuse the token to keep tests fast.
Wrap each test in a nested DB transaction; it rolls back automatically and gives you isolation without recreating schemas.
Reserve real server runs for integration or load testing; TestClient is not a replacement for those scenarios.

Happy testing!

How to Fix the copilot error code rate_limited in Production

Ayush Kumar — Thu, 20 Aug 2026 09:48:37 +0000

If you’re seeing the copilot error code rate_limited, it means GitHub Copilot has throttled your request because you exceeded the allowed request quota for your account or organization. The fix is simple: detect the response, pause, retry with back-off, and, if needed, adjust your quota or switch to a fallback model. Below I walk through the exact steps I use in production, from spotting the error in VS Code to wiring a resilient retry layer in Python.

What triggers the “rate_limited” error in GitHub Copilot?

GitHub Copilot enforces a per-minute request limit per user and a higher limit per organization. The limit is calculated on the number of autocomplete suggestions, inline completions, and chat-style calls you make. When you cross that threshold, the service returns an HTTP 429 response with a body that includes error_code: "rate_limited".

Typical triggers:

Trigger	Why it happens
Running a batch script that calls Copilot Chat for every line of a large codebase	Thousands of calls in seconds exceed the per-minute cap
Using a hot-key that fires on every keystroke in a large file	Autocomplete requests flood the API
Multiple developers sharing the same organization quota	Collective usage spikes push the org over its limit

In my own projects, the most common surprise came from a CI job that generated doc-strings for every function using Copilot Chat. The job blew past the limit within the first minute, and the pipeline failed with the cryptic rate_limited message.

How to detect and log Copilot rate-limit responses in your IDE

The first line of defense is to surface the error as soon as it arrives. In VS Code the Copilot extension already logs HTTP responses to the output panel, but the messages are easy to miss. I added a small wrapper around the extension’s API (via the @github/copilot npm package) that writes a structured log entry to a local file.

import json
import time
from pathlib import Path

LOG_FILE = Path.home() / ".copilot_rate_limit.log"

def log_rate_limit(payload: dict):
    entry = {
        "timestamp": time.time(),
        "error_code": payload.get("error_code"),
        "message": payload.get("message"),
        "request_id": payload.get("request_id")
    }
    LOG_FILE.write_text(json.dumps(entry) + "\n", encoding="utf-8", append=True)

# Example: a mock response handler
def handle_copilot_response(resp):
    if resp.status_code == 429 and resp.json().get("error_code") == "rate_limited":
        log_rate_limit(resp.json())
        raise RuntimeError("Copilot rate limit hit")
    return resp.json()

The log file gives you a timeline of when the throttling started, which is invaluable for tuning your usage. I also ship a tiny VS Code task that tails the file and pops a warning banner when a new entry appears.

Implementing exponential back-off and retry logic for Copilot calls

Simply sleeping a fixed 5 seconds after a 429 rarely works because the limit window slides. The reliable pattern is exponential back-off with jitter. Here’s a reusable decorator I keep in my utils module:

import random
import functools
import time
import httpx

def backoff_retry(max_tries: int = 5, base_delay: float = 1.0):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            delay = base_delay
            for attempt in range(1, max_tries + 1):
                try:
                    return await func(*args, **kwargs)
                except httpx.HTTPStatusError as exc:
                    if exc.response.status_code != 429:
                        raise
                    # Copilot specific payload check
                    if exc.response.json().get("error_code") != "rate_limited":
                        raise
                    if attempt == max_tries:
                        raise RuntimeError("Max retries exceeded for Copilot")
                    jitter = random.uniform(0, delay)
                    time.sleep(delay + jitter)
                    delay *= 2  # exponential growth
        return wrapper
    return decorator

# Example usage with httpx.AsyncClient
@backoff_retry()
async def fetch_completion(prompt: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.githubcopilot.com/v1/completions",
            json={"prompt": prompt},
            headers={"Authorization": f"Bearer {COPILOT_TOKEN}"}
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["text"]

The decorator catches the specific rate_limited payload, waits a growing amount of time, and adds a random jitter to avoid stampeding. In production I set max_tries=7 and base_delay=0.5 seconds, which keeps the average latency under 10 seconds even under heavy load.

Trade-offs

Latency vs. quota usage – Back-off reduces the number of retries, but each retry adds latency. For interactive IDE use you might cap retries at 3; for batch jobs you can afford more.
Complexity – Adding async retry logic means you need an async-compatible HTTP client (httpx works nicely). If your codebase is sync-only, you’ll need to wrap the async call with anyio.run or switch to requests with a sync decorator.

Adjusting Copilot usage quotas and organization settings

If back-off still isn’t enough, it’s time to look at the quota itself. GitHub offers two levers:

User-level purchase – Individual developers can buy extra “Copilot credits” that raise the per-minute limit.
Organization-level allocation – Admins can increase the shared pool from the GitHub settings page under Copilot → Usage.

When I moved my team from a solo indie project to a small startup, I requested a quota bump for the org. The admin console shows the current limit (e.g., 1 200 requests per minute) and the usage over the last 24 hours. After the bump, the same CI job that previously failed now runs uninterrupted.

Caution: Raising the quota costs money and can mask underlying inefficiencies. Before you pay, audit your code for unnecessary calls. In one case I discovered that a linter plugin was asking Copilot for a suggestion on every import statement – turning that off saved 30 % of our quota.

Fallback strategies: switching to local LLMs or alternative AI assistants

When you can’t guarantee a steady quota (e.g., during a product launch or a hackathon), having a local fallback prevents a hard stop. Two practical options:

Option	When it shines	Setup cost
Run a self-hosted Llama-2 or Mistral model	Low-latency, offline, no external limits	Requires GPU, Docker orchestration
Swap to an alternative AI assistant (e.g., Cursor, Claude)	Multi-model flexibility, can split load	Need API keys, adjust request format

Below is a minimal FastAPI endpoint that tries Copilot first, then falls back to a local Llama-2 instance served via vLLM.

from fastapi import FastAPI, HTTPException
import httpx
import asyncio

app = FastAPI()
COPILOT_URL = "https://api.githubcopilot.com/v1/completions"
LLAMA_URL = "http://localhost:8000/v1/completions"

async def copilot_call(prompt: str):
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            COPILOT_URL,
            json={"prompt": prompt},
            headers={"Authorization": f"Bearer {COPILOT_TOKEN}"}
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["text"]

async def llama_call(prompt: str):
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            LLAMA_URL,
            json={"prompt": prompt, "max_tokens": 256}
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["text"]

@app.post("/complete")
async def complete(prompt: str):
    try:
        return {"source": "copilot", "completion": await copilot_call(prompt)}
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 429 and exc.response.json().get("error_code") == "rate_limited":
            # Log and fall back
            LOG_FILE.write_text(
                json.dumps({"fallback": "llama", "timestamp": time.time()}) + "\n",
                encoding="utf-8",
                append=True,
            )
            return {"source": "llama", "completion": await llama_call(prompt)}
        raise HTTPException(status_code=500, detail="Unexpected error")

The endpoint returns the source of the completion, making it easy to monitor how often you fall back. In my own service, the fallback rate stayed under 5 % after I tuned the back-off parameters and increased the org quota.

When NOT to use a fallback

If your codebase relies on Copilot-specific prompts (e.g., # Copilot: generate test cases) the fallback model may not understand the syntax, leading to lower quality suggestions. In those cases, it’s better to pause the job and alert a human rather than serve a sub-par output.

Putting it all together in a production pipeline

Detect – Add the logging snippet to your IDE or CI runner.
Throttle – Wrap every Copilot request with the backoff_retry decorator.
Monitor – Export the rate-limit log to your observability platform (Datadog, Prometheus) and set alerts.
Scale – If alerts fire frequently, request a quota increase or redistribute load across multiple accounts.
Fallback – Deploy a lightweight local LLM behind a feature flag; switch only when the rate-limit metric spikes.

I’ve applied this exact flow to a FastAPI service that generates unit tests on demand. The first version kept hitting copilot error code rate_limited during a sprint, costing the team hours of debugging. After adding back-off, logging, and a 10 % quota bump, the error vanished. The fallback to a local model only triggered on the rare days when my VPN throttled outbound traffic.

If you’re stuck at any point and need a pair of hands to audit your usage or stitch together a fallback, feel free to drop a line on the hire page. I’m happy to help you get the AI assistant running smoothly again.

FAQ

Q: Does the rate-limit apply per file or per request?

A: It’s per request. Every autocomplete, inline suggestion, or chat call counts as one request, regardless of file size.

Q: Can I disable Copilot for a single project to avoid hitting the limit?

A: Yes. In VS Code you can set "github.copilot.enable": false in the workspace settings, which stops all calls from that project.

Q: How long does the rate-limit window last?

A: GitHub uses a rolling one-minute window. After you stop sending requests, the limit resets gradually as the window slides.

Q: Will using multiple GitHub accounts increase my total quota?

A: Each account has its own limit. Sharing a repo across accounts aggregates usage under the organization’s quota if you enable org-wide sharing.

Key Takeaways

The copilot error code rate_limited means you’ve exceeded the per-minute request quota.
Log the error early; a simple JSON

Cursor AI code editor tutorial: setup, tips, and CI/CD

Ayush Kumar — Thu, 20 Aug 2026 09:48:17 +0000

Quick answer: how do I get Cursor AI working for my FastAPI project?

Install the Cursor AI extension, point it at your virtual environment, let it suggest routes and data models, then add a linting step and a CI job that runs the same prompts in a headless mode. In a few minutes you’ll have a code-assistant that can generate, refactor, and even review your FastAPI code without leaving the editor.

I’ve been building FastAPI services for years, and the first time I tried Cursor AI I hit a wall: the assistant kept suggesting imports that didn’t exist in my Docker image, and the generated code broke my type checks. The fix was simple – configure the tool to see the exact same environment you run in production, and add a small wrapper that runs the same prompts as part of your CI pipeline. Below is a full cursor ai code editor tutorial that shows you how to avoid those pitfalls and get reliable AI-assisted development.

Installing and configuring Cursor AI on your workstation

What do I need to download and how do I make Cursor see my Python packages?

The answer is: the VS Code extension and a cursor.toml that points to your interpreter.

Get the extension – Open VS Code, go to Extensions, search for “Cursor” and click Install.
Create a config file – In the root of your repo, add cursor.toml:

[project]
python_path = "./.venv/bin/python"   # absolute path works too
requirements = "requirements.txt"

Refresh the environment – Open the Command Palette (Ctrl+Shift+P) and run Cursor: Refresh Project. Cursor now parses the same packages you have installed locally.

If you skip the python_path line, Cursor runs its own sandboxed interpreter. That sandbox can’t import your private libs, and you’ll see errors like ModuleNotFoundError: my_utils. I learned that the hard way during a sprint where the AI kept suggesting from my_utils import xyz and the code never compiled.

Common failure mode

Error: “Unable to resolve import fastapi”.

Why it happens: Cursor is using a different Python version than your project.

Fix: Make sure the path in cursor.toml points to the exact interpreter you use for uvicorn. Run which python inside your virtualenv and copy that path.

Setting up Cursor AI for Python and FastAPI projects

Can Cursor understand FastAPI’s async routing and Pydantic models?

Yes, but you have to give it a hint.

Add a small stub file named cursor_fastapi_stub.py:

# cursor_fastapi_stub.py – only for Cursor’s type inference
from fastapi import FastAPI, APIRouter, Depends
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., description="Item name")
    price: float

app = FastAPI()
router = APIRouter()

Place this file anywhere in the repo and add its path to cursor.toml:

[project]
extra_paths = ["./cursor_fastapi_stub.py"]

Now when you type @app.get("/items") or ask Cursor “create a POST endpoint for Item”, the assistant knows the exact shape of the model and produces code that passes mypy and pytest out of the box.

Real-world example

# main.py
from fastapi import FastAPI, HTTPException
from cursor_fastapi_stub import Item

app = FastAPI()

@app.post("/items")
async def create_item(item: Item):
    # Cursor can suggest validation here
    if item.price <= 0:
        raise HTTPException(status_code=400, detail="Price must be positive")
    return {"id": 123, **item.dict()}

When I first ran this, Cursor suggested a uuid4 import that wasn’t in my requirements.txt. Adding the stub and updating the config stopped the random import suggestions.

Using Cursor AI for code generation, refactoring, and debugging

What concrete commands let me generate FastAPI routes or fix bugs?

Cursor works through natural-language prompts. Here are the three most useful ones:

Prompt	What Cursor does
“Create a GET endpoint `/users/{id}` that returns a Pydantic `User` model”	Generates the route, imports, and a stub `User` model if missing.
“Refactor `create_item` to use a service layer”	Moves the business logic into `services/item.py` and updates the import.
“Find why the test `test_create_item` fails on validation”	Opens the test, runs a quick static analysis, and points out the missing `await` on the endpoint call.

Debugging with Cursor

When a test fails, open the test file, select the failing line, and press Ctrl+Shift+L (default “Explain”). Cursor will return a short explanation, the line that caused the failure, and a suggested fix. In production I’ve seen this cut debugging time from 30 minutes to under 5 minutes.

Trade-offs

Speed vs. correctness – Cursor can spew code fast, but you still need to run your test suite. Don’t merge AI-generated PRs without CI.
Complex logic – For deep algorithmic work, the assistant may hallucinate. Use it for scaffolding, not for the core business algorithm.

Leveraging Cursor AI’s built-in code review and linting features

Does Cursor replace my linter or code reviewer?

Not entirely, but it can act as a first line of defense.

Enable the built-in review by adding to cursor.toml:

[review]
enabled = true
linters = ["flake8", "black"]

When you save a file, Cursor runs the selected linters and shows warnings in the editor gutter. It also adds a “Code Review” comment block at the top of the diff with suggestions like “Replace assert with proper exception handling”.

I once merged a PR where Cursor flagged a hard-coded secret (API_KEY = "abcd"). The review step caught it before the code reached staging. That’s why I always keep the review turned on, especially for AI-generated code.

For a deeper dive on security, check out the ai generated code platform security checklist. It walks you through the exact steps to audit secrets, dependencies, and permissions.

Best practices and security considerations for AI-assisted coding

What should I watch out for when letting an AI write code?

Never trust imports – Always verify that any new third-party package appears in requirements.txt. Cursor may suggest the latest version of a library that has a known CVE.
Run static analysis – Add a bandit step in CI to catch insecure patterns that the AI might introduce.
Limit data exposure – If you use Cursor’s cloud mode, it sends snippets of your code to the provider. Keep proprietary logic out of the prompts or run the self-hosted version if you have compliance concerns.
Version pinning – When Cursor adds a dependency, pin it to a specific version. Unpinned ranges can cause builds to break later.

I learned this the hard way when a generated requests call used verify=False. The app worked locally, but security scans flagged it in staging and blocked the release. Adding bandit caught the issue before it went live.

For a practical guide on detecting AI-generated code issues, see Fixing AI Generated Code Quality Issues in Production.

Integrating Cursor AI workflows into CI/CD pipelines

Can I automate Cursor’s suggestions in my GitHub Actions?

Yes. The trick is to run the same prompt engine in a headless mode that outputs a diff, then let the CI job apply the diff if it passes tests.

Add a step to your workflow:

name: CI
on: [push, pull_request]

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - name: Install deps
        run: |
          python -m venv .venv
          source .venv/bin/activate
          pip install -r requirements.txt
      - name: Run Cursor suggestions
        env:
          CURSOR_API_KEY: ${{ secrets.CURSOR_API_KEY }}
        run: |
          curl -X POST https://api.cursor.com/v1/generate \
            -H "Authorization: Bearer $CURSOR_API_KEY" \
            -d @cursor_prompt.json \
            -o cursor_diff.patch
          git apply cursor_diff.patch || exit 1
      - name: Run tests
        run: |
          source .venv/bin/activate
          pytest -q
      - name: Lint
        run: |
          flake8 .

cursor_prompt.json contains a JSON payload like:

{
  "project_path": ".",
  "instruction": "Add type hints to all FastAPI endpoint functions",
  "output_format": "patch"
}

If the patch fails to apply, the job stops and you get a clear error. This pattern gives you the safety of a human review while still reaping the speed benefits of AI.

When NOT to use this

Critical security patches – Let a human write the fix, then let Cursor review it.
Large refactors – The diff can be massive, making CI noisy and hard to debug.

When to stop relying on Cursor

Cursor shines when you need repetitive scaffolding: creating CRUD endpoints, writing Pydantic models, or fixing simple lint errors. It falters on domain-specific logic, performance-critical code, or anything that requires deep business knowledge. In those cases, treat the AI as a junior teammate: let it draft, then have a senior engineer polish.

Getting hands-on help

If you’ve tried the steps above and still hit a wall - maybe your CI job keeps failing or the AI keeps suggesting insecure code - consider bringing in an experienced engineer for a few hours. I’m available for short-term contracts; you can learn more on the /hire/ page.

FAQ

Q: Does Cursor work with other editors besides VS Code?

A: Cursor currently offers extensions for VS Code and JetBrains IDEs. The configuration file (cursor.toml) is the same, so you can switch editors without changing anything else.

Q: How do I keep Cursor from sending my proprietary code to the cloud?

A: Use the self-hosted version of Cursor, which runs entirely on your machine, or enable the “local only” flag in the extension settings.

Q: Can Cursor format code automatically?

A: Yes. Enable black in the [review] section of cursor.toml. On each save Cursor will apply Black formatting and show any changes in the gutter.

Q: What if Cursor suggests a dependency that has a known vulnerability?

A: Run pip-audit or bandit in your CI pipeline. If a vulnerable package appears, block the merge and ask Cursor for an alternative library.

Key Takeaways

Install the Cursor VS Code extension and point it at the exact Python interpreter you use in production.
Add a small stub file for FastAPI and Pydantic so Cursor can generate correct type hints.
Use natural-language prompts for scaffolding, refactoring, and debugging, but always run your test suite.
Enable built-in linting and code review; treat the output as a first pass, not a final commit.
Apply a security checklist: verify imports, pin versions, run static analysis, and limit data sent to the cloud.
Automate Cursor’s suggestions in CI with a headless API call that produces a patch, then run tests on the patched code.
Reserve human expertise for core business logic and security-critical changes.

By following this cursor ai code editor tutorial, you’ll avoid the common pitfalls that trip up many indie builders and keep your FastAPI services reliable, secure, and ready for production.

RAG pipeline diagram: design, build, and scale

Ayush Kumar — Mon, 10 Aug 2026 08:25:36 +0000

Introduction

If you need a quick answer: a rag pipeline diagram maps the flow from raw documents through chunking, embedding, vector storage, retrieval, and finally LLM generation. I’ve built several of these in FastAPI, and the diagram helped me spot bottlenecks before they became outages. Below I walk through the core components, a step-by-step build, visualisation tricks with Mermaid and PlantUML, integration tips for vector stores and LLMs, and the scaling and monitoring practices that keep the pipeline healthy in production.

What are the core components of a RAG pipeline?

The first sentence answers the question: a RAG pipeline consists of ingestion, embedding, vector store, retrieval, and generation modules, all wired together by a thin API layer.

Component	Responsibility	Typical tech
Ingestion	Pull raw text from PDFs, DB rows, webhooks	`pdfminer`, `requests`, async generators
Chunking	Split documents into manageable pieces (≈200-400 tokens)	RecursiveCharacterTextSplitter, custom regex
Embedding	Convert chunks to dense vectors	OpenAI `text-embedding-ada-002`, HuggingFace `sentence-transformers`
Vector Store	Persist and index embeddings for fast similarity search	FAISS (local), Pinecone, Qdrant
Retrieval	Perform nearest-neighbor search, optionally apply hybrid filters	`faiss.IndexFlatIP`, Pinecone query API
Generation	Feed retrieved context to an LLM and stream the answer	OpenAI `gpt-4`, Llama-2 via vLLM
API / Orchestration	Expose the flow over HTTP, add caching, retries, logging	FastAPI, Redis, Celery

When any of these steps fails, the whole service can stall. I’ve been bitten by slow chunking (CPU bound) and by vector store latency spikes when the index grows beyond RAM. The diagram makes those dependencies explicit, so you can allocate resources where they matter.

How do I build a RAG pipeline step by step?

The answer is right up front: start with a minimal FastAPI endpoint, add async ingestion, plug in an embedding model, persist vectors, then layer retrieval and generation on top. Below is a minimal, production-ready skeleton.

# app/main.py
import os
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import aiohttp
import asyncio
import numpy as np
import faiss
import openai

app = FastAPI(title="RAG Service")

# ---- Config -------------------------------------------------
EMBEDDING_MODEL = "text-embedding-ada-002"
VECTOR_DIM = 1536
INDEX_PATH = "./faiss.index"
openai.api_key = os.getenv("OPENAI_API_KEY")

# ---- In‑memory FAISS index ----------------------------------
if os.path.exists(INDEX_PATH):
    index = faiss.read_index(INDEX_PATH)
else:
    index = faiss.IndexFlatIP(VECTOR_DIM)

# ---- Pydantic models ---------------------------------------
class QueryRequest(BaseModel):
    query: str
    top_k: int = 5

# ---- Helper functions ---------------------------------------
async def fetch_document(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            if resp.status != 200:
                raise HTTPException(status_code=502, detail="Document fetch failed")
            return await resp.text()

def embed(text: str) -> np.ndarray:
    resp = openai.Embedding.create(input=text, model=EMBEDDING_MODEL)
    return np.array(resp["data"][0]["embedding"], dtype="float32")

def add_to_index(vec: np.ndarray, metadata: dict):
    index.add(np.expand_dims(vec, 0))
    # In real code you would also persist metadata in a DB
    faiss.write_index(index, INDEX_PATH)

def retrieve(query_vec: np.ndarray, k: int):
    distances, ids = index.search(np.expand_dims(query_vec, 0), k)
    # Placeholder: return dummy texts; replace with DB lookup
    return ["doc #{}".format(i) for i in ids[0]]

async def generate_answer(contexts: list[str], query: str) -> str:
    prompt = "\n\n".join(contexts) + f"\n\nQuestion: {query}\nAnswer:"
    resp = await openai.ChatCompletion.acreate(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content.strip()

# ---- API routes ---------------------------------------------
@app.post("/query")
async def query(req: QueryRequest, background: BackgroundTasks):
    # 1. Embed the query
    q_vec = embed(req.query)

    # 2. Retrieve top‑k chunks
    docs = retrieve(q_vec, req.top_k)

    # 3. Generate answer
    answer = await generate_answer(docs, req.query)

    # 4. Fire‑and‑forget async logging
    background.add_task(log_interaction, req.query, answer)

    return {"answer": answer, "sources": docs}

async def log_interaction(query: str, answer: str):
    # Replace with proper structured logging / DB write
    print(f"[RAG] query={query!r} answer={answer[:50]!r}")

What the code shows

Async ingestion – fetch_document runs in an event loop, preventing the worker from blocking on I/O.
Embedding – a thin wrapper around OpenAI’s API; you can swap in a local model with the same signature.
Vector store – FAISS lives in RAM; I persist the index on disk after each write. For larger corpora you’d move to Pinecone or Qdrant.
Retrieval – simple nearest-neighbor search; add metadata filters if needed.
Generation – streamed via acreate to keep the request non-blocking.

You can expand this skeleton with background chunking workers, a message queue (Celery or RabbitMQ), and a Redis cache for repeated queries. I learned the hard way that mixing sync openai.Embedding.create with async FastAPI caused thread-pool exhaustion. Switching to the async version solved the problem.

How can I visualize a rag pipeline diagram with Mermaid or PlantUML?

You can generate a live diagram directly in your README or internal docs; the first sentence explains that both tools accept a textual DSL that maps nicely to the component table above.

Mermaid example

flowchart LR
    A[Ingestion] --> B[Chunker]
    B --> C[Embedding Service]
    C --> D[Vector Store (FAISS)]
    D --> E[Retriever]
    E --> F[LLM (GPT‑4)]
    F --> G[FastAPI Response]
    style D fill:#f9f,stroke:#333,stroke-width:2px

Copy the block into any Markdown renderer that supports Mermaid (GitHub, MkDocs, VS Code preview) and you have a living rag pipeline diagram.

PlantUML alternative

@startuml
skinparam backgroundColor #EFEFEF
node "Ingestion" as I
node "Chunker" as Ck
node "Embedding\nService" as E
node "FAISS Index" as V
node "Retriever" as R
node "LLM (GPT‑4)" as L
node "FastAPI\nResponse" as F

I --> Ck --> E --> V --> R --> L --> F
@enduml

PlantUML is handy when you need SVG output for documentation pipelines that don’t support Mermaid. Both formats let you version-control the diagram as code, which is a huge win for collaboration.

How do I integrate vector stores, LLMs, and retrieval modules?

The short answer: treat each as a pluggable service behind a thin adapter, and keep the contract to “vector + metadata” and “text prompt → answer”.

Vector store adapters

class VectorStore:
    def add(self, vec: np.ndarray, meta: dict): ...
    def search(self, vec: np.ndarray, k: int) -> list[dict]: ...

class FAISSStore(VectorStore):
    def __init__(self, dim: int, path: str):
        self.index = faiss.read_index(path) if os.path.exists(path) else faiss.IndexFlatIP(dim)
        self.path = path

    def add(self, vec, meta):
        self.index.add(np.expand_dims(vec, 0))
        # persist meta somewhere (PostgreSQL, DynamoDB)
        faiss.write_index(self.index, self.path)

    def search(self, vec, k):
        d, i = self.index.search(np.expand_dims(vec, 0), k)
        # fetch metadata by ids
        return [{"id": int(idx), "score": float(score)} for idx, score in zip(i[0], d[0])]

Swap FAISSStore for PineconeStore or QdrantStore without touching the rest of the pipeline.

LLM adapters

class LLM:
    async def complete(self, prompt: str) -> str: ...

class OpenAIChat(LLM):
    async def complete(self, prompt):
        resp = await openai.ChatCompletion.acreate(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
        )
        return resp.choices[0].message.content.strip()

Both adapters expose a single method, making unit testing trivial: inject a mock that returns deterministic text.

Retrieval module

class Retriever:
    def __init__(self, store: VectorStore, embed_fn):
        self.store = store
        self.embed = embed_fn

    async def retrieve(self, query: str, k: int = 5):
        q_vec = self.embed(query)
        hits = self.store.search(q_vec, k)
        # Load full chunk text from DB using hit["id"]
        return [await load_chunk(hit["id"]) for hit in hits]

The pattern keeps the rag pipeline diagram clean – each box is an interface, not a concrete implementation.

How do I scale and monitor a rag pipeline in production?

Scaling starts with separating concerns: ingestion workers, embedding workers, and query workers each get their own autoscaling group. The first sentence says that you should instrument every hop with latency histograms and error counters, then feed them to Prometheus and Grafana.

Horizontal scaling

Ingestion – run a Celery beat schedule or Cloud Pub/Sub subscriber that pulls new docs and pushes them to a task queue.
Embedding – batch embeddings to respect rate limits; use a pool of GPU-enabled workers if you switch to a local model.
Query – FastAPI can run behind Uvicorn workers behind an ALB; each worker holds a read-only copy of the vector index if it fits in RAM. For larger indexes, use a remote store (Pinecone) that auto-scales.

Monitoring checklist

Metric	Why it matters	Typical alert
`ingest_latency_seconds`	Detect slow I/O or parsing errors	> 5 s
`embed_rate_errors_total`	OpenAI quota or model downtime	> 0 in 5 min
`retrieval_latency_seconds`	Vector store overload	> 200 ms
`generation_latency_seconds`	LLM throttling	> 2 s
`api_5xx_total`	End-to-end failure	> 1 per minute

I once saw a spike in retrieval_latency_seconds after a weekend data dump; the FAISS index grew beyond RAM and the OS started swapping. The alert caught it before users hit timeouts. The fix was to shard the index across two nodes and add a health-check that restarts the service if memory usage exceeds 80 %.

Cost considerations

Embedding – OpenAI pricing is per 1 k tokens; batch 100 docs per request to reduce overhead.
Vector store – Pinecone charges per dimension and per million vectors; keep dimensions at 384-768 unless you need higher fidelity.
LLM generation – GPT-4 is pricey; for internal Q&A you can switch to a cheaper model (e.g., Llama-2 7B) behind vLLM and only fall back to GPT-4 for complex queries.

If the budget is tight, consider a hybrid approach: cheap local embeddings + remote vector store, and a cheap open-source LLM with a fallback to OpenAI for edge cases.

FAQ

What is the difference between RAG and plain retrieval-augmented generation?

RAG explicitly includes a generation step after retrieval, whereas plain retrieval just returns the most similar documents. The generation adds synthesis and answer formatting.

Can I use a relational database instead of a vector store?

You can approximate similarity with full-text search, but you lose the semantic matching that embeddings provide. Performance degrades sharply as the corpus grows.

Do I need to chunk documents at all?

Yes. Most LLM contexts are limited to a few thousand tokens. Chunking also improves embedding quality because shorter texts produce more focused vectors.

How often should I re-embed my corpus?

Whenever the source data changes or you switch embedding models. A nightly batch job is a common compromise.

Key Takeaways

A rag pipeline diagram clarifies the flow: ingestion → chunking → embedding → vector store → retrieval → LLM → API.
Build incrementally: start with a minimal FastAPI endpoint, then add async workers, adapters, and persistence.
Visualise the architecture with Mermaid or PlantUML; keep the diagram version-controlled alongside code.
Use pluggable adapters for vector stores and LLMs to avoid lock-in and simplify testing.
Scale each stage independently, monitor latency and error counters, and set budget-aware limits on embedding and generation calls.

By treating the diagram as a living contract, you’ll catch performance regressions early, keep costs predictable, and ship a RAG service that survives real-world traffic.

Building an ai agent python langchain with FastAPI

Ayush Kumar — Fri, 07 Aug 2026 10:39:13 +0000

What is a LangChain agent and how does its architecture work?

A LangChain agent is a wrapper around a language model that lets you define a loop of thought → action → observation.

In practice the agent receives a user prompt, asks the model to decide what to do next, executes the chosen tool (like a web search or a database query), feeds the result back to the model, and repeats until it produces a final answer.

The core pieces are:

Piece	Role
LLM	Generates the next step in natural language.
Prompt template	Shapes the model’s reasoning style.
Tools	Callable Python functions exposed to the agent.
Memory	Stores prior interactions for context.
Agent executor	Orchestrates the loop, handling retries and stop conditions.

In production I’ve seen agents choke when the LLM hallucinates a tool that doesn’t exist, or when the tool’s latency spikes. Guarding the executor with timeouts and a whitelist of allowed actions saves a lot of headaches.

How do I set up a Python environment and install LangChain?

First thing: isolate the stack. I use uvicorn for ASGI, python-3.11, and poetry for dependency management. The minimal pyproject.toml looks like this:

[tool.poetry]
name = "langchain-fastapi-agent"
version = "0.1.0"
description = "A FastAPI service exposing a LangChain AI agent"
authors = ["Your Name <you@example.com>"]
license = "MIT"
python = "^3.11"

[tool.poetry.dependencies]
fastapi = "^0.110.0"
uvicorn = {extras = ["standard"], version = "^0.27.0"}
langchain = "^0.1.0"
openai = "^1.3.0"
pydantic = "^2.5.0"
python-dotenv = "^1.0.0"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Run:

poetry install
poetry shell

Set your OpenAI key in a .env file:

OPENAI_API_KEY=sk-...

Why poetry? It pins exact versions, which is a lifesaver when a minor LangChain bump changes the tool-calling API. I’ve been bitten by a silent break after a pip install -U langchain that removed BaseTool in favor of Tool.

How can I build a simple AI agent with LangChain?

Let’s start with a “calculator” agent that can add, subtract, multiply, or divide numbers. The code below lives in agent.py.

import os
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, Tool
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

# Load LLM
llm = OpenAI(model="gpt-4o-mini", temperature=0)

# Define a single tool
def calc(expression: str) -> str:
    """Evaluate a basic arithmetic expression and return the result."""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as exc:
        return f"Error: {exc}"

calc_tool = Tool(
    name="Calculator",
    func=calc,
    description="Useful for evaluating simple arithmetic expressions, e.g. '2 + 2' or '12/4'."
)

# Memory keeps the conversation context
memory = ConversationBufferMemory(memory_key="chat_history")

# Prompt template – keep it short; long prompts increase token cost
prompt = PromptTemplate.from_template(
    """You are an assistant that can solve math problems.
    Use the provided tools when needed. Respond only with the final answer."""
)

# Build the agent
agent = initialize_agent(
    tools=[calc_tool],
    llm=llm,
    agent_type="zero-shot-react-description",
    memory=memory,
    prompt=prompt,
    verbose=True,
)

A quick test in the REPL:

>>> agent.run("What is 15 * 7?")
The answer is 105.

The zero-shot-react-description agent type is the most predictable for production because it doesn’t rely on few-shot examples that can drift over time.

Trade-offs

Speed vs. cost – Every loop hit calls the LLM, so a 5-step reasoning chain can be pricey. If you only need deterministic math, a pure Python function is cheaper.
Tool security – eval is dangerous. In the example I sandboxed it by removing __builtins__. In real services you should use a safe expression parser like asteval or numexpr.

How do I integrate the LangChain agent into a FastAPI application?

FastAPI gives us an async endpoint that forwards the user query to the agent. Create main.py:

import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dotenv import load_dotenv
from agent import agent

load_dotenv()  # pulls OPENAI_API_KEY into the environment

app = FastAPI(title="LangChain Agent Service")

class QueryRequest(BaseModel):
    prompt: str

class QueryResponse(BaseModel):
    answer: str

@app.post("/chat", response_model=QueryResponse)
async def chat(request: QueryRequest):
    try:
        # LangChain agents are sync; run it in a thread pool to avoid blocking the event loop
        from asyncio import to_thread
        answer = await to_thread(agent.run, request.prompt)
        return QueryResponse(answer=answer.strip())
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))

Why to_thread?

I tried calling agent.run directly from an async route and the server stalled under load. Offloading to a thread pool keeps the ASGI loop responsive, at the cost of a small thread-creation overhead. For high QPS you can pre-warm a pool with concurrent.futures.ThreadPoolExecutor(max_workers=8) and reuse it.

Run locally:

uvicorn main:app --reload

POST to http://localhost:8000/chat with JSON { "prompt": "What is 23 * 42?" } and you’ll get the computed answer.

Failure modes

Timeouts – If the LLM takes longer than your client’s patience, the request bubbles up as a 500. Wrap the call in asyncio.wait_for with a reasonable deadline (e.g., 10 s).
Model rate limits – OpenAI will return a 429. Cache recent results or implement exponential back-off.

How can I add memory and tool-use capabilities to the agent?

The simple calculator only needed one tool. Real-world agents often need:

Database lookup – fetch user profile or inventory.
External API – call a weather service or a payment gateway.
Long-term memory – persist conversation across sessions.

Adding a database tool

Assume a PostgreSQL table users(id, name, balance). Install asyncpg and add the tool:

import asyncpg
import json

async def get_balance(user_id: int) -> str:
    conn = await asyncpg.connect(os.getenv("DATABASE_URL"))
    row = await conn.fetchrow("SELECT balance FROM users WHERE id = $1", user_id)
    await conn.close()
    if row:
        return f"User {user_id} has a balance of ${row['balance']:.2f}"
    return f"No user found with id {user_id}"

Wrap it for LangChain:

from langchain.tools import BaseTool

class BalanceTool(BaseTool):
    name = "GetBalance"
    description = "Retrieve the account balance for a given user ID."

    async def _run(self, user_id: str) -> str:
        return await get_balance(int(user_id))

Add it to the agent initializer:

balance_tool = BalanceTool()
agent = initialize_agent(
    tools=[calc_tool, balance_tool],
    llm=llm,
    agent_type="zero-shot-react-description",
    memory=memory,
    prompt=prompt,
    verbose=False,
)

Now the agent can decide to call GetBalance when the user asks, “How much money does user 42 have?”

Persistent memory

ConversationBufferMemory lives only in RAM. For multi-instance deployments you need a shared store. I switched to RedisChatMessageHistory:

from langchain.memory import ConversationBufferMemory
from langchain.schema import BaseChatMessageHistory
from langchain.chat_message_histories import RedisChatMessageHistory

redis_history = RedisChatMessageHistory(url="redis://localhost:6379/0", session_id="session_123")
memory = ConversationBufferMemory(chat_memory=redis_history, memory_key="chat_history")

Now any FastAPI replica can read the same conversation history, which is essential when you run behind a load balancer.

When NOT to add memory

If your use-case is stateless (e.g., one-off calculation), persisting history adds latency and cost for no benefit. Also, be mindful of GDPR – storing personal data in Redis requires encryption and proper retention policies.

How should I deploy and scale the agent in production?

Containerization

Create a lightweight Docker image based on python:3.11-slim. Example Dockerfile:

FROM python:3.11-slim

WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install poetry && poetry config virtualenvs.create false && poetry install --no-dev

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Build and push:

docker build -t yourrepo/langchain-agent:latest .
docker push yourrepo/langchain-agent:latest

Orchestration

I run the service on Kubernetes with a Deployment of three replicas behind an Ingress. The key settings:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain-agent
  template:
    metadata:
      labels:
        app: langchain-agent
    spec:
      containers:
        - name: api
          image: yourrepo/langchain-agent:latest
          envFrom:
            - secretRef:
                name: openai-secret
          resources:
            limits:
              cpu: "500m"
              memory: "512Mi"
          ports:
            - containerPort: 8000
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10

Why three replicas?

LangChain calls are I/O bound (network to OpenAI). With three pods you can handle ~30 RPS before the CPU caps out. If you need higher throughput, increase the replica count or move to a larger instance type.

Cost considerations

LLM usage – each token costs. Enable a caching layer (Redis) for identical prompts. In my last project we saved ~20 % by caching repetitive “What is the weather in London?” queries.
Thread pool size – each concurrent request spawns a thread for the sync agent. Over-provisioning leads to context-switch thrashing. Tune max_workers to CPU * 2 as a rule of thumb.

Monitoring and alerts

Prometheus metrics from FastAPI (uvicorn exposes --metrics flag). Track request_latency_seconds, error_total.
OpenAI usage – export logs to a file, ship to Loki, and alert when daily spend exceeds a threshold.
Health endpoint – implement /health that pings Redis and runs a cheap LLM prompt like “Say OK”. This catches both DB and API outages.

When to avoid LangChain in production

Deterministic logic – if the problem can be solved with pure code, skip the LLM. The added latency and cost rarely justify a language-model wrapper.
Strict latency SLAs – even with a fast model, the network round-trip can be 100 ms+; add a buffer if you need sub-50 ms responses.

FAQ

What model should I use for a production LangChain agent?

Start with gpt-4o-mini or claude-3.5-sonnet for a good cost-performance balance. Reserve the larger gpt-4o for tasks that need richer reasoning.

Can I run LangChain locally without OpenAI?

Yes. LangChain supports Ollama, Llama.cpp, and other self-hosted backends. The API surface is the same, but you’ll need to manage GPU memory and model updates yourself.

How do I prevent the agent from calling a malicious tool?

Define a static whitelist of Tool objects and let the executor raise an error if the model proposes an unknown name. Also, keep tool functions pure and sandbox any external calls.

Is async support coming to LangChain agents?

Async tool execution is already possible via await tool.arun(...), but the high-level initialize_agent still runs synchronously. Until the library fully embraces async, wrap calls in asyncio.to_thread as shown above.

Key Takeaways

LangChain agents follow a reasoning loop; understand the loop to debug hallucinations.
Isolate dependencies with Poetry; pin versions to avoid silent breaking changes.
Wrap sync agents in asyncio.to_thread when exposing them via FastAPI.
Add memory only when you truly need cross-request context; Redis works well for distributed setups.
Guard tools with a whitelist and sandbox any external execution.
Containerize and orchestrate with Kubernetes; three replicas plus a modest thread pool handle typical production loads.
Monitor spend closely; caching identical prompts can cut token costs dramatically.

If you’re interested in a deeper case study, check out my post on How to build AI agent for stock analysis with FastAPI and the comparison of libraries in Comparing AI Agents Python Library Options for Production.

ai generated code platform security checklist

Ayush Kumar — Thu, 06 Aug 2026 05:44:59 +0000

If you’ve built an AI-generated code platform and it’s now handling real user traffic, the first thing you need to lock down is security. In short, secure your platform by threat modeling the generation pipeline, hardening against prompt injection, protecting model credentials, sandboxing every snippet, and wiring up continuous monitoring. Below is a concrete checklist you can apply today.

I’m a backend engineer who runs FastAPI, AI, and data services in production. I’ve been bitten by every one of the failure modes I’m about to describe, and I’ve paid the price in downtime, data leaks, and wasted developer hours. This post is the distilled version of those hard-earned lessons, focused on ai generated code platform security.

How do I threat model an AI-driven code generation platform?

The answer is: start by mapping every data flow that touches the model and the generated code. Threat modeling for an AI-driven platform looks a lot like traditional OWASP ASVS, but you add two extra edges – the prompt that reaches the model, and the code that leaves it.

Identify assets – model weights, API keys, user prompts, generated snippets, logs.
Identify entry points – HTTP endpoints, WebSocket streams, CLI tools, CI pipelines.
Identify attackers – malicious users, compromised CI runners, insider developers, third-party services.
Identify threats – prompt injection, credential leakage, code execution, data exfiltration, licensing violations.

Once you have a diagram, rank each threat by impact and likelihood. The highest-risk items for most builders are prompt injection and unsafe code execution. Treat those as your first mitigation targets.

Quick threat-model template (YAML)

assets:
  - model_weights: "private S3 bucket"
  - api_keys: "Vault secret"
  - user_prompt: "HTTP POST /generate"
  - generated_code: "temp file in /tmp/snippets"
entry_points:
  - "/generate": "FastAPI endpoint"
  - "ci_job": "GitHub Actions step"
attackers:
  - external_user
  - compromised_ci
threats:
  - prompt_injection:
      impact: high
      likelihood: medium
  - unsafe_execution:
      impact: high
      likelihood: high

Export this to your security backlog and treat each entry as a ticket.

What can I do to prevent malicious prompt injection and unsafe code execution?

The short answer: validate, sanitize, and isolate. Prompt injection is when a user sneaks malicious instructions into the prompt that the LLM interprets as code to write. Unsafe execution is when that code runs on your server without checks.

1. Prompt sanitization

Never feed raw user input straight to the model. Strip out language that could be interpreted as instructions. A simple whitelist approach works for most code-generation use-cases:

import re

SAFE_TOKENS = re.compile(r'^[\w\s\(\)\{\}\[\];,.\'\"-]+$')

def clean_prompt(user_prompt: str) -> str:
    if not SAFE_TOKENS.match(user_prompt):
        raise ValueError("Prompt contains unsafe characters")
    # Remove common instruction keywords
    for bad in ["run", "execute", "import os", "system", "subprocess"]:
        user_prompt = user_prompt.replace(bad, "")
    return user_prompt.strip()

2. Output validation

Treat the model’s output as untrusted. Run it through a static analyser before you ever compile or exec it. For Python, bandit or flake8 with security plugins catches many dangerous patterns.

bandit -r /tmp/snippets/generated.py

If the scanner flags anything, reject the snippet and log the incident.

3. Runtime sandbox

Even with static checks, you need a runtime barrier. Docker is a cheap, battle-tested sandbox for FastAPI services.

FROM python:3.11-slim
RUN pip install fastapi uvicorn
COPY entrypoint.sh /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]

entrypoint.sh launches the snippet inside a container with:

#!/bin/sh
set -e
# Drop privileges
useradd -m sandbox
gosu sandbox python /tmp/snippets/generated.py

The container runs with limited CPU, memory, and no network. If the snippet tries to open a socket, it fails fast.

How do I secure API keys and model access credentials?

A common failure is storing the OpenAI or Claude API key in source code or environment files that get checked into Git. I once pushed a repo with a live key and watched the usage spike to $2,000 overnight.

Best practices

Practice	Why it matters
Use a secret manager (AWS Secrets Manager, HashiCorp Vault)	Centralized audit trail, automatic rotation
Load keys at runtime, never commit them	Eliminates accidental exposure
Scope keys to specific models and endpoints	Limits blast radius if a key leaks
Rotate keys regularly	Reduces window of attack

In FastAPI, loading from Vault looks like:

import hvac
import os

def get_openai_key() -> str:
    client = hvac.Client(url=os.getenv("VAULT_ADDR"))
    client.token = os.getenv("VAULT_TOKEN")
    secret = client.secrets.kv.v2.read_secret_version(path='ai/openai')
    return secret['data']['data']['api_key']

Now the key lives only in memory, never on disk.

What does sandboxing and automated code review of AI-generated snippets look like in practice?

You already saw a Docker sandbox. The next layer is an automated review pipeline that runs on every generated piece before it reaches production. Think of it as a CI step for AI output.

Pipeline sketch (GitHub Actions)

name: Review AI Generated Code
on:
  workflow_dispatch:
    inputs:
      prompt:
        description: 'User prompt'
        required: true
jobs:
  generate-and-review:
    runs-on: ubuntu-latest
    steps:
      - name: Get API key from Vault
        id: vault
        uses: hashicorp/vault-action@v2
        with:
          url: ${{ secrets.VAULT_ADDR }}
          token: ${{ secrets.VAULT_TOKEN }}
          secrets: |
            secret/data/ai/openai key | OPENAI_KEY
      - name: Generate code
        id: gen
        run: |
          python scripts/generate.py "${{ github.event.inputs.prompt }}" > generated.py
      - name: Static analysis
        run: |
          pip install bandit
          bandit -r generated.py || exit 1
      - name: Run in sandbox
        run: |
          docker build -t snippet-sandbox .
          docker run --rm -v $(pwd)/generated.py:/tmp/snippet.py snippet-sandbox

If any step fails, the workflow aborts and you get a Slack notification. The whole loop takes under a minute, keeping latency acceptable for most SaaS products.

For deeper insight, see my earlier post on Fixing AI Generated Code Quality Issues in Production.

How do I handle compliance, licensing, and intellectual property for AI-generated code?

When you ask a model to write code, the output can be a mash-up of publicly available snippets, some of which are under GPL, MIT, or even proprietary licenses. Ignoring this can land you in legal hot water.

Steps to stay compliant

Record the prompt and model version – creates an audit trail.
Run a license detector – tools like scancode-toolkit can scan the generated file for known license headers.
Add a provenance header – prepend every snippet with a comment that documents the generation source.

# Generated by MyAI Platform v1.2.0
# Prompt: "Create a FastAPI endpoint that validates a JWT"
# Model: Claude-2.1

Maintain a denylist of disallowed licenses – reject any snippet that contains GPL text if you can’t ship GPL code.

If you need a deeper dive, read my guide on AI Generated Code Detection: Practical Guide for Builders.

How can I set up continuous monitoring, logging, and alerting for AI-generated code risks?

You can’t rely on a one-time review; threats evolve. The most reliable approach is to treat every generated snippet as a first-class asset that emits telemetry.

What to log

Event	Example
Prompt received	`{"user_id":123,"prompt":"..."}`
Generation result hash	`sha256:abcd1234`
Static analysis outcome	`{"file":"generated.py","issues":2}`
Sandbox exit code	`{"container_id":"xyz","status":"failed","reason":"network denied"}`
Credential usage	`{"key_id":"openai-prod","calls":42}`

FastAPI middleware makes this easy:

from fastapi import FastAPI, Request
import hashlib, json, logging

app = FastAPI()
logger = logging.getLogger("ai_security")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    body = await request.body()
    prompt = json.loads(body).get("prompt", "")
    resp = await call_next(request)
    log_entry = {
        "user": request.headers.get("X-User-ID"),
        "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest(),
        "status": resp.status_code,
    }
    logger.info(json.dumps(log_entry))
    return resp

Alerting

Push those logs to a SIEM (e.g., Elastic, Splunk) and create alerts on:

Repeated prompt injection failures – >5 in 10 minutes.
Sandbox crashes – any non-zero exit.
Credential spikes – >2× normal call volume.

When an alert fires, automatically quarantine the offending user and rotate the affected API key. Automation cuts mean-time-to-response from hours to minutes.

When should I NOT use an AI-generated code snippet in production?

If the snippet:

Requires privileged system calls (e.g., os.system, subprocess.Popen).
Touches sensitive data stores without explicit validation.
Comes from a model you cannot audit for bias or backdoors.

In those cases, treat the model’s output as inspiration only. Write the critical path yourself, or have a senior engineer manually vet the code before merge.

Need a hand implementing these controls?

I’ve helped teams stitch together the exact pipeline described above, from Vault integration to Docker sandbox orchestration. If you’re stuck on a specific failure or just want a sanity check, you can reach out through the hire page. I keep the tone conversational and focus on delivering a working solution, not a sales pitch.

FAQ

What is prompt injection?

Prompt injection is when a user embeds malicious instructions into the text sent to the LLM, causing it to generate dangerous code or disclose secrets. Sanitizing the prompt and limiting allowed tokens mitigates it.

Can I rely on the model’s own safety filters?

No. Model filters are useful but not foolproof. Treat every output as untrusted and run it through your own static analysis and sandbox.

Do I need to rotate API keys every week?

Weekly rotation is overkill for most startups. Rotate keys after any suspected leak, and use short-lived tokens for CI jobs where possible.

Is Docker the only way to sandbox generated code?

Docker is the simplest and most portable. Alternatives include Firecracker microVMs, gVisor, or language-specific sandboxes like pypy-sandbox. Choose based on your latency budget and compliance needs.

Key Takeaways

Map the data flow – threat model every prompt and snippet.
Sanitize and validate – treat prompts and outputs as untrusted.
Sandbox at runtime – Docker containers with no network and limited resources.
Protect credentials – store model keys in a secret manager, never in source.
Automate review – static analysis and CI pipelines catch most issues before they ship.
Track licensing – run a license detector on every generated file.
Monitor continuously – log prompts, hashes, sandbox results, and set up alerts.
Know when to reject – privileged or data-sensitive code should be hand-reviewed.

Securing an AI-generated code platform isn’t a one-off checklist; it’s an ongoing discipline. Apply the steps above, iterate fast, and keep the feedback loop tight. Your users, your data, and your peace of mind will thank you.

Choosing a python message queue library for production

Ayush Kumar — Thu, 06 Aug 2026 05:44:34 +0000

I’ve been asked a lot lately which python message queue library to spin up for a new service. The short answer: pick the one that matches your processing model, latency tolerance, and ops budget. If you need a full-featured task scheduler with retries, go with Celery. If you want something lightweight that plays nicely with asyncio, Dramatiq or a hand-rolled asyncio-Redis queue may be a better fit. Below I walk through the core concepts, compare the most popular libraries, and show how to build a minimal queue with asyncio and Redis. I also share the hard-earned lessons from running these stacks in production.

What are the core concepts behind a python message queue?

A message queue decouples producers (the code that creates work) from consumers (the workers that execute it). The typical flow is:

Publish – a producer serialises a payload and pushes it onto a broker (Redis, RabbitMQ, etc.).
Store – the broker persists the message until a consumer claims it.
Consume – a worker fetches the message, deserialises it, runs the task, and acknowledges success or failure.
Retry / Dead-letter – on failure the broker can re-queue the message or move it to a dead-letter queue.

In Python the heavy lifting is usually done by a library that abstracts the broker API, handles serialization, and provides retry/back-off logic. The library you choose also dictates how you write your worker code: synchronous functions, coroutine-based tasks, or even class-based workers.

Which python message queue library should I use? A quick comparison

Library	Primary broker support	Async support	Built-in retries	Scheduler	Ecosystem	Typical use-case
Celery	RabbitMQ, Redis, SQS, others	Limited (via `celery[redis]` async tasks)	Yes, exponential back-off	Yes (beat)	Huge, many extensions	Complex pipelines, periodic jobs
RQ	Redis only	No (sync only)	Simple retries	No	Small but active	Simple background jobs
Dramatiq	RabbitMQ, Redis, Kafka	Full async (asyncio, trio)	Yes, configurable	No built-in scheduler	Growing	Low-latency, async-first services
Kombu	RabbitMQ, Redis, SQS, others	Sync only (acts as transport layer)	No (you build it)	No	Underpins Celery	When you need a thin transport wrapper
APScheduler	In-process, Redis, Mongo	Sync/async via executors	Yes (misfire handling)	Yes	Moderate	Cron-like jobs inside a service

Why I sometimes avoid Celery

Celery is powerful, but the default configuration spawns a separate pool of processes, each with its own Python interpreter. In a containerised FastAPI service that already runs multiple workers, that extra process tree adds memory overhead (≈150 MiB per worker) and makes graceful shutdown fiddly. I’ve been bitten by “worker-shutdown hangs because a task never responded to SIGTERM”. If you can live without beat, consider Dramatiq or a custom asyncio queue.

Why RQ still has a place

RQ’s simplicity is refreshing. A single Redis list is enough, and the Python API is just a decorator. However, it blocks the event loop, so you can’t use it from an async FastAPI endpoint without off-loading to a thread pool. In a low-traffic internal tool I still run RQ because the ops team loves the single-process model.

Dramatiq’s sweet spot

Dramatiq ships with an asyncio worker out of the box, and its middleware system makes adding custom retry policies painless. I’ve used it to process image thumbnails on a 3-node Redis cluster with sub-second latency. The only downside is that the community is smaller than Celery’s, so you’ll sometimes have to write your own monitoring exporter.

How can I build a simple message queue with asyncio and Redis?

If you’re comfortable with asyncio and already have Redis in your stack, a few hundred lines of code give you a production-ready queue. Below is a minimal implementation that supports publishing, consuming, retries, and graceful shutdown.

# async_queue.py
import asyncio
import json
import os
import signal
import uuid
from dataclasses import dataclass
from typing import Any, Callable, Coroutine

import aioredis

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
QUEUE_NAME = "tasks"
RETRY_LIMIT = 3
RETRY_DELAY = 5  # seconds


@dataclass
class TaskMessage:
    id: str
    payload: Any
    attempts: int = 0

    def to_json(self) -> str:
        return json.dumps({"id": self.id, "payload": self.payload, "attempts": self.attempts})

    @staticmethod
    def from_json(raw: str) -> "TaskMessage":
        data = json.loads(raw)
        return TaskMessage(id=data["id"], payload=data["payload"], attempts=data["attempts"])


class AsyncQueue:
    def __init__(self, redis: aioredis.Redis):
        self.redis = redis
        self._stopped = asyncio.Event()

    async def publish(self, payload: Any) -> str:
        msg = TaskMessage(id=str(uuid.uuid4()), payload=payload)
        await self.redis.rpush(QUEUE_NAME, msg.to_json())
        return msg.id

    async def _process_message(self, raw: str, handler: Callable[[Any], Coroutine[Any, Any, None]]):
        msg = TaskMessage.from_json(raw)
        try:
            await handler(msg.payload)
        except Exception as exc:
            msg.attempts += 1
            if msg.attempts > RETRY_LIMIT:
                await self.redis.rpush(f"{QUEUE_NAME}:dead", msg.to_json())
                print(f"Task {msg.id} moved to dead‑letter after {msg.attempts} attempts")
            else:
                await asyncio.sleep(RETRY_DELAY)
                await self.redis.rpush(QUEUE_NAME, msg.to_json())
                print(f"Retrying task {msg.id}, attempt {msg.attempts}")
        else:
            print(f"Task {msg.id} completed successfully")

    async def worker(self, handler: Callable[[Any], Coroutine[Any, Any, None]]):
        while not self._stopped.is_set():
            # BLPOP blocks until a message arrives or timeout (5 sec)
            result = await self.redis.blpop(QUEUE_NAME, timeout=5)
            if result:
                _, raw = result
                await self._process_message(raw.decode(), handler)

    async def stop(self):
        self._stopped.set()


# Example usage
async def example_handler(payload: dict):
    # Simulate work
    await asyncio.sleep(0.2)
    if payload.get("fail"):
        raise RuntimeError("forced failure")
    print(f"Handled payload: {payload}")

async def main():
    redis = await aioredis.from_url(REDIS_URL, decode_responses=False)
    queue = AsyncQueue(redis)

    # Start a pool of 4 workers
    workers = [asyncio.create_task(queue.worker(example_handler)) for _ in range(4)]

    # Publish some demo tasks
    await queue.publish({"msg": "hello"})
    await queue.publish({"msg": "world", "fail": True})

    # Run for 15 seconds then shut down
    await asyncio.sleep(15)
    await queue.stop()
    await asyncio.gather(*workers)

if __name__ == "__main__":
    asyncio.run(main())

What this code does

publish pushes a JSON-encoded task onto a Redis list.
Workers use BLPOP to block until a task appears, keeping CPU usage low.
On exception the task is re-queued up to RETRY_LIMIT times, then moved to a dead-letter list.
Graceful shutdown is handled via an asyncio.Event that the workers check each loop.

Production-grade tweaks

Use redis-streams (XADD/XREADGROUP) for at-least-once guarantees.
Add Prometheus counters around publish, success, retry, and dead events.
Run the workers as separate containers behind a process manager (systemd, supervisord) to isolate crashes.
Enable TLS and ACLs on Redis; never bind to 0.0.0.0 in prod.

If you already have a FastAPI app, you can expose a /tasks endpoint that calls queue.publish and returns the task ID. The queue code lives in its own module, keeping the API layer thin.

How do I choose the right python message queue library for my use case?

Do you need a scheduler?

Celery’s built-in beat and periodic tasks make it the default for cron-like jobs. If you only need a few ad-hoc timers, a simple asyncio loop with asyncio.sleep or an external scheduler (e.g., Airflow) is cheaper.
Are you already using an async framework?

If your service is async-first (FastAPI, Quart), pick a library that speaks asyncio natively – Dramatiq or a custom queue as shown above. Mixing sync workers (Celery/RQ) forces you to spawn extra threads or processes, which adds latency and memory pressure.
What broker do you have in production?

RabbitMQ shines for high-throughput, durable workloads but requires a separate service and more ops effort. Redis is already in most stacks; it’s fast, but its persistence model is weaker (snapshotting, AOF). If you already run Redis for caching, start there.
How much traffic are you expecting?

For < 100 tasks / second, a single-process RQ or a hand-rolled asyncio queue is fine. Past that threshold, Celery’s pre-fetch and concurrency controls become valuable.
Do you need fine-grained monitoring?

Celery ships with Flower and a rich set of metrics. Dramatiq has an optional Prometheus exporter. With a custom queue you’ll have to instrument yourself (see the Prometheus example in the code).
Team familiarity and community support

Celery has been around since 2011; you’ll find StackOverflow answers for almost any error. Dramatiq’s docs are solid, but the community is smaller. If you’re hiring, expect more candidates to know Celery.

My rule of thumb – start with the simplest thing that works. I often begin with a tiny asyncio-Redis queue for a new microservice. If the load grows or the feature set becomes too limited, I migrate to Dramatiq or Celery. Migration is easier when you keep the payload format stable (JSON) and centralise the broker configuration.

What are the best practices for deploying, scaling, and monitoring a python message queue?

Containerise the worker – Build a Docker image that contains only the worker code and its runtime dependencies. Keep the image lean (use python:3.12-slim). Separate the API container from the worker container; they can scale independently.
Use health checks – Expose a /health endpoint in the worker that verifies Redis connectivity and that the event loop is alive. Kubernetes will restart a flaky pod automatically.
Limit prefetch – In Celery set worker_prefetch_multiplier=1 to avoid pulling too many tasks into memory. In Dramatiq use max_tasks on the worker. Over-prefetching is a common cause of memory spikes.
Back-pressure – If producers outrun consumers, Redis lists can grow unbounded. Guard against this by checking the queue length (LLEN) before publishing, or by using Redis streams with a max-len cap (XADD MAXLEN ~ 10000).
Observability – Export the following metrics:
- tasks_published_total
- tasks_success_total
- tasks_failed_total
- tasks_retried_total
- queue_length
- worker_process_cpu_seconds_total
- worker_process_memory_bytes Grafana dashboards can be built from these. I once missed a spike in queue_length because I only looked at CPU; the queue grew to 2 million items before the alert fired.
Graceful shutdown – Always listen for SIGTERM and give workers a chance to finish in-flight tasks. In Celery you can set worker_shutdown_timeout. In the custom asyncio example above, queue.stop() signals workers to exit after the current task.
Retry policies – Exponential back-off reduces load on downstream services. Celery’s retry_backoff=True and Dramatiq’s Retry middleware make this trivial. Never use a fixed 1-second retry in a high-traffic system; you’ll hammer the same failing endpoint.
Security – Use Redis ACLs to give the worker only LPUSH, BLPOP, and XADD permissions. Rotate the password regularly and store it in a secret manager (e.g., GCP Secret Manager). The same applies to RabbitMQ users.
Version pinning – Message format changes break workers silently. Keep the library version in requirements.txt and bump it only after a coordinated rollout. I once upgraded Celery from 5.2 to 5.3 and the default serializer switched from pickle to json, causing a cascade of deserialization errors.
Testing – Write integration tests that spin up a Redis container (docker run -p 6379:6379 redis:7) and verify that a task is retried the expected number of times. CI pipelines can use the same Docker Compose setup; see my post on Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions for a sample workflow.

FAQ

What is the difference between a message queue and a task queue?

A message queue is a generic conduit for any payload, often used for event streaming. A task queue adds semantics around execution (retries, result storage) and usually ties the message to a callable function.

Can I mix Celery and Dramatiq in the same project?

Technically yes, but it adds operational complexity. Different workers will talk to different brokers, and you’ll need separate monitoring pipelines. I recommend picking one library per service boundary.

Is Redis reliable enough for a production message queue?

Redis provides durability via RDB snapshots and AOF, but it’s not a fully transactional broker like RabbitMQ. For most web-scale workloads it’s fine, especially when you enable appendonly yes and run a replicated cluster.

How do I handle task result storage?

Celery has a built-in result backend (Redis, SQLAlchemy, etc.). Dramatiq leaves result handling to you; you can push the result onto another Redis key or a database table. For fire-and-forget jobs you can skip result storage entirely.

Key Takeaways

The python message queue library you choose should align with your async model, broker availability, and scaling needs.
Celery is the heavyweight champion; use it for complex pipelines and periodic jobs.
Dramatiq offers native asyncio support with a small footprint - great for low-latency services.
RQ shines when you only need a simple Redis list and want minimal ops overhead.
A custom asyncio-Redis queue can be production-ready if you add retries, dead-letter handling, and proper monitoring.
Deploy workers in their own containers, limit prefetch, watch queue length, and instrument with Prometheus.
Always test retry behavior and plan for graceful shutdown; a hanging worker is a silent outage.

Happy queuing!

Redis caching in FastAPI: practical setup and pitfalls

Ayush Kumar — Mon, 03 Aug 2026 07:47:29 +0000

Quick answer:

If you want fast, cheap look-ups for query results, session data, or computed values, add a Redis instance, connect with aioredis, wrap your endpoints in a small decorator, and let the cache expire or be cleared on writes. The whole stack fits in a few dozen lines and costs pennies per month on most cloud providers.

Below I walk through a production-ready implementation of redis caching in fastapi. I’ll show you how to install Redis, wire an async client, build reusable decorators, handle invalidation, spin the service up with Docker, and keep an eye on performance once you’re in the wild. I’ll also point out the places where things tend to break and what I would do differently next time.

Installing and configuring Redis for a FastAPI project

Do I need a separate Redis server?

Yes. Even a single-node instance gives you in-memory speed without adding latency to your API process. For local development you can run the official Docker image; for production you’ll want a managed service or a replicated cluster.

Local Docker setup

# docker-compose.yml
version: "3.9"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: ["redis-server", "--save", "60", "1", "--loglevel", "warning"]
    restart: unless-stopped

Run docker-compose up -d redis. The container is tiny (≈30 MB) and starts in a second. On a VM or Kubernetes you would expose the service via a ClusterIP or a cloud-managed endpoint.

Production connection string

# .env
REDIS_URL=redis://:myStrongPassword@redis-prod.internal:6379/0

Never hard-code credentials. Keep the URL in a secret manager and inject it at runtime. I’ve been bitten by missing redis:// prefixes when moving from localhost to a managed instance - FastAPI will raise a ValueError at startup.

How do I integrate an async Redis client with aioredis?

Can I use the same client for all requests?

Yes, but you should create a single connection pool at startup and share it via FastAPI’s dependency system. aioredis (now part of redis-py v4) gives you native asyncio support.

# app/redis.py
import os
from redis.asyncio import Redis
from fastapi import FastAPI

def get_redis_pool() -> Redis:
    return Redis.from_url(os.getenv("REDIS_URL"), decode_responses=True)

def add_redis(app: FastAPI) -> None:
    @app.on_event("startup")
    async def startup_redis():
        app.state.redis = get_redis_pool()
        # Test the connection early
        await app.state.redis.ping()

    @app.on_event("shutdown")
    async def shutdown_redis():
        await app.state.redis.close()

In any route you can now do:

from fastapi import Depends, Request

async def redis_dep(request: Request) -> Redis:
    return request.app.state.redis

Trade-offs

Single pool – reduces connection overhead but can become a bottleneck if you saturate the event loop with heavy Redis work. In that case, spawn a second pool for low-latency reads.
Blocking commands – avoid commands that return huge payloads; they block the event loop while the client decompresses data. Use SCAN instead of KEYS in production.

How can I build reusable cache decorators for FastAPI endpoints?

Is a decorator enough for all use cases?

For read-only or idempotent endpoints, a decorator is clean and keeps the view logic tidy. For write-heavy resources you’ll need explicit invalidation.

# app/cache.py
import functools
import json
from typing import Callable, Any
from redis.asyncio import Redis

def redis_cache(key_builder: Callable[..., str],
                ttl: int = 300):
    """Cache the result of an async endpoint.
    key_builder receives the same args/kwargs as the endpoint.
    """
    def decorator(func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            request = kwargs.get("request")
            redis: Redis = request.app.state.redis  # type: ignore
            cache_key = key_builder(*args, **kwargs)
            cached = await redis.get(cache_key)
            if cached is not None:
                return json.loads(cached)

            result = await func(*args, **kwargs)
            await redis.set(cache_key, json.dumps(result), ex=ttl)
            return result
        return wrapper
    return decorator

Example usage

# app/routers/items.py
from fastapi import APIRouter, Request
from .cache import redis_cache

router = APIRouter()

def item_key(request: Request, item_id: int):
    return f"item:{item_id}"

@router.get("/items/{item_id}")
@redis_cache(key_builder=item_key, ttl=600)
async def get_item(request: Request, item_id: int):
    # Simulate expensive DB call
    return {"id": item_id, "value": "expensive computation"}

The decorator automatically reuses the pool created earlier, serializes the response with json, and respects the TTL.

When not to use it

Streaming responses – you cannot cache a generator without buffering the whole payload.
Very large payloads – they waste memory and network bandwidth. Consider pagination or a separate blob store.

What are the best cache invalidation and expiration strategies?

How do I keep stale data from leaking?

Time-based expiration (TTL) – simplest. Set a reasonable ttl per endpoint. For data that changes every few minutes, a 60-second TTL is enough.
Write-through invalidation – after you write to the database, delete the related keys.

# app/routers/items.py
@router.put("/items/{item_id}")
async def update_item(request: Request, item_id: int, payload: dict):
    # Update DB here …
    await request.app.state.redis.delete(f"item:{item_id}")
    return {"status": "updated"}

Tag-based eviction – store a set of keys per logical group and delete the whole set on a bulk change.

# add tag on set
await redis.sadd(f"tag:user:{user_id}", cache_key)
# later invalidate all
keys = await redis.smembers(f"tag:user:{user_id}")
if keys:
    await redis.delete(*keys)
    await redis.delete(f"tag:user:{user_id}")

Pitfalls

Forgotten invalidation – I once added a new PUT endpoint and missed the delete call. The cache kept returning the old version for hours. Adding a unit test that checks cache freshness saved me a lot of tickets.
Over-eager expiration – setting TTLs too low defeats the purpose of caching and adds load to Redis. Profile your read/write ratio before deciding.

How do I deploy Redis with Docker and handle environment-specific settings?

Can I keep the same compose file for dev and prod?

Yes, but you should separate the image version and resource limits. In production you may want persistence, ACLs, and monitoring side-cars.

Production-ready compose snippet

# docker-compose.prod.yml
version: "3.9"
services:
  redis:
    image: redis:7-alpine
    command:
      - redis-server
      - /usr/local/etc/redis/redis.conf
    volumes:
      - redis-data:/data
      - ./redis.conf:/usr/local/etc/redis/redis.conf:ro
    environment:
      - REDIS_PASSWORD=${REDIS_PASSWORD}
    ports:
      - "6379:6379"
    deploy:
      resources:
        limits:
          memory: 512M
        reservations:
          memory: 256M
    restart: unless-stopped

volumes:
  redis-data:

redis.conf can enable ACLs:

requirepass ${REDIS_PASSWORD}
maxmemory 256mb
maxmemory-policy allkeys-lru

When deploying to Google Cloud Run you cannot run a side-car, so you’ll point the FastAPI container to a managed Memorystore instance instead. The same REDIS_URL env var works for both setups.

What I’d do differently

Separate config repo – keep redis.conf in a dedicated folder, version it, and inject secrets via the orchestrator. I once edited the file inside the container and lost the change after a redeploy.
Health checks – add a simple redis-cli ping health endpoint in your orchestrator. Without it, a pod could start while Redis was still booting, causing a cascade of connection errors.

How can I monitor and troubleshoot Redis cache performance in production?

What metrics matter most?

Hit-rate, latency, memory usage, and evicted keys. Redis ships with INFO and the Redis Exporter for Prometheus.

Prometheus exporter

# docker-compose.yml (add beside redis)
redis-exporter:
  image: oliver006/redis_exporter
  environment:
    - REDIS_ADDR=redis://:${REDIS_PASSWORD}@redis:6379
  ports:
    - "9121:9121"

Grafana dashboards can then show redis_keyspace_hits_total / (hits + misses) as a percentage. A dip below 80 % usually means your TTL is too short or you’re missing a cache-warming step.

Logging slow commands

Add to redis.conf:

slowlog-log-slower-than 10000   # microseconds
slowlog-max-len 128

You can query SLOWLOG GET from your FastAPI admin endpoint to see which queries are eating time. In one project I discovered a ZRANGEBYSCORE on a 10 M-element sorted set that took 250 ms; the fix was to add a secondary index.

Common failure modes

Symptom	Likely cause	Quick fix
`ConnectionError: Timeout`	Redis container restarted or network partition	Check Docker logs, add `restart: always`, verify health probe
Cache always miss	Wrong key format or missing `decode_responses=True`	Ensure the same key builder is used for set and get
Memory limit reached	`maxmemory` too low, or unbounded keys	Enable LRU eviction, monitor `used_memory_peak`

FAQ

How many Redis nodes do I need for a small FastAPI service?

A single node is fine for low-traffic workloads. Add a replica once you exceed ~100 K requests per second or need high availability.

Can I use Redis as a message broker instead of a cache?

Technically yes, but Pub/Sub semantics differ from a durable queue. For background jobs I prefer RabbitMQ or Google Pub/Sub.

Is it safe to store user-session data in Redis without encryption?

If the data is not personally identifiable, plain storage is okay. For anything sensitive, enable TLS on the Redis endpoint and store only a session token that references server-side data.

What’s the cost difference between managed Memorystore and a self-hosted Docker container?

Managed services start at a few dollars per GB per month and include automatic failover. A self-hosted container on a small VM can be cheaper (< $5) but you’re responsible for backups and scaling.

Key Takeaways

Install Redis as a separate service; use Docker locally and a managed instance in prod.
Create a single async Redis pool at FastAPI startup and inject it via request state.
Wrap read-only endpoints in a lightweight decorator that builds keys, checks the cache, and sets a TTL.
Invalidate on writes with explicit delete calls or tag-based sets for bulk eviction.
Deploy with environment-specific redis.conf, enable ACLs, and set sensible memory limits.
Export metrics to Prometheus, watch hit-rate and latency, and enable the slow-log for troubleshooting.

With these pieces in place, redis caching in fastapi becomes a low-maintenance performance boost that scales from a laptop dev box to a multi-region production fleet. Happy caching!

Kafka Python Tutorial: From Docker to FastAPI

Ayush Kumar — Mon, 03 Aug 2026 07:47:20 +0000

If you need a quick way to get Kafka running locally and talk to it from Python, this kafka python tutorial shows you how to spin up a broker with Docker, install the kafka-python client, produce messages, consume them asynchronously, and hook the flow into a FastAPI endpoint for background processing. By the end you’ll have a reproducible dev environment and a pattern you can push to production with minimal surprises.

How do I spin up a local Kafka broker with Docker?

The simplest way to start experimenting is to run a single-node Kafka cluster in Docker. The official confluentinc/cp-kafka image includes Zookeeper, so you don’t have to manage a separate service.

docker network create kafka-net

docker run -d \
  --name zookeeper \
  --network kafka-net \
  -p 2181:2181 \
  -e ZOOKEEPER_CLIENT_PORT=2181 \
  confluentinc/cp-zookeeper:7.5.0

docker run -d \
  --name kafka \
  --network kafka-net \
  -p 9092:9092 \
  -e KAFKA_BROKER_ID=1 \
  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  confluentinc/cp-kafka:7.5.0

A few things bite me early:

Port mapping – PLAINTEXT://localhost:9092 works for local dev but will need a proper DNS name in production.
Resource limits – the default container uses ~1 GiB RAM. On a low-spec laptop you’ll see OOM kills unless you set -e KAFKA_HEAP_OPTS="-Xmx512M -Xms512M".
Data persistence – without a volume, every docker rm wipes your topic data. Add -v kafka-data:/var/lib/kafka/data if you need to keep it across restarts.

Verify the broker is alive:

docker exec kafka kafka-topics --bootstrap-server localhost:9092 --list

You should see an empty list, which means the broker is reachable.

How to install and configure the kafka-python client?

For most quick prototypes kafka-python is enough. It’s pure Python, has a small footprint, and works well with FastAPI’s async model when you run the consumer in a background thread.

pip install kafka-python

Configuration is just a dict you pass to the KafkaProducer or KafkaConsumer. In production you’ll pull these values from environment variables or a config service.

KAFKA_CFG = {
    "bootstrap_servers": "localhost:9092",
    "security_protocol": "PLAINTEXT",   # switch to SASL_SSL in prod
    "client_id": "logicloop-producer",
}

A common mistake is to forget the acks setting on the producer. The default is acks=1, which can lose data if the leader crashes. For a safe dev run set:

producer = KafkaProducer(**KAFKA_CFG, acks="all", retries=5)

How to produce messages from a Python application?

Producing is straightforward. Serialize your payload to JSON (or Avro, if you have a schema registry) and send it to a topic.

import json
from kafka import KafkaProducer

producer = KafkaProducer(
    **KAFKA_CFG,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

def send_user_event(user_id: int, action: str):
    payload = {"user_id": user_id, "action": action, "ts": time.time()}
    future = producer.send("user-events", value=payload)
    try:
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic}:{record_metadata.partition}")
    except Exception as exc:
        print(f"Send failed: {exc}")

# Example usage
send_user_event(42, "login")
producer.flush()

What bites me in production is message ordering. Kafka only guarantees order within a partition. If you need strict ordering per user_id, hash the key to a single partition:

producer = KafkaProducer(
    **KAFKA_CFG,
    key_serializer=str.encode,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

producer.send("user-events", key=str(user_id), value=payload)

Now all events for the same user_id land in the same partition and preserve order.

How to consume messages in Python with async support?

kafka-python does not expose an async API, but you can run the consumer inside an asyncio task that delegates to a thread pool. The pattern works fine for FastAPI background workers.

import asyncio
import json
from kafka import KafkaConsumer
from concurrent.futures import ThreadPoolExecutor

consumer = KafkaConsumer(
    "user-events",
    **KAFKA_CFG,
    group_id="logicloop-group",
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    auto_offset_reset="earliest",
)

executor = ThreadPoolExecutor(max_workers=2)

async def consume_loop():
    loop = asyncio.get_running_loop()
    while True:
        # poll blocks, so we run it in a thread
        msgs = await loop.run_in_executor(executor, consumer.poll, 1.0)
        for tp, records in msgs.items():
            for record in records:
                await handle_event(record.value)

async def handle_event(event: dict):
    # Simulate async DB write
    await asyncio.sleep(0.01)
    print(f"Processed {event}")

# Run the loop in a separate thread when the app starts
def start_consumer_background():
    asyncio.run(consume_loop())

If you prefer a truly async client, swap kafka-python for aiokafka. The API is almost identical, but you must install the extra package and adjust the startup code. I keep the kafka-python version for simplicity because it plays nicely with the existing codebase and the extra thread overhead is negligible for low-throughput services.

When I first tried to run the consumer directly in the FastAPI event loop, the app hung on startup. The fix was to launch the consumer in a separate daemon thread (or use lifespan events). See the FastAPI async pitfalls in the internal link Fixing SQLAlchemy MissingGreenlet Error in FastAPI (Async Explained) for more context on mixing sync and async code.

How can I integrate Kafka with FastAPI for background processing?

FastAPI gives you two natural hooks: startup events to launch the consumer and background tasks to fire-and-forget work that originates from HTTP requests.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.on_event("startup")
async def start_kafka_consumer():
    loop = asyncio.get_event_loop()
    loop.create_task(consume_loop())   # consume_loop from previous section

@app.post("/events/")
async def publish_event(user_id: int, action: str, background: BackgroundTasks):
    send_user_event(user_id, action)   # sync producer, cheap call
    # Optionally schedule extra processing
    background.add_task(log_audit, user_id, action)
    return {"status": "queued"}

def log_audit(user_id: int, action: str):
    # This runs in a thread pool managed by FastAPI
    time.sleep(0.1)  # simulate I/O
    print(f"Audit log for {user_id}:{action}")

A few production-grade concerns:

Graceful shutdown – FastAPI’s shutdown event should close the consumer and the thread pool to avoid “dangling thread” warnings.
Back-pressure – If the consumer can’t keep up, messages accumulate in the broker. Tune max_poll_records and consider scaling the consumer group horizontally.
Observability – Export consumer lag metrics to Prometheus. The kafka-python client exposes consumer.position(tp) and consumer.end_offsets([tp]) which you can compare.

When I first deployed this pattern to Google Cloud Run, the container was killed after 15 minutes of idle time, causing the consumer to miss messages. The fix was to add a tiny “keep-alive” endpoint that Cloud Run pings, and to set the --timeout flag higher. The full deployment story is covered in Serverless Python: Deploying FastAPI to Google Cloud Run with Docker.

Common pitfalls and best-practice tips

Pitfall	Why it hurts	Remedy
Using the default `acks=1` in production	Possible data loss on leader failure	Set `acks="all"` and enable idempotent producer (`enable_idempotence=True`)
Forgetting to commit offsets manually	Consumer may reprocess the same batch after restart	Use `consumer.commit()` after successful handling, or enable `enable_auto_commit=False` and manage commits yourself
Running the consumer in the same thread as FastAPI	Event loop blocks, endpoint latency spikes	Run the consumer in a separate daemon thread or use `aiokafka`
Not configuring retries on the producer	Transient network blips cause message drop	Set `retries` and `retry_backoff_ms`
Over-committing partitions to a single consumer	Limits scalability	Keep the number of partitions ≥ number of consumer instances
Ignoring broker resource limits	Out-of-memory kills in Docker	Allocate appropriate heap (`KAFKA_HEAP_OPTS`) and monitor container memory

When you move beyond a single node, you’ll need to think about replication factor, multi-region latency, and security (SASL/SCRAM, TLS). Those concerns are out of scope for this tutorial but worth planning early.

FAQ

What’s the difference between kafka-python and aiokafka?

kafka-python is a synchronous client that works well with a thread-based approach. aiokafka is built on asyncio and offers true non-blocking APIs, but adds an extra dependency and a slightly steeper learning curve.

Do I need Zookeeper for a production Kafka cluster?

Modern Kafka versions can run without an external Zookeeper when you enable KRaft mode. The Docker images used in this tutorial still rely on Zookeeper for simplicity, but for new clusters consider KRaft to reduce operational overhead.

How many partitions should I create for a topic?

Start with a number that matches your expected consumer parallelism (e.g., 3-6 partitions for a small service). You can increase partitions later, but you cannot decrease them without recreating the topic.

Can I use this setup on Windows?

Docker Desktop runs Linux containers on Windows, so the broker part works the same. The only Windows-specific issue is the path handling in volume mounts; use absolute Unix-style paths inside the container.

Key Takeaways

Docker lets you spin up a single-node Kafka broker in seconds; remember to persist data if you need it across restarts.
kafka-python is a lightweight client; configure acks, retries, and proper serialization to avoid data loss.
Produce messages with a key if you need ordering per entity.
Consume asynchronously by delegating the blocking poll to a thread pool; for pure async workloads consider aiokafka.
FastAPI startup/shutdown events are the right place to launch and stop a Kafka consumer; background tasks handle fire-and-forget work.
Watch for common pitfalls: offset management, back-pressure, resource limits, and security settings.

With these pieces in place you have a solid foundation for building reliable, event-driven services in Python. The next step is to push the Docker image to your CI/CD pipeline - see the guide on Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions for a production-ready workflow.

Fixing AI Generated Code Quality Issues in Production

Ayush Kumar — Fri, 31 Jul 2026 10:31:32 +0000

When AI-generated code quality issues show up in production, the quickest fix is to stop the broken code from reaching users and put a safety net around the generation step. In short: add tests, lint, and a CI gate, then treat every AI snippet like a third-party library you would audit before import.

I’ve been bitten by this more than once. A single line of code that a tool like Cursor or Claude wrote crashed my FastAPI service under load, and the cost was a missed SLA and a frantic hot-fix sprint. Below I walk through the exact failure patterns I’ve seen, the tooling that catches them early, and the process tweaks that keep the pipeline clean. If you’re at the point where “it works locally but blows up in prod” feels like a mantra, this is for you.

What are the most common AI generated code quality pitfalls?

The first thing to know is that AI-generated code isn’t magically perfect. It often repeats three classic mistakes:

Missing edge-case handling – The snippet assumes happy-path inputs. In my last project, an AI-written validation function accepted any string, forgetting to strip whitespace. A single malformed request caused a ValueError that bubbled up to the FastAPI exception handler and returned a 500.
Incorrect async/sync mix – FastAPI is async-first, yet the model sometimes drops a blocking requests.get into an async endpoint. The result? The event loop stalls, latency spikes, and the autoscaler thinks the service is overloaded.
Undocumented side effects – The generated code may open files, create temp directories, or mutate global state without a comment. In a microservice that runs many concurrent requests, that leads to race conditions and flaky tests.

These pitfalls manifest as ai generated code quality issues that are easy to miss in a quick copy-paste but explode under real traffic.

How can I test AI-generated snippets automatically?

The answer is: treat every snippet as a unit that must pass the same test suite as the rest of your code. Start with a tiny harness that runs the snippet in isolation.

# test_generated_snippet.py
import pytest
from importlib import util
from pathlib import Path

def load_snippet(path: Path):
    spec = util.spec_from_file_location("generated", path)
    module = util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module

@pytest.fixture
def snippet():
    return load_snippet(Path("generated/validation.py"))

def test_input_validation(snippet):
    # Assume the AI created a function called `validate_user_input`
    assert snippet.validate_user_input("valid123") is True
    with pytest.raises(ValueError):
        snippet.validate_user_input("  bad  ")

Run this with pytest -q. If the snippet fails, you know the problem before it reaches CI. The same pattern works for async code - use pytest-asyncio and assert that no blocking calls are made.

Trade-off: Writing a harness for every snippet adds a few minutes of developer time, but it saves hours of debugging in production. If you generate dozens of snippets per week, automate harness creation with a small script that reads a comment header from the AI output (e.g., # test:validation).

Which static analysis and linting tools catch the most issues?

Static analysis is the cheapest guardrail. Here are the tools I rely on for ai generated code quality issues:

Tool	What it catches	Why it matters
ruff	Unused imports, undefined names, line length, async-sync mismatches	Fast, integrates with pre-commit
mypy	Type mismatches, missing return annotations	Forces the AI to be explicit
bandit	Security-oriented patterns (e.g., `eval`, insecure HTTP)	Prevents accidental vulnerabilities
pylint (configured with `disable=missing-docstring`)	Code smells, duplicate code, too complex functions	Highlights refactor opportunities

A minimal .pre-commit-config.yaml that runs these on every commit looks like this:

repos:
  - repo: https://github.com/charliermarsh/ruff-pre-commit
    rev: v0.5.0
    hooks:
      - id: ruff
  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.10.0
    hooks:
      - id: mypy
        additional_dependencies: [types-requests]
  - repo: https://github.com/PyCQA/bandit
    rev: 1.7.9
    hooks:
      - id: bandit
        args: ["-r", "generated/"]

Run pre-commit install and you’ll get instant feedback when an AI snippet violates any rule. In practice, I caught a missing await call that would have blocked the entire FastAPI worker within seconds of the snippet being saved.

When not to use: If you generate code in a language other than Python, swap in the appropriate linters (e.g., eslint for JavaScript). The principle stays the same.

How do I integrate quality gates for AI code into CI/CD?

The answer: add a dedicated “AI-generated” stage to your pipeline that runs the same linters and tests, and make it a required check before merge.

A typical GitHub Actions workflow:

name: CI

on: [push, pull_request]

jobs:
  lint-and-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 deps
        run: pip install -r requirements.txt
      - name: Run linters
        run: |
          ruff .
          mypy .
          bandit -r generated/
      - name: Run tests
        run: pytest -q

If any step fails, the PR cannot be merged. The extra cost is a few minutes of CI time, but it prevents the far more expensive rollback of a production incident.

Integration tip: Tag AI-generated files with a comment header, e.g., # AI_GENERATED. Then configure your linter to only scan those paths, keeping the rest of the repo fast.

What are best practices for reviewing and refactoring AI-generated code?

Even with tests and linters, a human eye is still valuable. Here’s my checklist:

Read the intent comment – Most generators let you prepend a description. Verify that the comment matches the implementation.
Check for hidden state – Look for global variables, file writes, or network calls. Wrap them in a context manager if they’re needed.
Add type hints – If the AI omitted them, add explicit TypedDict or pydantic models. This forces downstream code to respect the contract.
Isolate the snippet – Move it to its own module under generated/ and expose a clean public API. That makes future refactors easier.
Document edge cases – Write a short docstring that lists the inputs the function does not handle. This prevents the same mistake from being re-generated later.

When you refactor, keep the original snippet versioned. I store the raw AI output in a separate raw/ folder; that way I can compare before and after with git diff. It also helps when you need to audit who generated what, a common request from security teams.

If you need a concrete example, see how I turned an AI-written stock-analysis endpoint into a well-typed FastAPI router in the post How to build AI agent for stock analysis with FastAPI. The transformation reduced latency by 30 % and eliminated a hidden requests call that was blocking the event loop.

FAQ

Why does my AI-generated async function still block the event loop?

Because the snippet likely contains a synchronous library call (e.g., requests.get). Replace it with an async client like httpx or run the call in a thread pool with anyio.to_thread.run_sync.

Can I trust linting alone to catch security bugs in AI code?

No. Linters flag risky patterns, but a manual security review or a tool like bandit is still required for deeper analysis.

Do I need to write unit tests for every single AI snippet?

Ideally yes, but a pragmatic approach is to prioritize snippets that touch I/O, authentication, or external APIs. Those are the ones that cause the biggest production pain.

What if the AI keeps generating the same bug after I fix it?

Add the fix to the prompt you feed the model, or use a post-generation script that applies a template patch. Over time the model learns from the corrected examples.

Key Takeaways

Identify the failure: missing edge-case handling, async/sync mismatches, and hidden side effects are the most common AI generated code quality issues.
Test early: a small pytest harness for each snippet catches bugs before they enter CI.
Lint aggressively: ruff, mypy, and bandit together spot the majority of problems.
Gate the pipeline: make linting and testing a required CI stage to stop bad code from merging.
Review thoughtfully: isolate snippets, add type hints, and document edge cases to make future maintenance painless.

If you’ve hit a wall and need someone to step in, I’m taking on a few hands-on contracts right now. Feel free to reach out through the hire page and we can tackle those stubborn AI generated code quality issues together.

Fixing an AI Generated Code Vulnerabilities Report

Ayush Kumar — Fri, 31 Jul 2026 09:29:21 +0000

I’ve been there: you let an LLM write a FastAPI endpoint, the app ships, and a security scan flags a handful of issues you never saw coming. The good news is you don’t have to chase down each problem manually. An ai generated code vulnerabilities report tells you exactly what to fix, and with the right tooling you can make that report part of every pull request. Below I walk through why AI-generated snippets are prone to bugs, how to set up static analysis tools, automate scans in CI/CD, read the report, and apply best-practice mitigations.

Why does AI-generated code often contain security flaws?

The answer is simple: language models predict the next token, not the next safe token. They have seen millions of open-source projects, many of which contain outdated patterns, hard-coded secrets, or insecure defaults. When you ask a model to “write a login route with JWT”, it will often:

Use a static secret key (SECRET = "mysecret").
Skip input validation (username = request.json["user"]).
Trust user-provided data for file paths.

Those patterns compile and run, but they open the door to injection, credential leakage, and privilege escalation. In production I saw a bot-generated CSV import that used os.system on the supplied filename – a classic command injection that slipped through code review because the snippet was only a few lines.

Static analysis tools shine here because they look for the effect of the code, not the intent behind it. They can flag hard-coded secrets, unsafe deserialization, missing CSRF protection, and more. The key is to treat the output of the LLM as any other third-party dependency: run it through a scanner before it touches your codebase.

How do I set up static analysis tools for AI-generated snippets?

I use a mix of Bandit, Semgrep, and SonarQube. Bandit is quick for Python-specific issues, Semgrep lets me write custom rules that target the patterns LLMs love, and SonarQube provides a dashboard for the whole team.

1. Install Bandit

pip install bandit

Create a simple wrapper that runs Bandit only on the generated/ folder where you store AI-produced files:

# run_bandit.sh
#!/usr/bin/env bash
set -e
bandit -r generated/ -lll -f json -o bandit_report.json

The -lll flag treats low-severity findings as failures, which is useful when you want a clean bill of health before merging.

2. Add Semgrep with custom rules

First, install Semgrep:

pip install semgrep

Then create a rule file semgrep_rules.yml that catches common LLM pitfalls:

rules:
  - id: hardcoded-secret
    patterns:
      - pattern: $VAR = "$SECRET"
    message: "Avoid hard‑coded secrets in generated code."
    severity: HIGH
    languages: [python]
  - id: unsafe-os-system
    pattern: os.system(...)
    message: "os.system can lead to command injection."
    severity: HIGH
    languages: [python]

Run it with:

semgrep --config semgrep_rules.yml generated/ --json > semgrep_report.json

3. Hook SonarQube into the mix

If you already have SonarQube for the rest of the repo, just add the generated/ directory to the analysis scope in sonar-project.properties:

sonar.sources=app,generated
sonar.exclusions=tests/**

When you trigger a SonarQube scan, the UI will aggregate findings from both your hand-written code and the AI-generated snippets, giving you a single ai generated code vulnerabilities report.

How can I automate vulnerability scanning in my CI/CD pipeline?

Automation is the only way to keep the feedback loop short. Below is a GitHub Actions workflow that runs Bandit, Semgrep, and SonarQube on every PR. Adjust the steps if you use GitLab or CircleCI – the concepts are the same.

name: Scan AI‑Generated Code

on:
  pull_request:
    paths:
      - 'generated/**'

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      # Set up Python
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      # Install tools
      - name: Install scanners
        run: |
          pip install bandit semgrep
          # SonarScanner CLI
          curl -sSLo sonar-scanner.zip https://binaries.sonarsource.com/Distribution/sonar-scanner-cli/sonar-scanner-cli-5.0.1.3009-linux.zip
          unzip sonar-scanner.zip -d $HOME
          echo "$HOME/sonar-scanner-5.0.1.3009-linux/bin" >> $GITHUB_PATH

      # Run Bandit
      - name: Run Bandit
        run: ./run_bandit.sh

      # Run Semgrep
      - name: Run Semgrep
        run: semgrep --config semgrep_rules.yml generated/ --json > semgrep_report.json

      # Run SonarQube
      - name: SonarQube Scan
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
        run: |
          sonar-scanner \
            -Dsonar.projectKey=my-fastapi-app \
            -Dsonar.sources=app,generated \
            -Dsonar.host.url=https://sonar.mycompany.com \
            -Dsonar.login=$SONAR_TOKEN

      # Upload reports as artifacts
      - name: Upload reports
        uses: actions/upload-artifact@v3
        with:
          name: security-reports
          path: |
            bandit_report.json
            semgrep_report.json

The workflow fails the PR if any scanner produces a HIGH severity finding. That way the ai generated code vulnerabilities report becomes a gate, not an after-the-fact checklist.

How do I interpret and triage the generated vulnerability report?

When the pipeline finishes you’ll have three JSON files. Here’s a quick way to turn them into a readable markdown table that you can paste into the PR comment.

import json
from pathlib import Path

def load(path):
    return json.load(Path(path).open())

bandit = load('bandit_report.json')
semgrep = load('semgrep_report.json')
# SonarQube API can be queried for issues; for brevity we assume a local JSON export
sonar = load('sonarqube_issues.json')

def format_issue(issue, source):
    return f"| {source} | {issue['filename']} | {issue.get('line_number', '-') } | {issue['severity']} | {issue['issue_text']} |"

rows = []
for i in bandit['results']:
    rows.append(format_issue(i, 'Bandit'))
for i in semgrep:
    rows.append(format_issue(i, 'Semgrep'))
for i in sonar['issues']:
    rows.append(format_issue(i, 'SonarQube'))

markdown = "\n".join([
    "## AI Generated Code Vulnerabilities Report",
    "| Tool | File | Line | Severity | Description |",
    "|------|------|------|----------|-------------|",
    *rows
])
Path('vuln_report.md').write_text(markdown)
print("Report written to vuln_report.md")

A few triage tips:

Prioritize HIGH severity – they usually map to credential exposure or remote code execution.
Group by file – if a single generated file has three issues, consider rewriting it manually.
Check for duplicates – Bandit and SonarQube may flag the same hard-coded secret; keep only the highest severity entry.
Add remediation notes – in the PR comment, suggest concrete fixes (e.g., “replace static key with os.getenv('JWT_SECRET')”).

When you see the same pattern across many snippets, it’s a signal to adjust your prompt or post-process step. For example, after spotting repeated os.system usage, I added a prompt clause: “Never use os.system; use subprocess.run with a list of arguments”.

What best practices can I follow to mitigate risks in AI-generated code?

Never merge raw AI output – always run it through a linter and a scanner first.
Store generated code in a dedicated directory – makes scoping easier and keeps human-written code separate.
Enforce secret management – replace any literal that looks like a key or password with an environment variable lookup. Tools like python-dotenv help keep defaults empty.
Add unit tests immediately – a failing test catches logic errors that static analysis may miss. I usually generate a skeleton test with the same LLM, then flesh it out manually.
Version-control the prompts – if a specific prompt leads to insecure code, lock it down or add a comment that warns future users.
Run dependency-checkers – AI may suggest outdated libraries. Use pip-audit or safety to verify that the generated requirements.txt is safe.
Consider a “review bot” – a small FastAPI endpoint that receives generated files, runs the scanners, and returns the JSON report. See my post on AI agent python ollama: Build, Test, Deploy with FastAPI for a pattern you can adapt.

When you combine these practices with the automated pipeline above, the ai generated code vulnerabilities report becomes a routine health check rather than a surprise after a production incident.

FAQ

What if the scanners miss a vulnerability?

Static analysis is not a silver bullet. Complement it with runtime monitoring (e.g., Snyk IaC scans, OWASP ZAP for endpoints) and regular pen-testing.

Can I rely on Bandit alone?

Bandit is great for common Python pitfalls, but it doesn’t understand templating or configuration files. Adding Semgrep or SonarQube covers those gaps.

Is it safe to use AI for security-critical code?

Only if you treat the output as untrusted. Run it through the same review, testing, and scanning pipeline you would any third-party library.

How much does this add to CI time?

In my experience the three scanners together add ~30 seconds per PR. The cost is negligible compared to a production breach.

Key Takeaways

AI-generated code often repeats insecure patterns; treat it as third-party code.
Set up Bandit, Semgrep, and SonarQube to create a comprehensive ai generated code vulnerabilities report.
Automate the scans in CI/CD so every pull request is gated by security checks.
Use a small script to turn JSON findings into a readable markdown table for quick triage.
Apply best practices: separate directories, secret management, immediate unit tests, and prompt versioning.
When you need a hand setting this up for your FastAPI service, feel free to reach out at /hire/. I’m happy to help you tighten the loop between AI generation and production security.

How to build AI agent for stock analysis with FastAPI

Ayush Kumar — Sun, 26 Jul 2026 07:26:59 +0000

I’ll show you how to build an AI agent for stock analysis that runs end-to-end in production. The recipe is simple: pull reliable market data, stitch together a LangChain (or CrewAI) workflow, let an LLM write the analysis, add memory for portfolio context, backtest the whole thing, and finally serve it behind FastAPI. Below is a practical walk-through with the code I use in my own trading bots, plus the pitfalls that kept me up at night.

Which market data APIs should I use to build an AI agent for stock analysis?

The first thing that breaks most prototypes is bad data. You need a source that offers near-real-time quotes, historical OHLCV, and a stable free tier if you’re experimenting.

Provider	Real-time?	Historical depth	Free tier	Typical cost
Alpha Vantage	Delayed (15 min)	20 years	5 req/min	$50/mo for premium
IEX Cloud	Real-time (US)	30 years	500 req/day	$9/mo for paid tier
Yahoo Finance (yfinance)	Delayed	25 years	Free (scrape)	N/A
Polygon.io	Real-time + crypto	30 years	5 req/min	$199/mo

For a production-grade agent I combine a cheap real-time feed (IEX) with a free historical dump (yfinance). The real-time feed drives the decision loop; the historical data fuels backtesting and feature engineering.

import yfinance as yf
import requests
import os
from datetime import datetime, timedelta

IEX_TOKEN = os.getenv("IEX_TOKEN")
BASE_URL = "https://cloud.iexapis.com/stable"

def get_realtime_price(symbol: str) -> float:
    url = f"{BASE_URL}/stock/{symbol}/quote?token={IEX_TOKEN}"
    resp = requests.get(url, timeout=5)
    resp.raise_for_status()
    return resp.json()["latestPrice"]

def get_historical(symbol: str, start: str, end: str):
    df = yf.download(symbol, start=start, end=end, progress=False)
    df.reset_index(inplace=True)
    return df

What can go wrong?

Rate limits – IEX will throttle you after a few hundred calls. Cache the last price in Redis and only hit the API when the cache expires (e.g., 60 seconds).

Data mismatches – IEX and yfinance use different ticker conventions; always normalise symbols before you store them.

How do I design the agent architecture with LangChain or CrewAI?

Both LangChain and CrewAI give you a composable graph of LLM calls, tools, and memory. I started with LangChain because its tooling around agents is mature, then migrated a few modules to CrewAI for better parallel tool execution.

A minimal architecture looks like this:

Input layer – FastAPI endpoint receives a ticker list.
Orchestrator – LangChain AgentExecutor decides whether to call a price tool, a news summariser, or the analysis chain.
Tool layer – Python functions wrapped as Tools (price fetch, technical indicator, sentiment).
LLM chain – Prompt template that asks the model to generate a “buy / hold / sell” recommendation with confidence.
Memory – Redis vector store that stores the last N decisions and portfolio exposure.

from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
import redis
import json

# Redis for short‑term memory
redis_client = redis.Redis(host="redis", port=6379, db=0)

def price_tool(symbol: str) -> str:
    price = get_realtime_price(symbol)
    return f"The current price of {symbol} is ${price:.2f}"

def sentiment_tool(symbol: str) -> str:
    # Placeholder: call a news API and summarise with LLM
    return f"Sentiment for {symbol} is neutral."

price = Tool(
    name="price",
    func=price_tool,
    description="Get the latest market price for a ticker."
)

sentiment = Tool(
    name="sentiment",
    func=sentiment_tool,
    description="Return a short sentiment summary for a ticker."
)

prompt = PromptTemplate(
    input_variables=["price", "sentiment", "history"],
    template=(
        "You are a quantitative analyst. Given the price: {price}, "
        "sentiment: {sentiment}, and the last 5 decisions: {history}, "
        "output a JSON with fields: decision (buy/hold/sell), confidence (0‑1), "
        "and a one‑sentence rationale."
    )
)

llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
agent = initialize_agent(
    tools=[price, sentiment],
    llm=llm,
    agent_type="zero-shot-react-description",
    verbose=True,
)

def run_analysis(symbol: str):
    # Pull recent decisions from Redis
    past = redis_client.lrange(f"history:{symbol}", -5, -1)
    past_json = [json.loads(x) for x in past] if past else []
    history_str = ", ".join(d["decision"] for d in past_json)

    # Let the agent decide which tools to call
    result = agent.run(
        {
            "price": price_tool(symbol),
            "sentiment": sentiment_tool(symbol),
            "history": history_str,
        }
    )
    decision = json.loads(result)
    # Store decision for future memory
    redis_client.rpush(f"history:{symbol}", json.dumps(decision))
    redis_client.ltrim(f"history:{symbol}", -20, -1)  # keep last 20
    return decision

When not to use this?

If you only need a static statistical model (e.g., a mean-reversion rule), the overhead of an LLM agent is wasted CPU and cost. Also, LangChain’s AgentExecutor adds latency (~300 ms) per call – not ideal for high-frequency strategies.

Failure modes –

Tool hallucination: The LLM may request a tool that isn’t registered, causing a runtime error. Keep the verbose flag on while debugging.

State drift: Forgetting to prune Redis history leads to stale context influencing new decisions.

How can I add memory and tool use for portfolio management?

A stock-analysis agent is only useful if it knows what you already own. I store two kinds of memory:

Short-term – recent recommendations and positions (Redis).
Long-term – aggregated performance metrics (PostgreSQL).

The short-term store is already in the code above. For portfolio exposure I expose a tiny FastAPI route that reads the Redis cache and writes to a PostgreSQL table every hour.

# fastapi_app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncpg
import aioredis

app = FastAPI()
redis = aioredis.from_url("redis://redis")
pg_pool = None

class Order(BaseModel):
    symbol: str
    action: str  # buy or sell
    qty: int

@app.on_event("startup")
async def startup():
    global pg_pool
    pg_pool = await asyncpg.create_pool(dsn=os.getenv("DATABASE_URL"))

@app.post("/order")
async def place_order(order: Order):
    # naive simulation of a broker call
    if order.action not in {"buy", "sell"}:
        raise HTTPException(400, "Invalid action")
    # Record in Postgres
    async with pg_pool.acquire() as conn:
        await conn.execute(
            "INSERT INTO trades (symbol, action, qty, ts) VALUES ($1,$2,$3,now())",
            order.symbol, order.action, order.qty,
        )
    # Update memory
    await redis.rpush(f"history:{order.symbol}", json.dumps({
        "decision": order.action,
        "confidence": 1.0,
        "timestamp": datetime.utcnow().isoformat()
    }))
    return {"status": "ok"}

The agent can now query its own position before recommending a trade. Add a simple tool:

def position_tool(symbol: str) -> str:
    # Summarise current net position from Postgres
    async def fetch():
        async with pg_pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT SUM(CASE WHEN action='buy' THEN qty ELSE -qty END) AS net "
                "FROM trades WHERE symbol=$1", symbol
            )
            return row["net"] or 0
    net = asyncio.run(fetch())
    return f"Current net position for {symbol} is {net} shares."

Add it to the tools list and the LLM will automatically request it when needed.

Trade-offs –

Consistency: If the PostgreSQL write fails, the Redis cache and DB diverge. Wrap both in a transaction or use an outbox pattern.

Cost: Each LLM call costs a few cents; adding extra tools multiplies the number of calls per analysis. Keep prompts tight.

How do I backtest and deploy the stock analysis agent?

Backtesting proves whether the agent’s logic adds value before you risk real capital. I use backtrader because it integrates nicely with pandas data frames from yfinance.

import backtrader as bt
import pandas as pd

class LLMStrategy(bt.Strategy):
    params = dict(symbol="AAPL", cash=100_000)

    def __init__(self):
        self.dataclose = self.datas[0].close
        self.order = None

    def next(self):
        if self.order:
            return  # wait for order to fill

        decision = run_analysis(self.p.symbol)  # sync call for demo
        if decision["decision"] == "buy" and not self.position:
            self.order = self.buy(size=10)
        elif decision["decision"] == "sell" and self.position:
            self.order = self.sell(size=self.position.size)

# Load data
df = get_historical("AAPL", "2020-01-01", "2023-01-01")
data = bt.feeds.PandasData(dataname=df)

cerebro = bt.Cerebro()
cerebro.addstrategy(LLMStrategy, symbol="AAPL")
cerebro.adddata(data)
cerebro.broker.setcash(100_000)
cerebro.run()
print(f"Final portfolio value: ${cerebro.broker.getvalue():,.2f}")

Production notes

Speed – The backtest calls the LLM for every bar, which is far too slow. Replace the LLM with a deterministic rule derived from the backtest results (e.g., “if confidence > 0.8 and price < MA50, buy”).
Cost – Running a full-year backtest for 10 symbols can cost $30 in API usage. Cache the LLM outputs per day.
Deployment – Containerise the FastAPI service with Docker, use uvicorn --workers 4 behind an Nginx reverse proxy, and schedule a Celery beat job that triggers run_analysis nightly.

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "fastapi_app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Deploy to a Kubernetes cluster with a HorizontalPodAutoscaler that scales on CPU. Keep the Redis and Postgres pods in the same namespace for low latency.

When to skip this pipeline – If you only need a one-off signal for a single ticker, the overhead of Docker, Kubernetes, and backtesting is overkill. A simple script run locally is enough.

FAQ

What’s the cheapest way to get real-time market data for a hobby project?

Use the free tier of IEX Cloud (500 req/day) combined with yfinance for historical data. Cache results aggressively to stay inside the limit.

Can I replace the LLM with a rule-based system?

Yes. If latency or cost is a concern, train a lightweight classifier on the LLM’s historical outputs and use that model in production.

How often should I retrain or update the prompt?

At least monthly, or whenever you notice a drift in confidence scores. Prompt engineering is an ongoing maintenance task.

Do I need a GPU for this stack?

Not for inference with OpenAI’s hosted models. If you run a local model (e.g., Llama-3) you’ll need a GPU, but that adds hardware cost and complexity.

Key Takeaways

Choose a reliable market-data API and cache aggressively; rate limits bite fast.
LangChain (or CrewAI) gives you a clean tool-based agent, but each extra tool adds latency and cost.
Store short-term decisions in Redis and long-term performance in Postgres to keep the agent context-aware.
Backtest with a deterministic surrogate of the LLM to avoid prohibitive API bills.
Deploy the whole stack with FastAPI, Docker, and Kubernetes; monitor Redis, Postgres, and LLM usage to catch failures early.