10 Prompts for FastAPI: Endpoints, Pydantic, and Background Processing

## 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:** ```python 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:** ```python 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:** ```python 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:** ```python # 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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.*

10 Prompts for FastAPI: Endpoints, Pydantic, and Background Processing

Comments

Recent articles

The Real Barriers to AI Adoption in SMBs: Insights from 86 Deep Interviews

Maritime Law and Shipping (IMO, UN, DNV): How AI Learning Helps You Understand SOLAS, MARPOL, and PSC

Mastering Production Observability: Why the Observability Course on asibiont.com Is Your Career Catalyst

Netlify + ASI Biont AI Agent: DevOps Without Code and Routine — Automate Deployment and Monitoring in 5 Minutes

Amazon’s AI HR Revolution: When Vacation Requests Became a Nightmare

From Dispute to $2 Million Settlement: How the International Law & Arbitration — LL.M. Level Course Transforms Cross-Border Careers

From Clips to Career: Why the Video Editing & Animation Course on Asibiont.com Is a Game-Changer

OpenClaw Is Finally Available on Android and iOS: What This Means for AI Agents in 2026

How We Built Discovery AI: The Neural Search Powering Russia's Largest Content Database