Introduction
FastAPI has become one of the most popular Python web frameworks for building APIs, thanks to its speed, automatic OpenAPI documentation, and native async support. However, even experienced developers often struggle with crafting efficient endpoints, validating data with Pydantic, and managing background tasks. The right prompts for AI assistants can dramatically accelerate your development workflow — from generating boilerplate code to debugging complex async patterns.
In this guide, I’ll share 10 specific, ready-to-use prompts for FastAPI that cover three critical areas: endpoint creation, Pydantic validation, and background processing. Each prompt includes a clear explanation of its purpose and a concrete usage example. Whether you’re building a RESTful API for a machine learning service or a real-time data pipeline, these prompts will save you hours of trial and error.
1. Generate a Basic CRUD Endpoint with Async SQLAlchemy
Purpose: Quickly scaffold a fully functional async endpoint for creating, reading, updating, and deleting resources.
Prompt:
Generate a FastAPI CRUD endpoint for managing 'products' using async SQLAlchemy 2.0. Include:
- A GET /products/ endpoint that returns a list of products with pagination (skip/limit).
- A GET /products/{product_id} endpoint for a single product.
- A POST /products/ endpoint that accepts a Pydantic model.
- A PUT /products/{product_id} endpoint for updating.
- A DELETE /products/{product_id} endpoint.
Use async session dependency injection. Return proper HTTP status codes.
Usage Example:
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
import models, schemas, database
app = FastAPI()
async def get_db() -> AsyncSession:
async with database.SessionLocal() as session:
yield session
@app.get("/products/", response_model=list[schemas.ProductOut])
async def list_products(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(models.Product).offset(skip).limit(limit))
return result.scalars().all()
2. Create a Pydantic Model with Nested Validation
Purpose: Define complex data structures with nested objects, custom validators, and conditional fields.
Prompt:
Create a Pydantic v2 model for an e-commerce order. Include:
- OrderItem sub-model with product_id, quantity, price.
- ShippingAddress sub-model with street, city, zip_code, country.
- A field 'status' that can only be 'pending', 'confirmed', or 'shipped'.
- A custom validator that ensures total_price = sum(item.price * item.quantity for all items).
- Use model_validator (after) for cross-field validation.
Usage Example:
from pydantic import BaseModel, Field, model_validator
from typing import List
class OrderItem(BaseModel):
product_id: int
quantity: int = Field(gt=0)
price: float = Field(gt=0)
class ShippingAddress(BaseModel):
street: str
city: str
zip_code: str
country: str
class OrderCreate(BaseModel):
items: List[OrderItem]
shipping_address: ShippingAddress
status: str = Field(default="pending", pattern="^(pending
|confirmed|shipped)$")
@model_validator(mode='after')
def check_total(self):
total = sum(item.price * item.quantity for item in self.items)
if total <= 0:
raise ValueError('Total order value must be positive')
return self
3. Write an Async Background Task for Email Notifications
Purpose: Offload non-blocking operations like sending emails without slowing down the API response.
Prompt:
Implement a FastAPI background task that sends a welcome email after user registration. Use:
- BackgroundTasks from fastapi.
- A mock email function that simulates a 2-second delay (use asyncio.sleep).
- The endpoint should return immediately with status 202 Accepted.
- Include proper error handling: if email fails, log the error but don't fail the request.
Usage Example:
from fastapi import FastAPI, BackgroundTasks, status
import asyncio
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
async def send_welcome_email(email: str):
await asyncio.sleep(2) # Simulate SMTP delay
logger.info(f"Welcome email sent to {email}")
@app.post("/register/", status_code=status.HTTP_202_ACCEPTED)
async def register_user(email: str, background_tasks: BackgroundTasks):
# Add user to database here
background_tasks.add_task(send_welcome_email, email)
return {"message": "User registered. Email will be sent."}
4. Implement Celery Task Queue for Heavy Background Jobs
Purpose: For long-running tasks (e.g., image processing, report generation) that can't fit in memory or need retries, use Celery with Redis.
Prompt:
Create a Celery task configuration for FastAPI with Redis as broker. Include:
- A celery.py file that initializes the Celery app.
- A task that resizes an image (simulate with asyncio.sleep).
- An endpoint that triggers the task and returns task_id.
- A GET /tasks/{task_id}/status endpoint to check task state.
- Use celery.result.AsyncResult.
Usage Example:
# celery_app.py
from celery import Celery
celery_app = Celery('tasks', broker='redis://localhost:6379/0')
@celery_app.task(bind=True, max_retries=3)
def process_image(self, image_path: str):
# Image processing logic
return {"status": "completed", "path": image_path}
5. Build an Async File Upload Endpoint with Validation
Purpose: Accept file uploads with size limits, MIME type checks, and async processing.
Prompt:
Write a FastAPI endpoint for uploading PDF files. Validate:
- File size < 10 MB.
- Only PDF files (MIME type 'application/pdf').
- Save file to disk asynchronously using aiofiles.
- Return the filename and size in the response.
- Use UploadFile and File from fastapi.
Usage Example:
from fastapi import FastAPI, UploadFile, File, HTTPException
import aiofiles
app = FastAPI()
@app.post("/upload/")
async def upload_pdf(file: UploadFile = File(...)):
if file.content_type != "application/pdf":
raise HTTPException(400, detail="Only PDF files allowed")
content = await file.read()
if len(content) > 10 * 1024 * 1024:
raise HTTPException(400, detail="File too large")
async with aiofiles.open(f"uploads/{file.filename}", "wb") as f:
await f.write(content)
return {"filename": file.filename, "size": len(content)}
6. Add Rate Limiting with SlowAPI
Purpose: Protect your API from abuse by limiting requests per user or IP.
Prompt:
Implement rate limiting on a FastAPI endpoint using slowapi library. Configure:
- Limit to 5 requests per minute per IP.
- Return 429 Too Many Requests with Retry-After header.
- Use Redis as storage backend.
- Apply the limiter globally but allow an exception for /health.
Usage Example:
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from fastapi import FastAPI
limiter = Limiter(key_func=get_remote_address, storage_uri="redis://localhost:6379/0")
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)
@app.get("/data")
@limiter.limit("5/minute")
async def get_data(request: Request):
return {"data": "some data"}
7. Create a WebSocket Endpoint for Real-Time Chat
Purpose: Handle bidirectional communication for live updates, chat, or streaming.
Prompt:
Write a FastAPI WebSocket endpoint for a simple chat room. Features:
- Connect via /ws/{room_id}.
- Broadcast messages to all clients in the same room.
- Handle disconnection gracefully.
- Use a simple in-memory dictionary to track connections.
- Send a welcome message when a new client joins.
Usage Example:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
rooms = {}
@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
await websocket.accept()
if room_id not in rooms:
rooms[room_id] = []
rooms[room_id].append(websocket)
try:
while True:
data = await websocket.receive_text()
for client in rooms[room_id]:
await client.send_text(f"User: {data}")
except WebSocketDisconnect:
rooms[room_id].remove(websocket)
8. Implement JWT Authentication with OAuth2 Password Flow
Purpose: Secure your FastAPI endpoints with token-based authentication.
Prompt:
Build a JWT authentication system in FastAPI using OAuth2PasswordBearer. Include:
- POST /token endpoint that accepts username/password and returns access_token.
- Token expiry set to 30 minutes.
- A protected endpoint /users/me that returns current user details.
- Use passlib for password hashing.
- Store users in a simple dict (simulate database).
Usage Example:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# Validate user and return JWT
pass
9. Generate OpenAPI Custom Metadata and Tags
Purpose: Improve API documentation with custom descriptions, tags, and examples.
Prompt:
Enhance FastAPI's automatic OpenAPI documentation by:
- Adding a global description and version to the app.
- Grouping endpoints under tags: 'users', 'products', 'admin'.
- Adding example values to Pydantic models using Field(examples=...).
- Including a custom response description for 404 errors.
- Setting a separate license info.
Usage Example:
from fastapi import FastAPI
app = FastAPI(
title="E-Commerce API",
description="API for managing products and orders",
version="2.0.0",
contact={"name": "Support", "email": "support@example.com"},
license_info={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
)
@app.get("/products/", tags=["products"])
async def list_products():
return [{"id": 1, "name": "Laptop"}]
10. Write a Dependency for Database Session Management
Purpose: Create reusable dependencies for database sessions, authentication, or configuration.
Prompt:
Design a FastAPI dependency that:
- Provides an async database session using SQLAlchemy 2.0.
- Automatically closes the session after the request.
- Includes a sub-dependency that gets the current user from JWT token.
- Returns a 401 if token is invalid.
- Use Depends() to chain dependencies.
Usage Example:
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db() -> AsyncSession:
async with SessionLocal() as session:
yield session
async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
# Decode JWT and fetch user from DB
user = await db.get(User, payload["sub"])
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
@app.get("/profile/")
async def get_profile(current_user: User = Depends(get_current_user)):
return current_user
Conclusion
These 10 prompts cover the most common pain points in FastAPI development — from basic CRUD to advanced background processing and authentication. By using them as templates, you can cut down boilerplate code and focus on what makes your API unique. Remember that AI-generated code is a starting point; always test edge cases and review security aspects, especially for authentication and file uploads.
For teams that need to integrate these APIs with external services like Telegram or Salesforce, ASI Biont supports connecting to various APIs through its integration framework — learn more at asibiont.com/courses. Start experimenting with these prompts in your next FastAPI project, and you'll see how much faster you can ship production-ready endpoints.
Further reading: FastAPI official documentation at fastapi.tiangolo.com and Pydantic v2 docs at docs.pydantic.dev.
Comments