DEV Community: Ayush Kumar

Fix AI Generated Code: Practical Steps for Backend Engineers

Ayush Kumar — Mon, 14 Sep 2026 05:46:25 +0000

Fix AI generated code by starting with what you can see: broken endpoints, silent failures, or security warnings. I’ve been bitten by this more times than I can count - AI tools spit out code that looks right but fails in production. Here’s how to find and fix the real issues.

What are the most common vulnerabilities in AI generated code?

The most common vulnerabilities I see are hardcoded secrets, SQL injection via string concatenation, and missing input validation. AI often pulls patterns from public repos without understanding context. For example, it might generate a login endpoint like this:

@app.post("/login")
def login(username: str, password: str):
    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    return db.execute(query).fetchone()

This is an SQL injection waiting to happen. Never trust user input in a query string. Fix it by using parameterized queries with SQLAlchemy:

@app.post("/login")
def login(username: str, password: str):
    stmt = text("SELECT * FROM users WHERE username=:username AND password=:password")
    return db.execute(stmt, {"username": username, "password": password}).fetchone()

Another frequent issue is leaving debug modes on or exposing internal errors. AI doesn’t know your deployment setup. Always override defaults in production:

# In your FastAPI app setup
app = FastAPI(debug=False)  # Never True in prod

How do I detect logic errors in AI generated Python?

Detect logic errors by writing tests that mirror real user flows, not just unit tests on functions. AI often gets the “what” right but misses the “when” and “why.” For example, it might generate a user creation endpoint that doesn’t handle duplicate emails:

@app.post("/users")
def create_user(user: UserCreate):
    db_user = UserModel(**user.dict())
    db.add(db_user)
    db.commit()
    return db_user

This will crash on duplicate email if your DB has a unique constraint. Instead, catch the integrity error:

from sqlalchemy.exc import IntegrityError

@app.post("/users")
def create_user(user: UserCreate):
    db_user = UserModel(**user.dict())
    db.add(db_user)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(status_code=400, detail="Email already registered")
    return db_user

Write a test that tries to create two users with the same email. If it passes, your logic holds. I use Pytest with fixtures to isolate DB state:

def test_duplicate_email_rejected(client, db_session):
    client.post("/users", json={"email": "a@b.com", "password": "x"})
    response = client.post("/users", json={"email": "a@b.com", "password": "y"})
    assert response.status_code == 400
    assert "Email already registered" in response.json()["detail"]

How do I fix security flaws in FastAPI endpoints from AI output?

Fix security flaws by assuming every AI-generated endpoint is insecure until proven otherwise. Start with authentication, then input validation, then rate limiting. AI often skips auth entirely or uses fake tokens.

Here’s a typical AI-generated endpoint missing auth:

@app.get("/data")
def get_data():
    return {"sensitive": "info"}

Add real auth using FastAPI’s dependencies. I use JWT with a simple verification function:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    # In real life: verify signature, expiry, etc.
    if token != "valid-token-for-demo":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )
    return token

@app.get("/data")
def get_data(token: str = Depends(verify_token)):
    return {"sensitive": "info"}

Then validate inputs with Pydantic. Never accept raw dicts. AI often does this:

@app.post("/update")
def update_item(data: dict):  # Bad
    return {"received": data}

Fix it with a model:

from pydantic import BaseModel

class UpdateItem(BaseModel):
    item_id: int
    value: str

@app.post("/update")
def update_item(item: UpdateItem):
    return {"updated": item.item_id}

This catches malformed JSON early and documents your API.

How do I validate AI-generated SQLAlchemy models?

Validate SQLAlchemy models by checking constraints, relationships, and migrations. AI often creates models without nullable defaults, wrong cascade rules, or missing indexes. I once saw a model where created_at had no default and wasn’t indexed - causing slow queries and NULL errors.

Here’s a risky AI-generated model:

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String)  # No unique, no index
    password = Column(String)

Fix it by adding constraints and indexes that match your business rules:

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True, nullable=False)
    password = Column(String, nullable=False)
    created_at = Column(DateTime, server_default=func.now(), nullable=False)

Use Alembic to generate migrations after you fix the model. Then run them in a staging DB before deploying. Never skip this step - AI doesn’t know your schema evolution.

How do I test AI-generated API contracts with Pytest?

Test API contracts by treating your OpenAPI spec as the source of truth. AI often generates endpoints that don’t match the declared schema - missing fields, wrong types, or extra keys. I use Pytest to validate responses against the schema.

First, generate your OpenAPI JSON:

curl http://localhost:8000/openapi.json > openapi.json

Then write a test that checks every endpoint:

import jsonschema
import json

with open("openapi.json") as f:
    OPENAPI_SCHEMA = json.load(f)

def test_api_matches_schema(client):
    for path, path_item in OPENAPI_SCHEMA["paths"].items():
        for method, operation in path_item.items():
            if method.lower() in ["get", "post", "put", "delete", "patch"]:
                response = client.request(method, path)
                jsonschema.validate(
                    instance=response.json(),
                    schema=operation["responses"]["200"]["content"]["application/json"]["schema"]
                )

This catches mismatches early. If AI adds a field not in the schema, the test fails. If it omits a required one, same thing. It’s saved me from silent data corruption more than once.

How do I refactor AI-generated async code for production?

Refactor AI-generated async code by removing blocking calls and ensuring proper error propagation. AI often mixes sync and async or forgets to await. I’ve seen endpoints that call time.sleep(5) inside an async function - destroying concurrency.

Here’s a bad example:

@app.get("/slow")
async def slow_endpoint():
    time.sleep(2)  # Blocking!
    return {"done": True}

Fix it by using asyncio.sleep if you need a delay, or better - remove the delay entirely. For real work like HTTP calls, use httpx.AsyncClient:

import httpx

@app.get("/fetch")
async def fetch_external():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/data")
        return resp.json()

Also, watch for missing try/except blocks. AI often omits error handling in async code. Wrap external calls:

@app.get("/fetch")
async def fetch_external():
    try:
        async with httpx.AsyncClient() as client:
            resp = await client.get("https://api.example.com/data")
            resp.raise_for_status()
            return resp.json()
    except httpx.RequestError as e:
        raise HTTPException(status_code=502, detail=f"External error: {str(e)}")

Finally, use logging instead of print. AI loves print - it’s useless in production. Use structlog or Python’s logging module with JSON output.

When NOT to trust AI generated code

Don’t trust AI generated code for authentication, payment processing, or any code handling PII. I’ve seen AI generate OAuth flows that skipped state validation - critical for security. For those, use battle-tested libraries like fastapi-users or python-jose and read the docs.

Also, avoid using AI to generate migration scripts. Schema changes are too risky. Write them yourself or use Alembic’s autogenerate with careful review.

AI is great for boilerplate, repetitive CRUD, or getting unstuck on a tricky algorithm. But production systems need human judgment. Treat AI output like a junior engineer’s first draft: review it, test it, and break it on purpose.

If you’re stuck fixing AI generated code in your FastAPI or data stack, I’ve helped indie builders do this exact work. You can hire me to audit your endpoints, write tests, and make your AI-generated code production ready - no fluff, just fixes.

FAQ

How do I know if my AI generated code has SQL injection?

Look for string concatenation with user input in SQL queries. If you see f"SELECT ... {user_input}" or + user_input +, it’s vulnerable. Fix it with parameterized queries or ORM methods.

Can I use AI to generate my FastAPI auth system?

Not safely. AI often misses token validation, scope checks, or refresh token handling. Use a trusted library and have AI only generate non-critical parts like route stubs or response models.

What’s the fastest way to validate AI generated Python?

Run it through a linter like ruff or flake8, then write a test that exercises the happy path and one edge case. If it passes both, it’s likely safe for low-risk code.

Should I use AI to generate database migrations?

No. Migrations alter your data schema. A mistake can corrupt or lose data. Write them manually, test them in a copy of production, and review every line.

Key Takeaways

Fix AI generated code by assuming it’s insecure and untested until proven otherwise
Use parameterized queries, Pydantic models, and explicit error handling
Test API contracts against your OpenAPI schema to catch mismatches early
Never use AI for auth, payments, or migrations - human review is non-negotiable
Treat AI output as a draft: review, test, and break it before deploying to prod

Postgres Performance Tuning Best Practices for Backend Engineers

Ayush Kumar — Mon, 14 Sep 2026 05:46:20 +0000

Postgres performance tuning best practices start with knowing where your time is spent. I’ve seen teams waste weeks optimizing indexes when the real issue was a misconfigured work_mem or a runaway autovacuum. Let’s cut through the noise.

What are the most common Postgres performance bottlenecks?

The usual suspects are slow queries, lock contention, and I/O saturation. In my experience, slow queries often stem from missing indexes or poor planner estimates - not the query itself. I once spent a day tuning a query only to realize the index wasn’t being used because enable_indexscan was accidentally turned off in a session setting. Check pg_stat_activity for waiting queries and pg_locks for blockers. High buffers_read relative to buffers_hit means you’re hitting disk too often - time to look at shared_buffers or your indexing strategy.

Which configuration parameters actually matter for tuning?

Start with these five: shared_buffers, effective_cache_size, work_mem, maintenance_work_mem, and max_worker_processes. Set shared_buffers to 25% of RAM on a dedicated DB box - more if you’re not using the OS cache heavily. effective_cache_size should reflect total RAM available to Postgres and the OS cache combined. I’ve seen work_mem set too high cause swapping during complex sorts - start with 64MB and monitor pg_stat_activity for sort operations spilling to disk. For write-heavy workloads, bump max_worker_processes to match your CPU cores, but don’t go overboard - each worker eats memory.

How do I monitor Postgres performance in production?

I rely on pg_stat_statements for query-level insights - enable it in shared_preload_libraries and track total_time, calls, and rows. Pair it with pg_stat_bgwriter for checkpoint and buffer write stats. For system-level metrics, use pg_stat_replication if you’re streaming, and always monitor OS-level iostat, vmstat, and netstat. I’ve set up alerts in Prometheus using the Postgres exporter to flag when buffer_cache_hit_ratio drops below 95% or when deadlocks exceed 5/min. Don’t just collect metrics - act on them. A spike in temp_files often means you need to increase work_mem or rewrite a query.

What indexing strategies actually optimize queries?

B-tree indexes are your workhorse - use them for equality and range checks. But don’t index everything. I’ve seen tables with 20 indexes where writes slowed to a crawl because every INSERT had to update all of them. Use EXPLAIN ANALYZE to see if your index is being hit. For JSONB columns, a GIN index on jsonb_path_ops beats the default jsonb_ops for containment queries. Partial indexes save space - index only WHERE status = 'active' if that’s your hot path. And remember: indexes help reads, hurt writes. Benchmark your write load before adding one.

How should I handle connection pooling and worker scaling?

Never let your app open a raw Postgres connection per request. Use a pooler like PgBouncer in transaction mode - I’ve seen it cut connection overhead by 90% in FastAPI apps. Set max_connections in Postgres to 200-300, then let PgBouncer handle thousands of app connections via a smaller pool. In FastAPI, I use SQLAlchemy with pool_size=20 and max_overflow=10 behind PgBouncer. For async apps, asyncpg with a built-in pool works too - but never mix sync and async clients in the same service. If you’re using Kubernetes, scale your app pods based on CPU, not DB connections - let the pooler absorb the variance.

When should I tune vacuum and autovacuum for write-heavy workloads?

Autovacuum is non-negotiable - turning it off leads to transaction ID wraparound disaster. But defaults are too lazy for heavy writes. I tune autovacuum_vacuum_scale_factor to 0.05 (instead of 0.2) and autovacuum_analyze_scale_factor to 0.02 on write-heavy tables. Set autovacuum_naptime to 20s so it runs more frequently. Monitor pg_stat_user_tables for n_dead_tup - if it’s growing faster than vacuum can clean, you’ll see bloat. I’ve had to manually VACUUM FULL a table that hit 60% bloat because autovacuum couldn’t keep up - don’t let it get that far. For append-only tables, consider disabling autovacuum entirely and scheduling manual vacuums during low-traffic windows.

FAQ

How do I know if my Postgres instance is CPU-bound vs I/O-bound?
Check iostat for high %util on your data disk - if it’s consistently over 80%, you’re I/O-bound. If CPU is high but disk is idle, look at complex queries or high work_mem usage causing sorts/hashes in memory.

Is it ever safe to disable autovacuum?
Only on static, append-only tables with manual vacuum schedules - and even then, test wraparound risk first. For 99% of workloads, keep it on and tune the scale factors.

Should I increase max_connections to handle more traffic?
No. Raising max_connections without a pooler increases context switching and memory pressure. Use PgBouncer or similar to multiplex app connections safely.

What’s the fastest way to find missing indexes?
Run SELECT * FROM pg_stat_user_tables WHERE seq_scan > idx_scan; to find tables favoring sequential scans, then check pg_stat_user_indexes for zero-read indexes on those tables.

Key Takeaways

Tune shared_buffers and work_mem based on RAM and workload - monitor for spills to disk.
Use pg_stat_statements to find slow queries; fix missing indexes or planner issues before touching config.
Always use a connection pooler - PgBouncer in transaction mode is essential for production scale.
Adjust autovacuum scale factors for write-heavy tables; monitor bloat via n_dead_tup.
Indexes speed reads but slow writes - benchmark your write path before adding new ones.

RAG Chunking Best Practices for Production Systems

Ayush Kumar — Thu, 10 Sep 2026 11:42:32 +0000

RAG chunking best practices start with understanding that your embedding quality depends entirely on how you split your source text. I've seen teams waste weeks tuning LLMs only to find their retrieval failed because chunks were too big, too small, or ripped context apart at the worst possible moment. Getting chunking right isn't academic - it directly impacts latency, cost, and whether your RAG system actually works in production.

How do I choose the optimal chunk size for LLM embeddings?

The optimal chunk size balances semantic completeness with embedding model limits. For most sentence-transformers and similar models, aim for 256-512 tokens. Smaller chunks (128-256) work better for precise fact retrieval but risk losing context. Larger chunks (512-1024) preserve more narrative but dilute embedding focus and increase noise. I start with 384 tokens as a default - it fits comfortably within most models' context windows while leaving room for query tokens. Test with your actual data: embed chunks of varying sizes, then measure retrieval precision@k on a held-out set. If your documents are mostly short FAQs, lean smaller. For long-form reports or code, go larger - but never exceed your embedding model's max sequence length without truncation, which silently destroys information.

What overlap strategies preserve context in RAG without killing efficiency?

Overlap prevents context loss at chunk boundaries - a silent killer in retrieval accuracy. I use 10-20% overlap (e.g., 50 tokens on a 384-token chunk) as a rule of thumb. This costs minimal extra compute but significantly reduces the chance that a critical phrase gets split. For example, if a definition spans "The term 'X' refers to..." and the next chunk starts mid-sentence, overlap keeps it intact. Avoid fixed-token overlap; instead, overlap by sentences or semantic units when possible. I've seen teams use 50% overlap - it's overkill. It doubles your chunk count, increases vector store load, and adds latency with diminishing returns. Measure your specific failure mode: if you're losing answers that cross boundaries, increase overlap incrementally until precision stabilizes. Don't guess - measure.

How should I chunk tabular and structured data for RAG?

Tabular data breaks naive text chunking. Don't just CSV-dump rows into chunks - it destroys relationships. Instead, treat each row as a semantic unit: convert it to a structured text snippet like "Product: Widget A, Price: $29.99, Category: Electronics, Stock: 15". For hierarchical data (JSON, XML), preserve parent-child context in the chunk - e.g., "User: John Doe, Address: Street: 123 Main St, City: Springfield". I've used this approach in a product catalog RAG system where retrieving specs required joining attributes across columns. Without preserving structure, embeddings became meaningless averages. For wide tables, consider vertical chunking: group related columns (e.g., all pricing fields) into separate chunks. Always link chunks back to their source row ID - you'll need it for attribution and reranking.

How do I evaluate chunking impact on retrieval accuracy and latency?

Evaluation isn't optional - it's how you avoid shipping a broken system. I run two experiments: first, fixed retrieval latency budget (e.g., 200ms) and measure how chunk size/overlap affects recall@k. Second, fixed chunking strategy and measure latency vs. recall trade-offs. Use a representative query set with known relevant documents. Tools like ragas or custom scripts with sklearn.metrics work. Track: recall@k, mean reciprocal rank (MRR), and 95th percentile latency. I once saw a team improve recall by 18% just by increasing overlap from 0% to 15% - latency only went up 8%. But going from 15% to 30% overlap gave another 2% recall for 22% more latency - not worth it. Plot your own curve. Also monitor vector store index size - more chunks mean bigger indexes and higher costs. In production, I log chunk-level retrieval rates to spot drift: if certain chunk types consistently underperform, re-examine your splitting logic.

What tools and libraries help automate chunking in Python?

Don't roll your own chunker unless you have unusual needs. For text, langchain.text_splitter offers RecursiveCharacterTextSplitter (my go-to) and TokenTextSplitter. For semantic chunking, try experimental modules in llama-index or bert-chunker (though the latter needs GPU). For tabular data, I use pandas to iterate rows and jinja2 templates to format snippets - simple, fast, and testable. Example:

from langchain.text_splitter import RecursiveCharacterTextSplitter
import pandas as pd

def chunk_product_catalog(df: pd.DataFrame) -> list[str]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=384,
        chunk_overlap=50,
        length_function=len,
        separators=["\n\n", "\n", " ", ""]
    )
    chunks = []
    for _, row in df.iterrows():
        text = f"Product: {row['name']}, Price: {row['price']}, Category: {row['category']}, Stock: {row['stock']}"
        chunks.extend(splitter.split_text(text))
    return chunks

This keeps row context intact while allowing overlap within formatted text. For production, wrap this in a class with memoization - re-chunking the same data on every restart wastes CPU. I've seen teams use joblib or diskcache to persist chunk embeddings - valid if your source data changes infrequently.

Balancing semantic coherence with computational efficiency

Semantic coherence means chunks represent unified ideas - don't split a sentence, don't break a clause, don't isolate a pronoun from its antecedent. But chasing perfect semantics kills efficiency. I prioritize: first, avoid splitting within named entities or technical terms (use regex guards); second, prefer sentence boundaries; third, allow word-level splits only as a last resort. Efficiency wins when coherence gains plateau. In one legal doc project, we tried spaCy-based sentence chunking - great coherence, but 3x slower than recursive character splitting with minimal recall loss. We switched back and added overlap. Your bottleneck isn't usually the chunker - it's the embedding model and vector search. Spend cycles where they matter: optimizing your embedding batch size or FAISS index parameters. Chunking is a preprocessing step - make it fast, reliable, and good enough. If your retrieval pipeline spends >10% of time chunking, you've over-engineered it.

FAQ

What’s the best chunk size for RAG with LLMs?
Start with 384 tokens for most embedding models - test recall@k on your data to tune between 256-512 based on document type and query patterns.

Should I use overlap in RAG chunking?
Yes, use 10-20% token overlap to preserve context across boundaries - measure latency impact and stop when recall gains diminish.

How do I chunk CSV or Excel data for RAG?
Convert each row to a formatted text snippet preserving column semantics, then apply standard text chunking - keep row IDs for attribution and reranking.

Key Takeaways

Chunk size is a trade-off: 256-512 tokens works for most LLM embedding models - test with your actual retrieval metrics.
Overlap prevents context loss: 10-20% overlap is usually sufficient; more adds cost with little gain.
Structured data needs special handling: convert rows to semantic text snippets before chunking - never treat CSV as raw text.
Measure, don’t guess: evaluate chunking strategies using recall@k and latency on a representative query set.
Use battle-tested tools: langchain.text_splitter for text, pandas + templates for tabular - avoid over-engineering the chunker.

Python Message Queue Between Processes: IPC Options for FastAPI Workers

Ayush Kumar — Wed, 09 Sep 2026 11:38:08 +0000

I use multiprocessing.Queue as a python message queue between processes every day in production FastAPI apps. It’s not flashy but it works when you need to move data between worker pools and async request handlers without adding external dependencies.

How do I implement inter-process communication with multiprocessing.Queue?

You create a Queue instance in your parent process, pass it to child processes via fork or spawn, then use .put() and .get() to exchange data. Here’s a minimal FastAPI background worker pattern:

from multiprocessing import Process, Queue
import time
from fastapi import FastAPI

def worker(q: Queue):
    while True:
        item = q.get()
        if item is None:  # Sentinel to stop
            break
        print(f"Processing {item}")
        time.sleep(0.1)  # Simulate work

app = FastAPI()
q = Queue()
p = Process(target=worker, args=(q,))
p.start()

@app.post("/task")
async def add_task(data: dict):
    q.put(data)
    return {"status": "queued"}

@app.on_event("shutdown")
def shutdown_event():
    q.put(None)
    p.join()

This keeps things simple: no Redis, no RabbitMQ, just built-in Python. I’ve used this pattern for CPU-bound tasks like image resizing or model inference where the GIL blocks asyncio.

What’s the difference between multiprocessing.Queue, pipes, and shared memory?

multiprocessing.Queue is thread- and process-safe, handles serialization automatically, and scales to multiple producers/consumers. Pipes are faster for two-endpoint communication but lack built-in buffering and can deadlock if not managed carefully. Shared memory (via Array or Value) is fastest for raw bytes but requires you to handle synchronization yourself with Locks or Semaphores.

In practice, I reach for Queue first. It’s the safest default. I only drop to pipes when I’m doing low-latency, two-way signaling between exactly two processes - like a control plane and a data plane. Shared memory? Only when I’m moving large numpy arrays and profiling shows serialization is the bottleneck. Even then, I wrap it in a Queue-like interface to avoid footguns.

How do I handle serialization of complex objects in process queues?

Pickle is the default serializer for multiprocessing.Queue. It works for most Python objects but fails with lambdas, local functions, or objects with non-picklable attributes (like open file handles or database connections). I’ve been bitten by this when trying to pass SQLAlchemy sessions across processes - don’t do that.

Instead, pass primitive types or simple dataclasses. If you need to move complex data, serialize it yourself first:

import pickle
from multiprocessing import Queue

def safe_put(q: Queue, obj):
    q.put(pickle.dumps(obj))

def safe_get(q: Queue):
    return pickle.loads(q.get())

This adds overhead but gives you control. For FastAPI workers doing ML inference, I often pass model input as numpy arrays (which pickle handles fine) and keep the model itself loaded in each worker process to avoid repeated serialization.

How do I avoid deadlocks and race conditions in multiprocessing queues?

Deadlocks happen when you forget to consume items or use Queue.join() incorrectly. I once had a worker that crashed silently, leaving the queue full - then the producer blocked forever on .put(). The fix? Always use timeouts and monitor queue size.

try:
    q.put(item, timeout=1.0)
except Full:
    logger.warning("Queue full, dropping task")
    # or retry, or alert

For race conditions: Queue’s internal locks make .put() and .get() safe, but if you’re doing check-then-act (like if not q.empty(): q.get()), you’re already broken. Never rely on .empty() or .qsize() for logic - those are approximate in multiprocessing contexts due to process scheduling.

I’ve seen teams try to build priority queues on top of Queue using these methods. It never ends well. Use a proper priority queue implementation or switch to Redis if ordering matters.

What are the performance benchmarks of different IPC mechanisms in Python?

In my local tests (Linux, Python 3.11, 8-core CPU):

multiprocessing.Queue: ~50k msg/s for small dicts
Pipe (duplex): ~200k msg/s
Shared memory (Array): ~500k msg/s
Redis (local): ~100k msg/s (network overhead)

But raw speed isn’t the whole story. Queue adds ~10µs latency per message vs ~2µs for pipes. For FastAPI background workers handling 1k req/s, that difference is negligible. What matters more is reliability and ease of debugging. I’ve seen shared memory implementations corrupt data when a worker segfaults mid-write. Queue just drops the message - annoying but recoverable.

How do I integrate message queues with FastAPI background workers?

I combine multiprocessing.Queue with FastAPI’s lifespan events and dependency injection. Here’s a cleaner version using a manager class:

from contextlib import asynccontextmanager
from multiprocessing import Process, Queue
from fastapi import FastAPI

class TaskQueue:
    def __init__(self):
        self.q = Queue()
        self.process = None

    def start(self):
        self.process = Process(target=self._worker, args=(self.q,))
        self.process.start()

    def _worker(self, q):
        while True:
            task = q.get()
            if task is None:
                break
            self._process_task(task)

    def _process_task(self, task):
        # Your actual work here
        pass

    def stop(self):
        if self.process:
            self.q.put(None)
            self.process.join()

@asynccontextmanager
async def lifespan(app: FastAPI):
    task_queue = TaskQueue()
    task_queue.start()
    yield {"task_queue": task_queue}
    task_queue.stop()

app = FastAPI(lifespan=lifespan)

@app.post("/task")
async def add_task(data: dict, request: Request):
    request.state.task_queue.q.put(data)
    return {"status": "queued"}

This keeps the queue lifecycle tied to the app. I’ve used this in RAG pipelines where the worker does embedding generation and the API just enqueues text chunks. For more on choosing between built-in queues and external brokers, see my post on choosing a python message queue library for production.

When NOT to use multiprocessing.Queue

Don’t use it if:

You need persistence across reboots (use Redis or RabbitMQ)
You’re crossing machine boundaries (use a network queue)
You require strict ordering or transactions (Queue gives none of these)
Your workers are mostly I/O-bound (stick with asyncio and background tasks)

I’ve seen teams try to scale this to 50+ processes and hit context-switching overhead. At that point, a proper message broker wins. But for 2-8 workers on a single box doing CPU-heavy work? multiprocessing.Queue is still my go-to.

FAQ

Can I use multiprocessing.Queue with async FastAPI endpoints?

Yes, but don’t call .get() directly in an async route - it blocks the event loop. Use a separate worker process or run the Queue call in a threadpool via run_in_executor.

Is multiprocessing.Queue safe for multiple producers and consumers?

Yes. The Queue class handles locking internally so multiple processes can safely call .put() and .get() concurrently.

What’s the maximum size of a multiprocessing.Queue?

By default, it’s unlimited unless you specify a maxsize. If you set maxsize, .put() will block when the queue is full unless you use a timeout.

How do I monitor queue size in production?

You can’t rely on .qsize() for accuracy due to timing, but you can approximate it for logging. For real metrics, wrap your Queue in a proxy class that increments/decrements counters on put/get.

Key Takeaways

multiprocessing.Queue is a reliable, built-in python message queue between processes for FastAPI workers
Use it for CPU-bound tasks where you don’t need persistence or network transparency
Serialize carefully - prefer primitives over complex objects
Avoid .empty() and .qsize() for control flow; use timeouts and sentinel values instead
Benchmark your specific workload, but prioritize correctness over raw speed
Link to related production tips: FastAPI Async vs Sync: Benchmarks and When to Use Each and Bolt.new Python Backend: FastAPI from Zero to Production

Redis caching in FastAPI: practical setup and pitfalls

Ayush Kumar — Mon, 07 Sep 2026 05:35:20 +0000

Redis caching in FastAPI: practical setup and pitfalls is something I’ve wrestled with in production for over two years. It’s not just about slapping a cache layer on - it’s about knowing where it helps, where it hurts, and how to keep it from silently breaking your system. Here’s what I’ve learned running FastAPI services that handle tens of thousands of requests per day, backed by Redis for session storage and result caching.

How do I set up Redis with FastAPI for session caching?

Start with a lightweight async Redis client. I use redis-py with asyncio support because it plays nicely with FastAPI’s async endpoints. First, install the deps: pip install redis[async] python-jose[cryptography]. Then create a Redis connection pool at app startup - never create a new connection per request. That’s a rookie mistake I made early on, and it killed performance under load.

Here’s how I initialize it in main.py:

from fastapi import FastAPI
from redis.asyncio import Redis
import aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.redis = Redis.from_url("redis://localhost:6379", encoding="utf-8", decode_responses=True)

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

For sessions, I store JSON-serialized user data with a TTL. Here’s a dependency that gets or sets session data:

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_session(token: HTTPAuthorizationCredentials = Depends(security)):
    redis: Redis = app.state.redis
    session_data = await redis.get(f"session:{token.credentials}")
    if not session_data:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired session")
    return json.loads(session_data)

I set sessions to expire after 24 hours - short enough to limit risk, long enough to avoid annoying users. But I’ve seen teams set TTLs to weeks, then wonder why old sessions linger after a password reset. Don’t do that. Always invalidate on critical events.

How do I use Redis as a result cache for expensive API calls?

This is where Redis shines - caching the output of slow external APIs or heavy DB queries. I wrap expensive functions with a decorator that checks Redis first. The key includes the function name and hashed arguments to avoid collisions.

import hashlib
import json
from functools import wraps

def cache_result(ttl: int = 300):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            redis: Redis = app.state.redis
            key = f"cache:{func.__name__}:{hashlib.sha256(json.dumps((args, kwargs), sort_keys=True).encode()).hexdigest()}"
            cached = await redis.get(key)
            if cached:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await redis.set(key, json.dumps(result), ex=ttl)
            return result
        return wrapper
    return decorator

I use this on endpoints that call third-party ML models or aggregate analytics. A 5-minute TTL works for most - stale data is better than no data during a spike. But I learned the hard way: never cache POST/PUT/DELETE results. Only cache idempotent GETs. I once cached a payment confirmation endpoint - yeah, don’t be me.

What are effective cache invalidation strategies in FastAPI?

Time-based TTLs are lazy. Smart invalidation is better. I use two patterns: explicit deletes on data mutation, and tag-based invalidation for related data.

When a user updates their profile, I delete their session and any cached user-specific data:

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: UserUpdate):
    # ... update DB ...
    redis: Redis = app.state.redis
    await redis.delete(f"session:{user_id}")  # if tied to user
    await redis.delete(f"cache:get_user:{user_id}")
    await redis.delete(f"cache:get_user_posts:{user_id}")
    return {"status": "updated"}

For broader invalidation - like when a product price changes and affects multiple caches - I use tags. I maintain a Redis set per tag, and cache keys are added to those sets. On invalidation, I fetch all keys in the tag and delete them. It adds complexity, but it’s worth it for frequently related data.

How do I handle Redis connection failures gracefully in FastAPI?

Assume Redis will fail. It will. Network blips, restarts, OOM kills - they happen. I wrap Redis calls in a circuit breaker pattern. If three consecutive calls fail, I switch to a fail-open mode: skip the cache and go straight to the source, but log loudly and alert.

Here’s a simplified version:

class RedisCache:
    def __init__(self, redis: Redis):
        self.redis = redis
        self.failures = 0
        self.max_failures = 3
        self.open_until = 0

    async def get(self, key: str):
        if time.time() < self.open_until:
            return None  # fail-open: skip cache
        try:
            return await self.redis.get(key)
        except Exception:
            self.failures += 1
            if self.failures >= self.max_failures:
                self.open_until = time.time() + 60  # open for 60s
            return None

I also set timeouts on the Redis client itself - socket_connect_timeout=2, socket_timeout=2 - so a hung Redis doesn’t tie up workers. And I monitor: if fail-open mode activates more than once an hour, I page someone. Cache should never be a single point of failure.

Redis vs in-memory caching for FastAPI: when should I use which?

I’ve used both. In-memory (like lru_cache or cachetools) is blazing fast - no network hop. But it’s per-process. In a multi-worker setup (Uvicorn with Gunicorn, say), each worker has its own cache. That means cache misses multiply, and memory usage scales with worker count. I’ve seen services OOM because someone cached 100MB of data in-memory across 8 workers.

Redis adds ~1ms latency per call (localnet), but it’s shared. One source of truth. For session stores or result caches that need consistency across instances, Redis wins. For tiny, read-heavy, worker-local data (like config or static lookup tables), in-memory is fine - just know its limits.

I use in-memory for things like API key validation (low cardinality, infrequent change). For anything that scales with users or requests - sessions, API results, rate limiting - Redis is the safer choice.

How do I monitor Redis cache hit rates in production?

You can’t optimize what you don’t measure. I track two metrics: hit rate (hits / (hits + misses)) and memory usage. I use redis-cli info stats scraped by Prometheus, or better yet, export via redis_exporter.

In FastAPI, I instrument my cache layer to increment counters:

from prometheus_client import Counter, Gauge

CACHE_HITS = Counter('cache_hits_total', 'Total cache hits')
CACHE_MISSES = Counter('cache_misses_total', 'Total cache misses')
CACHE_MEMORY = Gauge('cache_memory_bytes', 'Redis memory used')

# In get/set methods:
# On hit: CACHE_HITS.inc()
# On miss: CACHE_MISSES.inc()
# Periodically: CACHE_MEMORY.set(await redis.info('memory')['used_memory'])

A healthy hit rate for result caching is 60-80%. Below 40%, I look at TTLs or key design. Above 90%, I wonder if I’m caching too aggressively - maybe the data isn’t that expensive to compute. I alert if hit rate drops suddenly or memory grows past 80% of maxmemory.

I’ve been bitten by assuming “it’s just cache” - until a Redis OOM killed session auth during a Black Friday spike. Now I treat cache like any other critical service: monitor it, test failure modes, and never let it become a silent bottleneck.

FAQ

Can I use Redis for rate limiting in FastAPI?

Yes - I use a token bucket or fixed window counter stored in Redis with short TTLs (e.g., 60s). It’s atomic and shared across workers. Just watch for clock skew if you distributed Redis.

Should I cache database query results directly?

Only if the query is expensive and the data doesn’t change often. I prefer caching at the API layer - it’s easier to invalidate based on business events than table changes.

What’s a safe starting TTL for API result caches?

Start with 5-15 minutes for most external API calls. Adjust based on data volatility and staleness tolerance. Monitor hit rate and backend load to tune.

How do I test cache logic in FastAPI tests?

I use fakeredis in pytest - it mimics Redis API without needing a real server. I inject it via dependency override in tests. See my guide on FastAPI Testing With Pytest: Practical Patterns for patterns.

Key Takeaways

Initialize Redis once at app startup - never per request - to avoid connection storms.
Cache only idempotent GET responses; never cache mutating operations.
Use explicit deletes or tag-based invalidation for data mutations - don’t rely solely on TTL.
Wrap Redis calls in failure handling (timeouts, circuit breakers) to avoid cascading failures.
Prefer Redis over in-memory caching for shared state; use in-memory only for tiny, worker-local data.
Monitor hit rate and memory - alert on sudden drops or sustained high usage.
Treat cache as infrastructure: test failure modes, scale it, and don’t let it become a silent single point of failure.

Vibe Coding vs Agentic Coding: When to Use Which

Ayush Kumar — Wed, 02 Sep 2026 10:51:43 +0000

Short answer: vibe coding leans on AI-generated scaffolding that you tweak, while agentic coding treats an AI as an autonomous teammate that can run, test, and refactor code on its own. Knowing the trade-offs helps you decide which style to apply in a FastAPI service that has to stay reliable under load.

I’ve been building FastAPI back-ends for AI products for three years. I’ve watched teams start with a “write everything in one prompt” vibe, then hit mysterious bugs, timeouts, or security warnings. Later I introduced an agentic loop that could spin up a sandbox, run unit tests, and propose fixes. The shift saved weeks of firefighting, but it also required new monitoring and cost controls.

Below I break down vibe coding vs agentic coding, compare the day-to-day workflow, show concrete FastAPI snippets, and list the pitfalls you’ll hit if you ignore them.

What is vibe coding?

Vibe coding is the practice of prompting an LLM (Cursor, Claude, ChatGPT, etc.) for a full file or function, then pasting the result into your project. The AI supplies the vibe – the overall structure, naming conventions, and a first pass at logic. You are still the one who decides where the code lives, runs it, and fixes the errors.

Typical flow:

Write a prompt like “FastAPI endpoint that accepts a JSON payload, validates with Pydantic, and stores in PostgreSQL.”
Copy the generated main.py snippet into your repo.
Run uvicorn locally, watch the traceback, and edit until it works.

What breaks most often? Missing imports, mismatched async/sync calls, and subtle type errors that only appear when the endpoint is hit under real traffic. In my experience, the biggest cost is the time spent hunting those “it works in the IDE” bugs.

What is agentic coding?

Agentic coding treats the LLM as an autonomous agent that can execute code, run tests, and even open pull requests. You give the agent a goal – “add a new /predict route that calls the TensorFlow model and returns a confidence score” – and the agent:

Generates the code in a temporary workspace.
Spins up a Docker container or a virtualenv, runs pytest.
If tests pass, creates a PR; if they fail, iterates with a new prompt.

The agent is effectively a “coding robot” that can self-correct. The term agentic comes from the AI research community, where an “agent” perceives, decides, and acts in an environment. In practice, tools like AutoGPT, LangChain agents, or custom scripts built on top of OpenAI function calling provide this capability.

What breaks most often? The agent can get stuck in an infinite loop of “regenerate until tests pass” if the test suite is flaky, or it can generate code that violates security policies (e.g., unsafe eval). You need guardrails: timeouts, static analysis, and a review step before merging.

How does vibe coding vs agentic coding change my workflow and output?

Aspect	Vibe Coding	Agentic Coding
Speed of first draft	Instant – you get a file in seconds.	Slightly slower – the agent must spin up a sandbox and run tests.
Reliability	Depends on your manual QA; often fragile.	Higher, if you have a solid test suite that the agent respects.
Cost	Mostly developer time.	Compute cost for containers + LLM calls (often 2-3× higher).
Collaboration	Human-centric; code reviews are straightforward.	Requires a “review the agent’s PR” step; may need extra reviewers.
Scalability	Hard to scale across many micro-services; each file is handcrafted.	Scales well when you standardize the agent’s prompt library.

In a FastAPI project that serves an LLM-powered recommendation engine, I started with vibe coding to scaffold the /recommend endpoint. It compiled, but the async call to the model timed out under load. I switched to an agentic loop that generated a background task using FastAPI’s BackgroundTasks, added a retry decorator, and automatically updated the Dockerfile. The new code passed my integration test suite on the first try.

Can I mix vibe coding and agentic coding in a FastAPI project?

Absolutely. Most production systems benefit from a hybrid approach: use vibe coding for exploratory prototypes, then let an agent polish, test, and ship the code.

Example: Adding a new `/classify` route

Step 1 – Vibe prompt (quick prototype)

# prompt to LLM
"""
Write a FastAPI POST endpoint `/classify` that accepts a JSON body
with a field `text: str`. Use a pre‑trained HuggingFace sentiment
model to return `label` and `score`. Validate input with Pydantic.
"""

Result (copy-paste into app/api.py):

from fastapi import APIRouter
from pydantic import BaseModel
from transformers import pipeline

router = APIRouter()
sentiment = pipeline("sentiment-analysis")

class TextPayload(BaseModel):
    text: str

@router.post("/classify")
async def classify(payload: TextPayload):
    result = sentiment(payload.text)[0]
    return {"label": result["label"], "score": result["score"]}

Run locally – it works for a single request but crashes when the model loads multiple times (memory leak).

Step 2 – Agentic refinement

Create an agent script (using LangChain’s AgentExecutor) that:

Detects the memory issue.
Refactors the code to load the model once at startup.
Writes a unit test that simulates 100 concurrent requests.
Opens a PR with the changes.

# pseudo‑code for the agent (not full implementation)
from langchain.agents import initialize_agent, Tool
from tools import run_pytest, open_pull_request

def fix_memory_leak():
    # generate new code with model as a global singleton
    new_code = """
sentiment = pipeline("sentiment-analysis")

@router.post("/classify")
async def classify(payload: TextPayload):
    result = sentiment(payload.text)[0]
    return {"label": result["label"], "score": result["score"]}
"""
    return new_code

agent = initialize_agent(
    tools=[Tool(name="pytest", func=run_pytest), Tool(name="pr", func=open_pull_request)],
    llm=ChatOpenAI(),
    agent_type="zero-shot-react-description"
)

agent.run("Fix the memory leak in the classify endpoint and ensure the test suite passes.")

After the agent finishes, the PR contains the corrected singleton pattern and a new tests/test_classify.py. The CI pipeline runs, all tests pass, and the PR merges.

Key takeaway: Use vibe coding to get a rough shape, then hand the file to an agent for reliability. The agent can also enforce security checks – see my post on Fixing an AI Generated Code Vulnerabilities Report for details.

Best practices and common pitfalls for each approach

Vibe Coding

Best practices

Keep prompts narrow. “FastAPI endpoint that returns JSON” is too vague; specify request model, async vs sync, and error handling.
Run a linter (ruff, flake8) immediately after pasting. It catches missing imports before you run the server.
Write a minimal test (even a single assert) before you trust the code.

Pitfalls

Assuming the AI knows your production environment (Docker base image, env vars). It will generate code that works locally but fails in CI.
Over-relying on the AI's default security posture. It may suggest eval(payload.text) for quick parsing – never accept that without review.
Ignoring async pitfalls. A common failure mode is mixing requests (sync) with FastAPI’s async routes, leading to “event loop is closed” errors.

Agentic Coding

Best practices

Define a strong test suite first. The agent’s value is proportional to the quality of the tests it runs against.
Sandbox the agent. Use Docker or uvicorn --reload in an isolated environment so the agent can’t affect production resources.
Set explicit timeouts on LLM calls and container runs. Prevent runaway loops that rack up costs.

Pitfalls

Forgetting to audit the PR. Even if tests pass, the agent might introduce a subtle logic change that breaks a business rule.
Ignoring cost monitoring. Each generation call costs tokens; each container spin-up costs compute. Track these in your observability platform.
Relying on a single agent. Different LLMs excel at different tasks; sometimes Claude writes better docstrings, while GPT-4 nails type hints.

When not to use either approach

Regulatory environment: If you must certify every line of code (e.g., medical devices), AI-generated code may not meet audit requirements.
Zero-downtime guarantees: Vibe code that you manually merge can be reviewed in a controlled release pipeline. An agentic PR that auto-merges could unintentionally introduce latency spikes.
Very small teams: The overhead of maintaining an agent (Docker, prompts, monitoring) may outweigh the benefits if you only ship one endpoint a month.

Putting it all together in a FastAPI production pipeline

Prototype with vibe coding. Get a skeleton endpoint.
Add tests (pytest-asyncio) that cover happy path, validation errors, and load.
Run the agent (via a CI step) to refactor, add logging, and enforce best practices.
Deploy using the serverless pattern I described in Serverless Python: Deploying FastAPI to Google Cloud Run with Docker.
Monitor for the two common failures: missing async handling and security warnings from the vulnerability report tool.

Here’s a minimal CI job (GitHub Actions) that runs the agent after tests:

name: CI

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - name: Install deps
        run: pip install -r requirements.txt
      - name: Run tests
        run: pytest -q
      - name: Run agentic refactor
        if: success()
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python scripts/run_agent.py \
            --goal "Refactor FastAPI routes for async safety and add logging"

If the agent produces a new commit, you can automatically open a PR or push directly to a dev branch for further review.

FAQ

Q: Does vibe coding work for database migrations?

A: It can generate Alembic scripts, but you still need to run alembic upgrade head and verify the schema. The AI won’t know about existing data constraints, so manual review is mandatory.

Q: How much does an agentic loop cost per PR?

A: Roughly $0.02–$0.05 for the LLM calls plus a few cents for container runtime. In high-volume teams the cost adds up, so track it with a budget alert.

Q: Can I use agentic coding with serverless deployments?

A: Yes. The agent can output a Dockerfile or a Cloud Run service definition, then trigger a Cloud Build. Just make sure the generated image passes your security scan.

Q: What if the agent fails to fix a bug?

A: Configure a fallback: after N unsuccessful attempts, the pipeline stops and alerts a human. The logs will show the LLM prompts and responses for debugging.

Key Takeaways

Vibe coding = fast scaffold, heavy manual QA. Good for quick prototypes, but expect bugs around imports, async, and security.
Agentic coding = AI as a self-testing teammate. Higher reliability if you have a solid test suite and sandboxing.
Mix both: prototype with vibe, then let an agent polish and enforce standards.
Always keep a human review step before production merges; AI can still hallucinate.
Monitor costs and performance; the extra compute for agents is real, not free.

If you’ve hit a wall trying to get AI-generated code to run reliably in FastAPI, or you need a custom agent built for your specific workflow, feel free to reach out through the hire page. I’m happy to help you set up a production-grade pipeline that blends vibe and agentic coding without the typical surprises.

Hybrid Search RAG Implementation: A Practical Guide

Ayush Kumar — Wed, 02 Sep 2026 10:28:45 +0000

What is hybrid search and why combine vector similarity with traditional keyword search?

Hybrid search blends dense vector similarity with classic inverted-index keyword matching. The first sentence answers the question: it lets you capture semantic relevance while preserving the precision of term-based lookup. In practice, pure vector search can miss exact phrase matches, and pure keyword search can overlook paraphrases. By running both in parallel and merging the scores, you get the best of both worlds - recall from embeddings and precision from terms.

I hit this wall early on when my chatbot kept returning answers that were on topic but missed the exact entity the user asked for. Adding a keyword fallback cut the error rate in half without any extra model calls.

What are the core components of a hybrid RAG architecture?

A hybrid RAG pipeline consists of four moving parts: a document loader, a chunker, a vector store that also exposes a keyword index, a retriever that fuses the two scores, and finally a generator (usually an LLM) that consumes the retrieved context. The first sentence answers the question: each component must be production-ready, observable, and cheap enough to run at scale.

Document loader & chunker – Pull PDFs, HTML, DB rows; split into 500-token chunks with overlap.
Embedding model – OpenAI text-embedding-3-large or a local Mistral encoder.
Hybrid-enabled vector store – Qdrant, Pinecone, or Milvus can store both vectors and a BM25 index.
Retriever – LangChain’s HybridRetriever or LlamaIndex’s VectorStoreRetriever with keyword mode.
LLM – OpenAI gpt-4o-mini or any self-hosted model behind FastAPI.

When any of these pieces misbehave, the whole pipeline stalls. I’ve been bitten by a stale BM25 index that never refreshed after new docs arrived, causing the keyword arm to return empty results. The fix was to hook the indexing step into the same async job that writes the vectors.

How can I implement a hybrid search RAG using LangChain and a vector database?

Below is a minimal but production-ready script. It assumes you have a FastAPI endpoint that receives a user query and returns the LLM answer.

# hybrid_rag.py
import os
from fastapi import FastAPI, HTTPException
from langchain_community.vectorstores import Qdrant
from langchain_community.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers import HybridRetriever
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 1️⃣ Load and chunk docs (run once or on a schedule)
def ingest_documents(path: str):
    from pathlib import Path
    from langchain_community.document_loaders import TextLoader

    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    docs = []
    for file in Path(path).glob("*.md"):
        loader = TextLoader(str(file))
        raw = loader.load()
        docs.extend(splitter.split_documents(raw))
    return docs

# 2️⃣ Create Qdrant collection with hybrid support
def init_vector_store(docs):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
    # Qdrant must be started with `vectors_config` enabling HNSW and `payload_index` for BM25
    vector_store = Qdrant.from_documents(
        docs,
        embeddings,
        url=os.getenv("QDRANT_URL", "http://localhost:6333"),
        collection_name="hybrid_rag",
        # Enable hybrid search (BM25 on payload fields)
        payload_field="text",
        distance="Cosine"
    )
    return vector_store

# 3️⃣ Build a hybrid retriever
def get_hybrid_retriever(vector_store):
    return HybridRetriever(
        vectorstore=vector_store,
        # weight controls contribution of each arm; 0.6 = 60% vector, 40% BM25
        alpha=0.6,
        k=5,  # top‑k per arm before merging
    )

# 4️⃣ Wire LLM + RetrievalQA
def build_qa_chain(retriever):
    llm = OpenAI(model="gpt-4o-mini")
    return RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True,
    )

# FastAPI entry point
app = FastAPI()
vector_store = None
qa_chain = None

@app.on_event("startup")
def load_pipeline():
    global vector_store, qa_chain
    docs = ingest_documents("./data")
    vector_store = init_vector_store(docs)
    retriever = get_hybrid_retriever(vector_store)
    qa_chain = build_qa_chain(retriever)

@app.get("/ask")
def ask(question: str):
    if not qa_chain:
        raise HTTPException(status_code=503, detail="Pipeline not ready")
    result = qa_chain({"query": question})
    return {
        "answer": result["result"],
        "sources": [doc.metadata.get("source") for doc in result["source_documents"]],
    }

Why this works in production

Async-friendly – FastAPI runs the retrieval in the same event loop; Qdrant’s HTTP API is non-blocking.
Observability – Each step logs latency; you can plug Prometheus metrics into the on_event hooks.
Cost control – The alpha parameter lets you lean more on the cheap BM25 side for high-traffic queries, falling back to vectors only when needed.

I learned the hard way that loading all embeddings at startup caused a 30-second cold start on my 2-core VPS. The fix was to persist the vector store on disk and lazy-load only the metadata needed for BM25, keeping the vector cache warm with a background thread.

Which vector store should I pick for hybrid search and how do I configure it?

All three major providers - Qdrant, Pinecone, Milvus - support hybrid retrieval, but each has quirks.

Store	Hybrid support	Pricing / limits	Ops pain points
Qdrant	Built-in BM25 payload index; open-source, self-hosted or SaaS	Free tier on SaaS, modest CPU/RAM on self-hosted	Need to allocate enough RAM for payload storage; occasional GC pauses
Pinecone	“Hybrid” mode combines vector and metadata filters; no native BM25	Pay-as-you-go, generous free tier	Vendor lock-in, limited custom scoring functions
Milvus	Supports scalar field indexing (e.g., IVF_FLAT + BM25)	Open source, but requires a cluster for high QPS	Complex helm charts, harder to monitor

Quick Qdrant config snippet

# qdrant.yaml
service:
  grpc_port: 6334
  http_port: 6333
points:
  vectors:
    config:
      distance: Cosine
      size: 1536
payload_index:
  text:
    type: text
    tokenizer: default
    min_ngram: 3
    max_ngram: 3
    stop_words: []

Deploy with Docker:

docker run -p 6333:6333 -p 6334:6334 \
  -v $(pwd)/qdrant.yaml:/qdrant/config/qdrant.yaml \
  qdrant/qdrant

When I first tried Pinecone’s hybrid mode, I discovered the “metadata filter” approach was slower than a true BM25 index, especially for long queries. Switching to Qdrant shaved 120 ms off the 95th percentile latency.

How do I evaluate and tune hybrid search performance?

Evaluation is a two-step process: relevance testing and latency budgeting.

Ground-truth dataset – Build a set of question-answer pairs with known source documents.
Metric suite – Use ragas for factuality, answer relevance, and context recall. The internal guide "RAG evaluation using ragas: a practical guide" walks through the exact script.
A/B test alpha – Vary the alpha weight from 0.0 (pure keyword) to 1.0 (pure vector) and capture nDCG@10. I found 0.55 gave the highest overall score for a mixed corpus of legal contracts and FAQs.
Latency profiling – Wrap the retriever call with time.perf_counter(). If vector search exceeds 200 ms, fall back to keyword-only for that request and retry the vector arm asynchronously.
Cache hot paths – Frequently asked queries can be cached at the LLM layer; the cache key is the raw question plus the chosen alpha. This saved ~30 % of LLM token usage in my production bot.

When NOT to use hybrid search

If your corpus is tiny (<10 k docs) and you have a strong lexical signal (e.g., product SKUs), pure keyword search is cheaper and easier to monitor. Also, if you run on a strict budget and cannot afford vector storage (GPU-enabled embeddings can be pricey), stick to BM25 and consider periodic re-embedding only when the domain shifts.

FAQ

Q: Can I use LangChain’s HybridRetriever with a self-hosted Milvus instance?

A: Yes, but you need to expose a separate BM25 service (e.g., Elasticsearch) and manually merge scores, because Milvus does not ship a native keyword index yet.

Q: How often should I re-embed documents?

A: For static knowledge bases, once per quarter is fine. For rapidly changing data (e.g., news), set up a nightly pipeline that re-runs the ingestion script.

Q: Does hybrid search increase LLM token cost?

A: Only marginally. You still send the same number of retrieved chunks to the LLM. The extra token cost comes from the optional source_documents field, which you can drop in production.

Q: What monitoring metrics are most useful?

A: Track retriever_latency_ms, vector_search_hits, keyword_search_hits, and fallback_rate. Alert if fallback exceeds 20 % of traffic.

Key Takeaways

Hybrid search combines semantic vector similarity with exact keyword matching, delivering higher recall and precision.
Core components are loader → chunker → embeddings → hybrid-enabled vector store → retriever → LLM.
LangChain’s HybridRetriever plus Qdrant (or Pinecone/Milvus) gives a concise, production-ready stack.
Choose the vector store based on latency, cost, and ops expertise; Qdrant is the most flexible for on-premise setups.
Evaluate with ragas, tune the alpha weight, monitor latency, and cache hot queries to keep costs in check.

By following the steps above you can ship a hybrid search RAG implementation that survives real-world traffic, handles edge cases gracefully, and stays within budget. Happy building!

Why ai generated code cannot be copyrighted and what to do

Ayush Kumar — Wed, 02 Sep 2026 10:28:26 +0000

I’ll get straight to the point: ai generated code cannot be copyrighted, which means you don’t own the snippet the way you own hand-written functions. In practice that shows up as licensing headaches, open-source compliance surprises, and sometimes a cease-and-desist from a vendor who claims the model’s output is theirs. Below I walk through the legal landscape, how to license safely, and what I’ve learned protecting IP in production.

What is the legal status of AI-generated code in the US, EU, and elsewhere?

The short answer is that most jurisdictions treat AI-generated works as non-human creations, and therefore not eligible for copyright protection.

United States – The U.S. Copyright Office’s policy (2023) states that works created without human authorship are not copyrightable. A prompt-to-code interaction is viewed as a tool, not an author.
European Union – The EU’s Copyright Directive leaves the author definition open, but case law (e.g., Infopaq and SAS-Institute) leans toward requiring a human “intellectual contribution”. Most member states follow that line.
United Kingdom – The UK’s Copyright, Designs and Patents Act 1988 allows “computer-generated works” if there is a “human author” who made the necessary choices. The threshold is low, but a pure AI output still falls short.
Canada & Australia – Both require a human author. The Australian Copyright Council’s 2022 briefing explicitly says AI-generated code is not protected.

In every major market the default is no copyright unless you can prove a human contributed enough original expression. That’s why you’ll see “AI-generated code cannot be copyrighted” pop up in legal memos and blog posts.

Why this matters for builders

When you ship a FastAPI microservice that contains a snippet copied from a Claude suggestion, you can’t claim exclusive rights over that snippet. If the same snippet appears in a competitor’s repo, you have no legal ground to stop them. Worse, the model provider may claim the output belongs to them, especially if the model was trained on copyrighted material.

How do I properly license AI-generated snippets?

Treat every AI-suggested block as a third-party component that needs a license. Here’s a practical workflow I use in production:

Capture provenance – Store the prompt, model version, and timestamp alongside the generated file. A simple JSON log works:

{
  "file": "app/router.py",
  "prompt": "Create a FastAPI endpoint that returns the sum of two numbers",
  "model": "claude-2.1",
  "version": "2024-08-01",
  "generated_at": "2024-08-29T14:22:00Z",
  "license": "MIT"
}

Run a license detector – Tools like FOSSology or the open-source project Licensee can scan the snippet for known copyrighted patterns. I integrate the check into CI:

# .github/workflows/license-check.yml
name: License Check
on: [push, pull_request]
jobs:
  detect:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Scan generated code
        run: |
          pip install licensee
          licensee detect ./generated/

Assign a permissive license – If the snippet is original (no matching patterns), I add an MIT header. If you’re unsure, default to Apache-2.0; it includes an explicit patent grant that can protect you from later claims.

# generated/sum_endpoint.py
# SPDX-License-Identifier: MIT
from fastapi import APIRouter

router = APIRouter()

@router.get("/sum")
def sum_endpoint(a: int, b: int):
    """Return the sum of a and b."""
    return {"result": a + b}

Document intent – Add a comment linking back to the prompt log. Future auditors love that trail.

# Prompt: "FastAPI endpoint for sum"
# Model: claude-2.1, 2024‑08‑29

Following this process keeps you on the safe side of both copyright law and open-source compliance.

What are the best practices for protecting IP when using AI code assistants?

Never ship raw AI output – Run it through your normal linting, type-checking, and security scans. The Fixing AI Generated Code Quality Issues in Production guide shows how a missing await caused a race condition in a production service.
Version-lock the model – Record the exact model version you used. If the provider updates the model and the new output is later deemed infringing, you have a record of the safe version you shipped.
Separate AI-generated files – Place them in a generated/ directory. That isolates them for licensing audits and makes it easy to replace a snippet if a legal issue arises.
Use a corporate policy – Draft a short policy that says: “All AI-generated code must be reviewed, logged, and licensed before merge.” Enforce it with a GitHub branch protection rule that requires the license-check workflow to pass.
Consider a “dual-license” approach – If you want to keep a commercial edge, release the AI-generated parts under a commercial license while the rest stays open source. Be clear which files fall under which license.

How does this affect open-source contributions and commercial projects?

Open-source projects often accept contributions without checking provenance. If a contributor pastes an AI-generated snippet, the project may unintentionally incorporate unlicensed code. This can force a retroactive relicensing or cause the project to be taken down.

In a commercial setting, the risk is higher. A client could demand that you remove a piece of AI-generated code after you’ve shipped it, citing copyright concerns from the model provider. The cost of a hot-fix, regression testing, and legal review can quickly eclipse the time saved by the AI assistant.

My rule of thumb: treat every AI suggestion as a potential third-party dependency and handle it like any other library. That mindset saved me when a client asked me to replace a Claude-generated JWT validation routine that turned out to match code from an existing open-source library with a GPL license. I swapped it out, updated the requirements.txt, and the project stayed compliant.

Frequently asked questions and common misconceptions

Does using an AI assistant automatically put my code under the provider’s license?

No. The provider’s terms of service may grant you a license to use the output, but they rarely claim ownership. The real issue is whether the output contains copyrighted material from the training set. That’s why the provenance log is crucial.

Can I claim copyright if I heavily edit an AI-generated snippet?

Possibly, but the edit must be substantial and add original expression. A simple rename or formatting change is not enough. Courts have yet to define a precise threshold, so err on the side of caution.

Are there any jurisdictions where AI-generated code is copyrightable?

A few, like Japan, have started to recognize computer-generated works under a “neighboring right” framework, but the protection is limited and still requires a human author for full copyright. For most builders targeting a global market, the default is non-copyrightable.

Should I avoid AI tools altogether to stay safe?

Not necessarily. AI assistants can speed up boilerplate creation, but you need the safeguards described above. Treat them as a productivity tool, not a legal shield.

Key Takeaways

ai generated code cannot be copyrighted in the US, EU, UK, Canada, and Australia unless a human adds enough original expression.
Log every prompt, model version, and timestamp; store the log with the generated file.
Run a license detection step in CI and assign a permissive license (MIT, Apache-2.0) to original snippets.
Isolate AI-generated files, review them for security and quality, and enforce a policy via branch protection.
Open-source projects must audit contributions for AI-generated code; commercial projects should plan for possible hot-fixes.

If you’ve hit a wall trying to untangle licensing or security issues from AI-generated snippets, I can help you set up the right workflow and get your FastAPI services back on track. Feel free to reach out through the hire page for hands-on assistance.

FAQ

Q: Can I use AI-generated code in a proprietary product?

A: Yes, but you must verify the snippet isn’t a verbatim copy of copyrighted material and assign your own license. Treat it like any third-party library.

Q: Does the MIT license protect me from future claims by the model provider?

A: MIT only covers the rights you grant to downstream users. It does not shield you from claims that the model provider owns the output. Keep the provenance log to defend your position.

Q: How do I detect if an AI snippet matches existing open-source code?

A: Use tools like GitHub’s code scanning or FOSSology in your CI pipeline. They can flag high-similarity matches and point you to the original source.

Q: What if I accidentally ship a GPL-licensed snippet generated by an AI?

A: Replace the snippet with a clean implementation, update your license files, and, if necessary, notify downstream users of the change.

If you’re ready to tighten up your AI-code workflow, check out the related posts on Fixing an AI Generated Code Vulnerabilities Report and the AI Generated Code Detection: Practical Guide for Builders. They dive deeper into scanning and remediation strategies.

Bolt.new Python Backend: FastAPI from Zero to Production

Ayush Kumar — Wed, 02 Sep 2026 10:28:08 +0000

If you need a bolt new python backend that’s up and running in minutes, you can get a FastAPI service with a single command and a sensible folder layout. It works for prototypes, but it also scales to production when you add async DB support, proper logging, and a CI pipeline. Below I’ll walk through exactly how I did it, the hiccups that tripped me up, and when you might want to stick with a hand-crafted FastAPI project instead.

What is Bolt.new and how does it simplify backend development?

Bolt.new is a CLI that generates a ready-to-run FastAPI skeleton. It chooses sensible defaults: Pydantic models for request validation, Uvicorn as the ASGI server, and a docker-compose.yml that spins up a PostgreSQL container. The generated code follows the “router-service-repository” pattern, so you can drop in business logic without fighting the structure.

The biggest win is the speed of scaffolding. Where a vanilla FastAPI project might take you an hour to set up routing, dependency injection, and a Dockerfile, Bolt.new does it in under a minute:

pip install bolt-new
bolt new myapp --framework fastapi
cd myapp
docker compose up -d
uvicorn app.main:app --reload

You get a working API at http://localhost:8000/health within seconds. That’s the answer to “how do I get a bolt new python backend running?” – run the CLI, and you have a functional service.

Step-by-step: scaffolding a FastAPI project with Bolt.new

Install the CLI – pip install bolt-new (requires Python 3.9+).
Create the project – bolt new ai-service --framework fastapi.
- The command creates ai_service/ with app/, tests/, Dockerfile, docker-compose.yml, and a pyproject.toml.
Review the generated router – app/api/v1/health.py contains:

from fastapi import APIRouter

router = APIRouter()

@router.get("/health")
async def health_check() -> dict:
    return {"status": "ok"}

Mount the router – app/main.py already includes:

from fastapi import FastAPI
from .api.v1 import health

app = FastAPI(title="Bolt.new Demo")
app.include_router(health.router, prefix="/v1")

Run locally – docker compose up -d starts PostgreSQL, then uvicorn app.main:app --reload.
Test – curl http://localhost:8000/v1/health returns {"status":"ok"}.

That’s it. The scaffold gives you a clean entry point, a dev-ready Docker setup, and a test suite stubbed with pytest. If you need a quick AI endpoint, just drop a new router file under app/api/v1/ and import it in main.py.

Adding async database support (SQLAlchemy/Prisma) in a Bolt.new backend

Bolt.new ships with a synchronous SQLAlchemy example, but production workloads need async I/O. Here’s how I swapped in SQLModel (built on async SQLAlchemy) and why I considered Prisma as an alternative.

1. Install async dependencies

poetry add sqlmodel[async] asyncpg

2. Create the async engine

app/db.py

from sqlmodel import SQLModel, create_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import AsyncEngine

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/bolt_db"

engine: AsyncEngine = create_engine(DATABASE_URL, echo=False, future=True)

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

3. Define a model

app/models/item.py

from sqlmodel import Field, SQLModel

class Item(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    name: str
    description: str | None = None

4. Add CRUD routes

app/api/v1/items.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel.ext.asyncio.session import AsyncSession
from ..db import get_session
from ..models.item import Item

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

@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item, session: AsyncSession = Depends(get_session)):
    session.add(item)
    await session.commit()
    await session.refresh(item)
    return item

@router.get("/{item_id}", response_model=Item)
async def read_item(item_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.get(Item, item_id)
    if not result:
        raise HTTPException(status_code=404, detail="Item not found")
    return result

5. Register the router

In app/main.py add:

from .api.v1 import items
app.include_router(items.router, prefix="/v1")

6. Run migrations

Bolt.new doesn’t include Alembic out of the box, so I added:

poetry add alembic
alembic init alembic

Configure alembic/env.py to use the async engine and run alembic revision --autogenerate -m "create items" followed by alembic upgrade head.

Why not Prisma?

Prisma’s Python client is still experimental, and the generated async client adds another layer of code generation. For a solo builder who wants stability, async SQLAlchemy (or SQLModel) is the safer bet. If you already use Prisma in a Node service and love its type-safety, you can spin up a separate microservice and talk over HTTP, but that defeats the “single bolt new python backend” simplicity.

Deploying the Bolt.new FastAPI service to cloud platforms (Railway, Render, Fly.io)

Once the API works locally, the next question is “where do I host it without blowing the budget?” I’ve tried three cheap options. All of them accept a Docker image, so the Dockerfile generated by Bolt.new works unchanged.

Railway

Connect the repo, enable “Dockerfile” detection.
Set environment variable DATABASE_URL to the Railway-provided Postgres URL.
Deploy – Railway builds the image, runs it, and gives you a public URL.

Gotchas: Railway’s free tier sleeps after 30 minutes of inactivity. If your AI endpoint is latency-sensitive, you’ll see a cold-start delay (2-3 seconds). The workaround is to add a tiny “keep-alive” cron that pings /health every 5 minutes.

Render

Render’s free web services also sleep, but you can upgrade to the “Starter” plan for $7/mo and get always-on containers. The platform automatically provisions a Postgres instance and injects DATABASE_URL. Deploy steps are identical to Railway.

Pitfall: Render expects the container to listen on the $PORT env var. The default Dockerfile from Bolt.new uses EXPOSE 8000. Change the entrypoint:

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "${PORT}"]

Fly.io

Fly.io is great for global latency. You define a fly.toml:

app = "bolt-fastapi"
kill_signal = "SIGINT"
kill_timeout = 5

[env]
  DATABASE_URL = "postgres://..."

[[services]]
  internal_port = 8000
  protocol = "tcp"
  [[services.ports]]
    handlers = ["http"]
    port = 80

Deploy with fly deploy. Fly keeps the container warm, so cold starts are rare.

Cost: The free tier gives you 3 GB-hours per month, enough for a low-traffic prototype. Once you exceed that, you pay $5/mo per 256 MB of RAM. For a single FastAPI instance, $5–$7/mo is typical.

Common pitfalls and performance tuning tips for production-grade Bolt.new apps

1. MissingGreenlet error

When I first added async SQLAlchemy, the app crashed with:

sqlalchemy.exc.MissingGreenletError: Greenlet is required for async support

The fix is to install greenlet explicitly:

poetry add greenlet

Bolt.new’s generated requirements.txt doesn’t include it because the default scaffold uses sync SQLAlchemy. Always double-check async dependencies.

2. Database connection pool exhaustion

FastAPI spawns many coroutines; each request can open a new DB connection if you’re not careful. In app/db.py set a pool size:

engine = create_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    echo=False,
)

Monitor the pool with SELECT * FROM pg_stat_activity; on your Postgres instance.

3. Logging overhead

The default logger from Bolt.new prints every request line, which hurts performance under load. Switch to uvicorn[standard] and configure a JSON logger:

import logging
logging.basicConfig(level=logging.INFO, format='%(message)s')

Send logs to a service like Logtail or Papertrail for easier debugging.

4. Cold starts on serverless platforms

If you deploy to Cloud Run or a serverless variant of Railway, the first request after a period of inactivity can take 2–5 seconds. Mitigate by:

Keeping the container warm with a health-check ping.
Reducing the container size to 256 MiB (cold starts are faster on smaller images).
Pre-warming the model if you’re loading a large ML artifact; load it at import time, not per request.

5. Secret management

Never hard-code API keys in the repo. Bolt.new’s .env.example shows the pattern, but you must add the real values to the platform’s secret store. Missing a secret leads to a 500 error that’s hard to trace.

Bolt.new vs a traditional FastAPI setup: pros, cons, and cost considerations

Aspect	Bolt.new	Hand-crafted FastAPI
Speed of initial setup	Seconds with a CLI	Hours to configure Docker, linting, CI
Learning curve	Low – opinionated defaults	Higher – you choose every library
Flexibility	Moderate – you can replace parts, but the scaffold expects certain paths	Unlimited – you decide project layout
Production readiness	Good for small-to-medium services; you still need to add logging, migrations, monitoring	Can be built with enterprise-grade patterns from the start
Cost	Free CLI, same hosting costs as any Docker app	Same hosting costs; no extra licensing
When to avoid	You need custom auth flow, complex event-driven architecture, or multi-service monorepo	You already have a mature CI/CD pipeline and want full control

In practice, I start with Bolt.new for a new AI microservice because it lets me ship a prototype in a day. When the traffic grows beyond a few hundred RPS, I audit the generated code, replace the router with a more modular approach, and add OpenTelemetry. The transition is painless because the scaffold follows standard FastAPI conventions.

FAQ

How do I add CORS support to a bolt new python backend?

Add from fastapi.middleware.cors import CORSMiddleware in app/main.py and configure:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Can I use Bolt.new with a non-Postgres database?

Yes. Change DATABASE_URL in the .env file to point to MySQL, SQLite, or any SQLAlchemy-compatible DB, and adjust the driver in poetry add. The scaffold doesn’t lock you to Postgres.

What’s the best way to run background tasks (e.g., model inference) in this setup?

FastAPI’s BackgroundTasks works for quick fire-and-forget jobs. For longer jobs, spin up a Celery worker or use a managed queue like Temporal. Keep the worker image similar to the API image so you can reuse the same code base.

Is there a way to generate OpenAPI docs automatically?

FastAPI already serves Swagger UI at /docs and ReDoc at /redoc. Bolt.new doesn’t hide this; just hit those endpoints after deployment.

Key Takeaways

Bolt.new gives you a working FastAPI service in seconds – run the CLI, Docker compose, and you’re live.
Add async DB support with SQLModel (or async SQLAlchemy) and remember to install greenlet.
Deploy to Railway, Render, or Fly.io using the generated Dockerfile; watch for environment-variable port handling and cold-start sleeps.
Common failures include missing greenlet, exhausted DB pools, and secret leakage – fix them early.
Weigh trade-offs: Bolt.new accelerates prototyping, but you may need to refactor for large-scale production requirements.

If you’ve hit a wall turning a bolt new python backend into a reliable service, I’m happy to roll up my sleeves and help you get it into production. Feel free to check out my hire page for a hands-on engagement.

Microservices Python Django: Practical Setup & Scaling

Ayush Kumar — Wed, 02 Sep 2026 10:27:51 +0000

If you’re looking for a straightforward way to run microservices python django in production, the answer is: treat each Django app as a small, self-contained service, expose its API with Django Rest Framework, and let Docker plus Kubernetes handle the rest. Below I walk through the choices, the wiring, and the pitfalls I’ve hit on the last two years of shipping AI-enabled data pipelines on this stack.

Why choose Django for microservices?

The first question most engineers ask is whether Django is even a good fit for a microservice architecture. The short answer: yes, when you need a full-stack framework with batteries included and a mature ORM. Django gives you admin, authentication, migrations, and a robust request/response lifecycle out of the box, which cuts down on third-party glue code.

That said, Django was originally built for monoliths, so you pay a small price in start-up latency and memory footprint. In a service-per-container world that cost is usually negligible compared with the operational benefits of a single, well-known codebase. If you’re chasing ultra-low latency or need a tiny footprint, a lightweight framework like FastAPI might win, but for most data-heavy back-ends Django’s ergonomics win.

How do I set up a Django project as a microservice?

Start with the standard django-admin startproject command, but treat the generated project as a single service. Keep the settings modular: one base file, plus dev.py, prod.py, and test.py. Pull secret values from environment variables – never hard-code them.

# Create the service
django-admin startproject orderservice
cd orderservice

# Create a clean virtualenv
python -m venv .venv
source .venv/bin/activate
pip install django djangorestframework gunicorn psycopg2-binary

orderservice/settings/base.py (excerpt):

import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

SECRET_KEY = os.getenv('DJANGO_SECRET_KEY')
DEBUG = os.getenv('DJANGO_DEBUG', 'False') == 'True'

ALLOWED_HOSTS = os.getenv('DJANGO_ALLOWED_HOSTS', '*').split(',')

INSTALLED_APPS = [
    'django.contrib.contenttypes',
    'django.contrib.auth',
    'rest_framework',
    'orders',               # our app
]

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': os.getenv('POSTGRES_DB'),
        'USER': os.getenv('POSTGRES_USER'),
        'PASSWORD': os.getenv('POSTGRES_PASSWORD'),
        'HOST': os.getenv('POSTGRES_HOST', 'db'),
        'PORT': os.getenv('POSTGRES_PORT', 5432),
    }
}

Create an app (python manage.py startapp orders) and keep it focused on a single domain (order processing). The rest of the project stays lean – no templates, no static files unless you truly need them.

How do I build RESTful APIs with Django Rest Framework?

DRF is the natural companion for a Django microservice. Define serializers that map directly to your models, and use viewsets to get CRUD endpoints with a single line.

orders/models.py:

from django.db import models

class Order(models.Model):
    customer_id = models.UUIDField()
    product_id = models.UUIDField()
    quantity = models.PositiveIntegerField()
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = 'orders'

orders/serializers.py:

from rest_framework import serializers
from .models import Order

class OrderSerializer(serializers.ModelSerializer):
    class Meta:
        model = Order
        fields = '__all__'

orders/views.py:

from rest_framework import viewsets
from .models import Order
from .serializers import OrderSerializer

class OrderViewSet(viewsets.ModelViewSet):
    queryset = Order.objects.all()
    serializer_class = OrderSerializer

Wire it up in orderservice/urls.py:

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from orders.views import OrderViewSet

router = DefaultRouter()
router.register(r'orders', OrderViewSet)

urlpatterns = [
    path('api/', include(router.urls)),
]

That’s it. You now have GET /api/orders/, POST /api/orders/, etc., all with proper validation and OpenAPI schema generation (/api/schema/). In production I swap the default runserver for Gunicorn behind Uvicorn workers when I need async support, but the synchronous stack is more than enough for most CRUD workloads.

What inter-service communication patterns work best with Django?

HTTP/REST

The simplest pattern is HTTP. Each Django service publishes a clean REST endpoint. Use a shared Python client library (generated from OpenAPI) to keep contracts stable. The downside is higher latency and tighter coupling; a change in one service’s URL or auth scheme can break the whole chain.

gRPC

If you need low latency and strong typing, gRPC works, but Django doesn’t have first-class support. I usually spin up a thin grpc-gateway container that forwards gRPC calls to the Django HTTP layer. This adds operational complexity, so I only use it for high-throughput pipelines (e.g., streaming telemetry).

Message queues

For eventual consistency and decoupling, push events to a broker (RabbitMQ, Kafka, or AWS SQS). Django can publish messages via Celery or a simple producer wrapper. Consumers can be other Django services, FastAPI workers, or even serverless functions. I’ve been bitten by “message ordering” bugs when mixing Kafka partitions with Django’s auto-commit transaction handling, so I always enable idempotent processing and store the last offset in the DB.

Here’s a minimal Celery task that publishes an order_created event:

# orders/tasks.py
from celery import shared_task
import json
import os
import pika  # simple RabbitMQ client

@shared_task
def publish_order_created(order_id):
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host=os.getenv('RABBIT_HOST', 'rabbit'))
    )
    channel = connection.channel()
    channel.exchange_declare(exchange='orders', exchange_type='fanout')
    payload = json.dumps({'order_id': str(order_id)})
    channel.basic_publish(exchange='orders', routing_key='', body=payload)
    connection.close()

Trigger it in the OrderViewSet perform_create hook.

How do I containerize and orchestrate Django microservices with Docker and Kubernetes?

Dockerfile (multi-stage for speed):

# Build stage
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt

# Runtime stage
FROM python:3.11-slim
ENV PYTHONUNBUFFERED=1
WORKDIR /app
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH
COPY . .
RUN python manage.py collectstatic --noinput

EXPOSE 8000
CMD ["gunicorn", "orderservice.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "4"]

Kubernetes deployment (simplified):

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: order-service
  template:
    metadata:
      labels:
        app: order-service
    spec:
      containers:
        - name: django
          image: ghcr.io/yourorg/order-service:latest
          ports:
            - containerPort: 8000
          envFrom:
            - secretRef:
                name: order-service-secrets
          readinessProbe:
            httpGet:
              path: /health/
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: order-service
spec:
  selector:
    app: order-service
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000

A couple of things that bite me:

Image size – the multi-stage build drops the build-time packages, but I still saw >200 MB images because of psycopg2-binary. Switching to the slim psycopg2 wheels saved ~30 MB.
Health checks – Django’s default /admin/ page is heavy; expose a lightweight /health/ view that only checks DB connectivity.
Resource limits – start with 200 Mi CPU and 256 Mi RAM, then adjust after monitoring. Over-provisioning kills cost efficiency on cloud spot instances.

How do I deploy and scale Django microservices in production?

CI/CD pipeline – I use GitHub Actions to build the Docker image, run unit tests, and push to a private registry. The docker/build-push-action with --cache-from speeds up subsequent builds.
Helm charts – parameterize replica count, env vars, and resource limits. Keep the chart in the same repo as the service for version alignment.
Observability – instrument Django with OpenTelemetry and ship traces to Jaeger. For logs, forward JSON-formatted logs to Loki; for metrics, enable django-prometheus.
Autoscaling – Horizontal Pod Autoscaler (HPA) based on CPU and custom metric http_requests_total. I’ve seen spikes where a single order burst saturates a pod; HPA reacts in ~30 seconds, which is acceptable for my SLA.
Database migrations – run them as a one-off Kubernetes Job before rolling the new pods. Never let a pod start before the migration is complete; otherwise you get ProgrammingError: relation does not exist.

When you need to evolve a service, remember the single responsibility principle: if a Django app starts handling authentication, email, and reporting, it’s time to split it into separate microservices. Over-splitting is also a risk – each service adds network hop and operational overhead. The sweet spot is “one domain per service, one database per service, and a shared message bus for async events.”

When NOT to use Django for microservices

Ultra-low latency – if you need sub-millisecond response times, the Python runtime and Django’s middleware stack add measurable overhead.
Stateless, compute-heavy functions – for pure inference or data-transform pipelines, a serverless or FastAPI approach may be cheaper.
Highly dynamic schemas – Django’s ORM expects a relatively stable schema; rapid schema churn can cause migration lock contention.

If you find yourself in those scenarios, consider mixing stacks: keep data-centric services in Django, and spin up FastAPI or plain ASGI workers for the hot path. I wrote about that hybrid approach in my post on Building microservices python fastapi: design to production.

FAQ

What database does Django microservice use in production?

PostgreSQL is the default choice because of its strong ACID guarantees, native JSON support, and excellent Django integration. Use a managed cloud instance (RDS, CloudSQL) for HA and automated backups.

Can I share the same Django settings module across multiple services?

Yes, keep a common settings/base.py and import it in each service’s dev.py / prod.py. Override only what differs (e.g., ALLOWED_HOSTS, DATABASES). Avoid importing service-specific apps into the base file.

How do I handle secret rotation without redeploying?

Leverage Kubernetes Secrets or HashiCorp Vault. Mount secrets as files or expose them as env vars, and configure Django to read them at runtime. For rotating DB passwords, use a sidecar that updates the env var and triggers a graceful pod restart.

Is Celery mandatory for async work in Django microservices?

Not mandatory, but highly recommended for background jobs and message-driven workflows. If you only need occasional async endpoints, Django 4’s native async views can call await-compatible libraries, but they won’t give you retries or scheduling.

Key Takeaways

Django provides a mature, batteries-included foundation for microservices python django when you need admin, ORM, and rapid development.
Keep each service small, modular, and backed by its own database; expose a clean DRF API.
Choose the right communication pattern: HTTP for simplicity, gRPC for performance, message queues for decoupling.
Docker multi-stage builds and Kubernetes Helm charts make deployment repeatable and scalable.
Monitor health, logs, and traces; autoscale with HPA; run migrations as a pre-deploy job.
Avoid Django for ultra-low-latency or highly dynamic schema workloads; mix in FastAPI or serverless where it makes sense.

By treating Django as a first-class citizen in a microservice world, you get the best of both worlds: developer velocity and production-grade reliability. Happy coding!

How to Fix Claude Code Error 529 in Production

Ayush Kumar — Wed, 02 Sep 2026 10:18:48 +0000

Short answer: Claude code error 529 means the service rejected your request because it hit an internal throttling limit. The fix is to back off, retry with jitter, and make sure you respect the per-minute quota that Claude enforces.

If you’re seeing a JSON payload like {"error":"code_error","code":529,"message":"Too many concurrent code generations"}, you’re already at the point where the API is telling you to slow down. Below I walk through why this shows up, how to debug it, and what to put in place so it stops breaking your production pipeline.

Understanding the 529 error response from Claude API

Claude returns a 529 when it cannot schedule a code-generation job because its internal queue is full. It’s not a “bad request” or a syntax problem; it’s a resource-exhaustion signal. The response looks like:

{
  "error": "code_error",
  "code": 529,
  "message": "Too many concurrent code generations"
}

The API docs call it “temporary throttling.” In practice it means you’ve either:

Sent more concurrent POST /v1/complete calls than Claude allows, or
Hit a burst of requests that exceeds the per-minute token budget.

Claude’s back-end will eventually free capacity, but if you keep hammering it you’ll see the same 529 over and over. The cost is two-fold: wasted compute credits and a user-visible slowdown that feels like a bug.

What causes Claude code error 529 in code generation?

The most common root causes are:

Cause	Why it triggers 529	Typical symptom
Unbounded parallelism	Each request occupies a slot in Claude’s internal pool. Fire off dozens from an async loop and you saturate it instantly.	Errors appear after the first few successful generations.
Missing back-off	Retries without jitter cause a thundering herd.	Same request fails repeatedly even after a short sleep.
Large prompt + high token limit	Claude counts tokens in the prompt and the max tokens you request. A big prompt pushes you over the per-minute token ceiling.	Errors show up only on “big” prompts, not on tiny ones.
Mis-configured client library	Using `httpx.AsyncClient` with `limit=0` removes the built-in connection pool limits, flooding the API.	Errors appear under load testing but not in dev.

I’ve been bitten by all of these. The first time I ran a Celery worker that spawned 50 async tasks per second, Claude started returning 529 within minutes. The logs filled up, and my users saw “code generation failed” messages.

How can I debug and retry Claude 529 errors?

The first thing to do is capture the error and stop the retry loop before it spirals. Here’s a minimal async wrapper that respects Claude’s limits:

import asyncio
import httpx
import random
import time

CLAUDE_URL = "https://api.anthropic.com/v1/complete"
API_KEY = "sk-..."

# Simple exponential back‑off with jitter
async def call_claude(payload: dict, max_retries: int = 5) -> dict:
    async with httpx.AsyncClient(timeout=30) as client:
        for attempt in range(max_retries):
            resp = await client.post(
                CLAUDE_URL,
                json=payload,
                headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
            )
            if resp.status_code == 200:
                return resp.json()
            if resp.status_code == 529:
                # log and back off
                wait = (2 ** attempt) + random.uniform(0, 0.5)
                print(f"Claude 529 – backing off for {wait:.2f}s (attempt {attempt+1})")
                await asyncio.sleep(wait)
                continue
            # other errors – raise immediately
            resp.raise_for_status()
        raise RuntimeError("Exceeded max retries for Claude code generation")

Key points:

Check resp.status_code == 529 before trying to parse JSON.
Exponential back-off (2 ** attempt) spreads retries over time.
Jitter (random.uniform) prevents many workers from lining up on the same second.
Limit retries – after 5 attempts give up and surface a clear error to the caller.

If you prefer a sync version for a FastAPI endpoint, replace httpx.AsyncClient with httpx.Client and drop the awaits. The logic stays the same.

How do I manage rate limits and back-off with Claude?

Claude doesn’t publish a hard “X requests per minute” number, but you can infer it by monitoring 529 frequency. A practical pattern is to track tokens used per minute in a Redis bucket:

import redis
import time

r = redis.Redis(host="localhost", port=6379, db=0)

TOKEN_BUCKET = "claude:tokens:minute"
MAX_TOKENS_PER_MIN = 120_000  # example budget

def can_send(tokens: int) -> bool:
    now = int(time.time() // 60)  # current minute bucket
    key = f"{TOKEN_BUCKET}:{now}"
    used = r.incrby(key, tokens)
    r.expire(key, 70)  # keep bucket a bit longer than a minute
    return used <= MAX_TOKENS_PER_MIN

Before calling call_claude, estimate the token count (len(prompt.split()) + max_tokens) and abort early if you’re over budget. Combine this guard with the back-off wrapper above and you’ll rarely see a 529.

If you already have a task queue like Celery, you can also set a concurrency limit on the worker that talks to Claude:

celery -A myapp worker -Q claude_queue -c 4

Four concurrent workers is a safe starting point for most free-tier accounts. Adjust upward only after you verify the 529 rate stays near zero.

What best practices stop future 529 errors?

Cap parallel calls – never let a user-triggered request spawn more than a handful of Claude calls. Batch work when possible.
Use a shared token budget – store per-minute usage in Redis or a DB and gate new requests.
Prefer smaller prompts – strip comments, reuse snippets, and keep the prompt under 1 000 tokens unless you really need more context.
Graceful degradation – if you hit the limit, fall back to a cached answer or a simpler heuristic instead of looping forever.
Instrument everything – record response.status_code, request size, and latency. Those metrics make it easy to spot a rising 529 trend before it hurts users.

I once tried to “just increase the pool size” in my FastAPI app, thinking the error was a client-side socket limit. It wasn’t. The real fix was to add the token bucket guard and reduce parallelism. After that, the 529 vanished and my error-rate dropped from 12 % to <0.2 %.

If you’re already using SQLAlchemy with async sessions, you might wonder whether the retry logic interferes with DB transactions. It doesn’t, as long as you keep the DB call outside the Claude wrapper. See my post on Fixing SQLAlchemy MissingGreenlet Error in FastAPI (Async Explained) for details on keeping async DB work tidy.

How should I monitor and log Claude interactions for reliability?

Treat Claude like any other external dependency:

What to log	Why it matters
Request payload size (tokens)	Correlates with throttling
Response status (200, 529, 500)	Shows health trends
Latency (ms)	Detects degradation before errors
Retry count	Helps tune back-off parameters

A simple FastAPI middleware can add this automatically:

from fastapi import Request, FastAPI
import time, logging

app = FastAPI()
log = logging.getLogger("claude")

@app.middleware("http")
async def claude_logging(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = (time.time() - start) * 1000
    log.info(
        "path=%s method=%s status=%s duration_ms=%.1f",
        request.url.path,
        request.method,
        response.status_code,
        duration,
    )
    return response

Push those logs to a structured system (Elastic, Loki, or CloudWatch) and set an alert on a spike of status=529. A sudden rise often means you’ve introduced a new feature that fires more Claude calls than before.

When to avoid the retry-back-off pattern

Retrying is great for transient throttling, but not for:

Invalid prompts – Claude will return a 400 error, not a 529. Retrying wastes time.
User-canceled requests – If the client disconnects, abort the retry loop early.
Critical path where latency matters – If you need a response under 500 ms, give up after the first attempt and fall back to a cached result.

In those cases, surface a clear error message to the user and let them decide whether to try again.

If you’ve tried the suggestions above and still see Claude code error 529 popping up in production, it might be a deeper quota issue or a bug in your orchestration layer. I’ve helped indie teams untangle these problems end-to-end. Feel free to reach out through my hire page for a hands-on session.

FAQ

Q: Does Claude return 529 for rate-limit errors only?

A: Mostly. 529 signals that Claude’s internal queue is full, which usually stems from exceeding the per-minute token or request limit.

Q: Can I increase my quota to eliminate 529?

A: You can request a higher limit from Anthropic, but the safer route is to respect the existing limits with back-off and token budgeting.

Q: Will adding more workers to my Celery queue fix the problem?

A: No. More workers just increase concurrent calls, which makes 529 more likely. Scale down or add a rate-limit guard instead.

Q: Is 529 the same as HTTP 429?

A: Conceptually similar - both mean “too many requests” - but Claude uses 529 for internal throttling while 429 is the standard HTTP rate-limit code.

Key Takeaways

Claude code error 529 means the service is throttling you, not that your code is wrong.
Guard concurrent calls, track per-minute token usage, and back off with jitter.
Use a small retry wrapper (see the example) and stop after a few attempts.
Log status, latency, and token count; alert on spikes of 529.
Apply the same disciplined rate-limit handling you’d use for any external API.

By treating Claude like any other rate-limited dependency, you keep your AI-generated code pipeline reliable and your users happy.

Using Cursor AI Code Tracking API with FastAPI

Ayush Kumar — Tue, 25 Aug 2026 06:55:11 +0000

If you need to capture every edit that Cursor AI makes to your codebase, the Cursor AI Code Tracking API is the tool you’re looking for. In a few minutes you can authenticate, pull change events, persist them, and build a simple dashboard – all inside a FastAPI service you already run in production.

Below is a step-by-step walkthrough that shows exactly how to do it, where things tend to break, and what you should keep an eye on. I’ve stripped out the fluff and kept the focus on things that actually matter when you’re shipping a product.

Overview of Cursor AI Code Tracking API

What does the Cursor AI Code Tracking API give you? It streams JSON payloads that describe each file change, the prompt that triggered it, and a timestamp. The endpoint is GET https://api.cursor.com/v1/tracking. It returns a paginated list of events, each looking like:

{
  "id": "evt_12345",
  "file_path": "app/main.py",
  "change_type": "modification",
  "prompt": "Refactor the request handler",
  "diff": "-old line\n+new line",
  "created_at": "2024-08-14T12:34:56Z"
}

That’s all you need to reconstruct a timeline of AI-generated edits. The API is low-latency, but it enforces rate limits (30 calls per minute) and requires a valid API key for every request.

Authentication and API key setup

How do you securely store and use the Cursor API key? The service expects the key in an Authorization: Bearer <token> header. In production I keep the token in an environment variable called CURSOR_API_KEY. Never hard-code it.

import os
from fastapi import Depends, HTTPException, status

def get_cursor_token() -> str:
    token = os.getenv("CURSOR_API_KEY")
    if not token:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Cursor API key not configured"
        )
    return token

If the key is missing, the service fails fast with a 500 error – that’s intentional. It prevents you from silently sending unauthenticated requests and hitting obscure 401 responses later in the pipeline.

Common failure: forgetting to add the env var to your Docker or Kubernetes secret. The result is a cascade of 401 errors that look like “invalid credentials” but actually stem from a missing secret. Double-check your deployment manifest.

Integrating the API with a FastAPI service

What does the integration code look like? Below is a minimal FastAPI router that pulls the latest 100 events and returns them as a list.

import httpx
from fastapi import APIRouter, Depends

router = APIRouter()

CURSOR_ENDPOINT = "https://api.cursor.com/v1/tracking"

@router.get("/cursor/events")
async def get_cursor_events(token: str = Depends(get_cursor_token)):
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(
            CURSOR_ENDPOINT,
            headers={"Authorization": f"Bearer {token}"},
            params={"limit": 100}
        )
        if resp.status_code != 200:
            # Propagate the exact error for debugging
            raise HTTPException(
                status_code=resp.status_code,
                detail=f"Cursor API error: {resp.text}"
            )
        return resp.json()["events"]

A few things to note:

Async client – FastAPI runs on ASGI, so use httpx.AsyncClient to avoid blocking the event loop.
Timeout – set a reasonable timeout (10 seconds here) to prevent hanging workers.
Error propagation – surface the exact error; otherwise you’ll see generic 500s that make troubleshooting harder.

Trade-off: pulling 100 events on every request is cheap for a dev dashboard but could become expensive at scale. If you need real-time updates, consider a webhook or a background poller instead of on-demand fetching.

Persisting tracking data in PostgreSQL or Redis

Where should you store the events? Two common patterns:

PostgreSQL (relational, query-heavy)

Create a table that mirrors the JSON payload:

CREATE TABLE cursor_events (
    id TEXT PRIMARY KEY,
    file_path TEXT NOT NULL,
    change_type TEXT NOT NULL,
    prompt TEXT,
    diff TEXT,
    created_at TIMESTAMPTZ NOT NULL
);

Insert with async SQLAlchemy:

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import insert
from models import CursorEvent  # SQLAlchemy model matching the table

async def save_events(events: list[dict], db: AsyncSession):
    stmt = insert(CursorEvent).values(events)
    await db.execute(stmt)
    await db.commit()

Failure mode: duplicate id values raise an integrity error. Wrap the insert in a try/except block and ignore duplicates if you’re polling the same page repeatedly.

Redis (fast, volatile, analytics)

If you only need recent changes for a dashboard, a Redis stream works well:

import aioredis

redis = await aioredis.from_url("redis://localhost:6379")

async def push_to_redis(events: list[dict]):
    for ev in events:
        await redis.xadd("cursor:events", ev)

Redis gives you O(1) writes and easy range queries (XRANGE). The downside is data loss on restart unless you enable persistence, and you lose relational querying capabilities.

When NOT to use Redis: when you need to join events with other domain tables (e.g., linking a change to a ticket). In that case PostgreSQL is the safer bet.

Building a dashboard to visualize code changes

How can you turn stored events into a useful UI? I built a tiny Vue front-end that calls a FastAPI endpoint /dashboard/events. The endpoint reads from PostgreSQL and returns a paginated list with optional filters.

from fastapi import Query

@router.get("/dashboard/events")
async def dashboard_events(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, le=100),
    file_path: str | None = None,
    db: AsyncSession = Depends(get_db)
):
    query = select(CursorEvent).order_by(CursorEvent.created_at.desc())
    if file_path:
        query = query.where(CursorEvent.file_path == file_path)
    result = await db.execute(query.offset(skip).limit(limit))
    return result.scalars().all()

The UI shows a table with diff previews, timestamps, and the original prompt. Clicking a row expands the diff. I added a simple heat-map that counts changes per file; the query is a GROUP BY file_path aggregation.

Pitfall: pulling the full diff for every row can bloat the response. Send only a snippet (first 200 characters) and let the UI request the full diff on demand via another endpoint.

Security and best-practice considerations

What should you watch out for when exposing AI-generated code data? A few hard-earned lessons:

Never expose raw diff data to unauthenticated users. Diffs may contain secrets that the AI unintentionally inserted (API keys, passwords). Always scrub before rendering. A quick regex replace works for most cases, but consider a dedicated secret-detection step – see my AI generated code platform security checklist.
Rate-limit your own endpoint to avoid cascading failures if Cursor imposes stricter limits. A simple slowapi decorator can throttle calls to /cursor/events.
Validate the schema of incoming JSON. Cursor can evolve its payload format; using Pydantic models catches mismatches early:

from pydantic import BaseModel, Field

class CursorEventModel(BaseModel):
    id: str
    file_path: str
    change_type: str
    prompt: str | None
    diff: str | None
    created_at: datetime

Log responsibly. Store only event IDs and timestamps in your audit log. Full diffs belong in a secure data store, not in plain-text logs that may be shipped to external services.

If you’re already wrestling with code-quality regressions caused by AI, check out my post on Fixing AI Generated Code Quality Issues in Production for additional context.

FAQ

How often can I poll the Cursor API?

The official limit is 30 requests per minute. Exceeding it returns a 429 response. Use exponential back-off or a background worker that respects the limit.

Can I use the API without FastAPI?

Absolutely. The endpoint is language-agnostic; any HTTP client can call it. FastAPI just makes integration painless for Python services.

What happens if an event’s diff contains a secret?

Treat it as a security incident. Scrub the diff before persisting, rotate the leaked secret, and consider adding the event to a secret-detection pipeline.

Is there a webhook alternative?

Cursor currently only offers a pull-based API. If you need push notifications you’ll have to implement a poller that writes to a message queue (e.g., RabbitMQ) and lets downstream services consume the events.

Key Takeaways

The Cursor AI Code Tracking API provides JSON change events; authenticate with a bearer token stored in an env var.
Use httpx.AsyncClient inside FastAPI to fetch events without blocking.
Persist events in PostgreSQL for relational queries or Redis streams for fast, temporary analytics.
Build a lightweight dashboard that filters and paginates results; avoid sending full diffs unless requested.
Scrub secrets, enforce rate limits, and validate payloads to keep the system secure.

If you’ve hit a wall trying to stitch these pieces together, I’m happy to help you get a production-ready pipeline up and running. Feel free to reach out via the hire page and we can tackle the rough spots together.

DEV Community: Ayush Kumar

Fix AI Generated Code: Practical Steps for Backend Engineers

What are the most common vulnerabilities in AI generated code?

How do I detect logic errors in AI generated Python?

How do I fix security flaws in FastAPI endpoints from AI output?

How do I validate AI-generated SQLAlchemy models?

How do I test AI-generated API contracts with Pytest?

How do I refactor AI-generated async code for production?

When NOT to trust AI generated code

FAQ

How do I know if my AI generated code has SQL injection?

Can I use AI to generate my FastAPI auth system?

What’s the fastest way to validate AI generated Python?

Should I use AI to generate database migrations?

Key Takeaways

Postgres Performance Tuning Best Practices for Backend Engineers

What are the most common Postgres performance bottlenecks?

Which configuration parameters actually matter for tuning?

How do I monitor Postgres performance in production?

What indexing strategies actually optimize queries?

How should I handle connection pooling and worker scaling?

When should I tune vacuum and autovacuum for write-heavy workloads?

FAQ

Key Takeaways

RAG Chunking Best Practices for Production Systems

How do I choose the optimal chunk size for LLM embeddings?

What overlap strategies preserve context in RAG without killing efficiency?

How should I chunk tabular and structured data for RAG?

How do I evaluate chunking impact on retrieval accuracy and latency?

What tools and libraries help automate chunking in Python?

Balancing semantic coherence with computational efficiency

FAQ

Key Takeaways

Python Message Queue Between Processes: IPC Options for FastAPI Workers

How do I implement inter-process communication with multiprocessing.Queue?

What’s the difference between multiprocessing.Queue, pipes, and shared memory?

How do I handle serialization of complex objects in process queues?

How do I avoid deadlocks and race conditions in multiprocessing queues?

What are the performance benchmarks of different IPC mechanisms in Python?

How do I integrate message queues with FastAPI background workers?

When NOT to use multiprocessing.Queue

FAQ

Key Takeaways

Redis caching in FastAPI: practical setup and pitfalls

How do I set up Redis with FastAPI for session caching?

How do I use Redis as a result cache for expensive API calls?

What are effective cache invalidation strategies in FastAPI?

How do I handle Redis connection failures gracefully in FastAPI?

Redis vs in-memory caching for FastAPI: when should I use which?

How do I monitor Redis cache hit rates in production?

FAQ

Key Takeaways

Vibe Coding vs Agentic Coding: When to Use Which

What is vibe coding?

What is agentic coding?

How does vibe coding vs agentic coding change my workflow and output?

Can I mix vibe coding and agentic coding in a FastAPI project?

Example: Adding a new /classify route

Best practices and common pitfalls for each approach

Vibe Coding

Agentic Coding

When not to use either approach

Putting it all together in a FastAPI production pipeline

FAQ

Key Takeaways

Hybrid Search RAG Implementation: A Practical Guide

What is hybrid search and why combine vector similarity with traditional keyword search?

What are the core components of a hybrid RAG architecture?

How can I implement a hybrid search RAG using LangChain and a vector database?

Why this works in production

Which vector store should I pick for hybrid search and how do I configure it?

Quick Qdrant config snippet

How do I evaluate and tune hybrid search performance?

When NOT to use hybrid search

FAQ

Key Takeaways

Why ai generated code cannot be copyrighted and what to do

What is the legal status of AI-generated code in the US, EU, and elsewhere?

Why this matters for builders

How do I properly license AI-generated snippets?

Example: Adding a new `/classify` route