12 Prompts for FastAPI: Master Endpoints, Pydantic, and Async Background Tasks

FastAPI has taken the Python web development world by storm, offering an unbeatable combination of speed, type safety, and automatic API documentation. Whether you're building a minimal REST service or a high-throughput async application, FastAPI's design enables you to get more done with less boilerplate. But even seasoned developers can struggle with structuring prompts for AI assistants or code generators when working with FastAPI. That's where a well-crafted prompt becomes your secret weapon.

In this article, I’ve curated 12 practical prompts for FastAPI, organized into three skill levels: basic, advanced, and expert. Each prompt includes a clear task, a ready-to-use prompt you can copy-paste into an AI assistant like Claude, GPT-4, or Gemini, and a real-world example result. By the end, you’ll not only speed up your FastAPI development but also learn how to think in prompts—breaking down complex backend challenges into precise, actionable instructions.

Basic Prompts

These prompts are perfect for beginners who want to build simple APIs, understand validation, and get comfortable with FastAPI's core concepts. They assume no prior async experience and focus on clean, readable code.

1. Task: Create a Simple Health Check Endpoint

Prompt:

Write a FastAPI application with a GET endpoint /health that returns a JSON response {"status": "ok"}. Include the necessary imports and run the app using Uvicorn. Keep it minimal and clean.

Example Result:

from fastapi import FastAPI

app = FastAPI()

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

2. Task: Implement a CRUD API with Pydantic Models

Prompt:

Build a FastAPI CRUD API for a Product model with fields id, name, price, and in_stock. Implement endpoints to create, read, update, and delete products. Use Pydantic's BaseModel for request/response schemas, and store items in a simple Python list. Include proper status codes for each operation.

Example Result:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Product(BaseModel):
    id: int
    name: str
    price: float
    in_stock: bool = True

products = []

@app.post("/products/", status_code=status.HTTP_201_CREATED)
async def create_product(product: Product):
    products.append(product)
    return product

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    for product in products:
        if product.id == product_id:
            return product
    raise HTTPException(status_code=404, detail="Product not found")

@app.put("/products/{product_id}")
async def update_product(product_id: int, updated: Product):
    for i, product in enumerate(products):
        if product.id == product_id:
            products[i] = updated
            return updated
    raise HTTPException(status_code=404, detail="Product not found")

@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int):
    for i, product in enumerate(products):
        if product.id == product_id:
            products.pop(i)
            return
    raise HTTPException(status_code=404, detail="Product not found")

3. Task: Add Request Validation with Query Parameters

Prompt:

Enhance a FastAPI endpoint that accepts a query parameter q (string, optional) and limit (integer, default 10). Validate that limit is between 1 and 100. Also add path parameter item_id that must be an integer. Use FastAPI's validation features and return a descriptive error message if validation fails.

Example Result:

from fastapi import FastAPI, Query, Path, HTTPException
from typing import Optional

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., description="The ID of the item to retrieve"),
    q: Optional[str] = Query(None, max_length=50),
    limit: int = Query(10, ge=1, le=100)
):
    return {"item_id": item_id, "q": q, "limit": limit}

4. Task: Return a List of Items with Filtering and Pagination

Prompt:

Write a FastAPI endpoint /items that returns a large list of items. Support query parameters skip (int, default 0) and limit (int, default 10) for pagination, and category (optional string) for filtering. Use a mock list of dictionaries as the data source. Ensure the endpoint is async and efficient.

Example Result:

from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

# Mock data
items_db = [
    {"id": i, "name": f"Item {i}", "category": "A" if i % 2 == 0 else "B"}
    for i in range(1000)
]

@app.get("/items", response_model=List[dict])
async def list_items(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    category: Optional[str] = None
):
    filtered = [item for item in items_db if category is None or item["category"] == category]
    return filtered[skip : skip + limit]

Advanced Prompts

Once you're comfortable with the basics, these prompts will help you build production-ready APIs with file uploads, background processing, custom error handling, and dependency injection.

5. Task: Handle File Upload with Async Processing

Prompt:

Create a FastAPI endpoint that accepts a single file upload via UploadFile. Save the file to disk asynchronously, then return its metadata (filename, size, content type). Also include a variant that accepts multiple files. Use aiofiles for non-blocking file I/O.

Example Result:

from fastapi import FastAPI, File, UploadFile
import aiofiles
import os

app = FastAPI()

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    async with aiofiles.open(file_path, "wb") as buffer:
        content = await file.read()
        await buffer.write(content)
    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

@app.post("/upload-multiple/")
async def upload_multiple(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        file_path = os.path.join(UPLOAD_DIR, file.filename)
        async with aiofiles.open(file_path, "wb") as buffer:
            content = await file.read()
            await buffer.write(content)
        results.append({"filename": file.filename, "size": len(content)})
    return results

6. Task: Use BackgroundTasks for Sending Emails

Prompt:

Extend a FastAPI endpoint that creates a new user. After successfully creating the user, add a background task to send a welcome email. The email sending should be simulated with a simple print statement or a mock function. Ensure the background task runs after the response is sent, and the endpoint returns immediately.

Example Result:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    email: str
    name: str

def send_welcome_email(email: str, name: str):
    # Simulate sending an email
    print(f"Sending welcome email to {email} for user {name}")

@app.post("/users/")
async def create_user(user: User, background_tasks: BackgroundTasks):
    # Here you would normally save the user to a database
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    return {"message": "User created", "user": user}

7. Task: Custom Exception Handlers and Error Responses

Prompt:

Write a FastAPI application that defines a custom exception ItemNotFoundError. Create a global exception handler that catches this exception and returns a JSON response with a 404 status code and a custom error message. Also override the default HTTPException handler to format errors consistently.

Example Result:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException

app = FastAPI()

class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item with id {exc.item_id} not found"}
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "status_code": exc.status_code}
    )

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id > 100:
        raise ItemNotFoundError(item_id)
    return {"item_id": item_id}

8. Task: Dependency Injection for Database Sessions

Prompt:

Create a FastAPI dependency that provides a database session. Use a fake database connection pool. The dependency should ensure the session is closed after the request is done. Then use this dependency in a route to fetch a user from the database. Use yield to handle cleanup.

Example Result:

from fastapi import FastAPI, Depends, HTTPException
from typing import Generator

app = FastAPI()

class DatabaseSession:
    def __init__(self):
        self.connection = "postgresql://mock"
    def fetch_user(self, user_id: int):
        return {"id": user_id, "name": "John Doe"}
    def close(self):
        print("Closing database session")

def get_db() -> Generator[DatabaseSession, None, None]:
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: DatabaseSession = Depends(get_db)):
    user = db.fetch_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Expert Prompts

These prompts push FastAPI to its limits: integrating with message queues, streaming real-time data, building WebSocket endpoints, and creating complex custom validation logic.

9. Task: Implement a Background Task Queue with Celery and Redis

Prompt:

Set up a FastAPI application that dispatches a long-running task to a Celery worker. Use Redis as the message broker and result backend. Define a Celery task that simulates heavy processing (e.g., sleeping for 5 seconds) and returns a result. In FastAPI, create an endpoint to start the task and another to check its status using AsyncResult. Include instructions for running the worker and the FastAPI app.

Example Result:

# celery_app.py
from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1"
)

@celery_app.task
def long_running_task(duration: int):
    import time
    time.sleep(duration)
    return "Task completed after {} seconds".format(duration)

# main.py
from fastapi import FastAPI
from celery.result import AsyncResult
from celery_app import long_running_task

app = FastAPI()

@app.post("/start-task/")
async def start_task(duration: int):
    task = long_running_task.delay(duration)
    return {"task_id": task.id}

@app.get("/task-status/{task_id}")
async def get_status(task_id: str):
    result = AsyncResult(task_id)
    if result.ready():
        return {"status": "completed", "result": result.get()}
    return {"status": "pending"}

Run: celery -A celery_app worker --loglevel=info and uvicorn main:app --reload.

10. Task: Streaming Responses with Async Generators

Prompt:

Build a FastAPI endpoint that streams a large dataset as a continuous JSON array using StreamingResponse. The data should be generated asynchronously in chunks. For demonstration, generate numbers from 0 to 99 with a small delay between each. Set appropriate media type and headers so the client can process the stream.

Example Result:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()

async def generate_numbers():
    for i in range(100):
        # Simulate async I/O
        await asyncio.sleep(0.01)
        yield json.dumps({"index": i}) + "\n"

@app.get("/stream-numbers")
async def stream_numbers():
    return StreamingResponse(
        generate_numbers(),
        media_type="application/x-ndjson",
        headers={"X-Content-Type-Options": "nosniff"}
    )

11. Task: WebSocket Endpoint for Real-Time Updates

Prompt:

Implement a WebSocket endpoint in FastAPI that accepts a client connection and sends real-time updates every second. The updates should be generated by an async generator. Handle disconnections gracefully and log when clients connect or disconnect. Also provide a simple HTML client page that connects to the WebSocket and displays the messages.

Example Result:

# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        for i in range(10):
            await websocket.send_text(f"Update {i}")
            await asyncio.sleep(1)
    except WebSocketDisconnect:
        print("Client disconnected")

12. Task: Custom Pydantic Validators and Field Serialization

Prompt:

Create a Pydantic model that represents an Order with fields product_id, quantity, unit_price, and total_price. Use a field validator to ensure quantity is positive and unit_price is non-negative. Add a computed property for total_price that multiplies quantity by unit_price. Also define a custom serializer for a discount_code field that uppercases any input and strips whitespace.

Example Result:

from pydantic import BaseModel, Field, validator

class Order(BaseModel):
    product_id: int
    quantity: int
    unit_price: float
    discount_code: str = None

    @validator("quantity")
    def validate_quantity(cls, v):
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

    @validator("unit_price")
    def validate_price(cls, v):
        if v < 0:
            raise ValueError("Price cannot be negative")
        return v

    @validator("discount_code", always=True)
    def normalize_code(cls, v):
        if v:
            return v.strip().upper()
        return v

    @property
    def total_price(self) -> float:
        return self.quantity * self.unit_price

# Usage
order = Order(product_id=1, quantity=3, unit_price=9.99, discount_code=" save20 ")
print(order.total_price)  # 29.97
print(order.discount_code)  # "SAVE20"

Performance Comparison at a Glance

Why Prompting Matters for FastAPI

The prompts in this article aren't just copy-paste snippets—they teach you a systematic approach to turning requirements into production-grade code. When you craft a prompt for FastAPI, you're essentially doing a mini code review: you specify the input/output contracts, the error handling, and the concurrency model. That level of detail is what separates a good API from a secure, maintainable one.

As you get more comfortable, you'll start mixing and extending these prompts. For example, combine the dependency injection pattern from #8 with the Celery integration from #9 to build a fully decoupled microservice architecture. Or add the custom validators from #12 to a WebSocket payload for type safety.

The Next Step

I hope these 12 prompts become your go-to starting point for any FastAPI project. Copy them, adapt them, and make them your own. Have a prompt that's worked surprisingly well? Share it in the comments below—I'd love to see what you're building.

If you want to dive deeper into FastAPI, check out the official documentation or my other tutorials on database integrations and testing. And remember: the best API is the one that solves the problem without introducing new ones. Now go build something amazing!