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

Create FastAPI endpoints for CRUD operations on a User resource. Use SQLAlchemy for ORM and Pydantic for request/response schemas. Include GET /users/{id}, POST /users, PUT /users/{id}, DELETE /users/{id}. Return proper HTTP status codes: 200 for success, 201 for creation, 404 for not found, 204 for deletion. Use async database session.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel

app = FastAPI()

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

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
    # implementation
    pass

Modify the GET /users endpoint to accept query parameters 'page' (default 1) and 'size' (default 10). Return a JSON object with 'items', 'total', 'page', 'size', and 'pages' fields. Use async SQLAlchemy count and limit/offset.

Write a FastAPI dependency called 'get_current_user' that extracts a token from the Authorization header, validates it using PyJWT, and returns the user object from the database. Raise 401 if invalid. Use OAuth2PasswordBearer.

Define a Pydantic BaseModel for Product with fields: name (str, 3-50 chars), price (float, >0), contact_email (EmailStr). Add a custom validator that ensures the name is title-cased. Use field_validator and Field.

from pydantic import BaseModel, Field, field_validator, EmailStr

class Product(BaseModel):
    name: str = Field(min_length=3, max_length=50)
    price: float = Field(gt=0)
    contact_email: EmailStr

    @field_validator("name")
    def title_case_name(cls, v):
        return v.title()

Create Pydantic models for Order: OrderItem (product_id, quantity, unit_price), Order (id, customer_name, items: list[OrderItem], total_price computed as sum of items). Use model_validator to compute total after validation.

Implement a discriminated union in Pydantic v2 for an Order that can be either StandardOrder (delivery_date) or ExpressOrder (priority_level). Use discriminated union with a 'type' field.

Add a background task to the POST /users endpoint that simulates sending a welcome email. Use FastAPI's BackgroundTasks. The task should accept user email and name, and print a message after a 2-second delay (simulating SMTP call).

from fastapi import BackgroundTasks

def send_welcome_email(email: str, name: str):
    import asyncio
    asyncio.sleep(2)  # simulate
    print(f"Welcome email sent to {name} at {email}")

@app.post("/users", status_code=201)
async def create_user(user: UserCreate, bg_tasks: BackgroundTasks):
    db_user = await create_user_in_db(user)
    bg_tasks.add_task(send_welcome_email, user.email, user.name)
    return db_user

Set up a Celery task for resizing an uploaded image to 300x300 pixels. In FastAPI, accept a file upload, save it temporarily, and call the Celery task asynchronously. Return a task ID that the client can poll for status.

Use APScheduler's AsyncIOScheduler to run a function every hour that deletes tokens older than 24 hours from the database. Start the scheduler on app startup via lifespan.

Implement a custom exception handler for RequestValidationError that returns a JSON with 'error_code', 'message', and 'details' containing field-specific errors. Use @app.exception_handler(RequestValidationError).

Write a middleware that tracks request counts per IP using an in-memory dictionary with timestamps. If a client exceeds 10 requests in 60 seconds, return 429 Too Many Requests. Use BaseHTTPMiddleware.

Refactor a GET /reports endpoint that aggregates data from multiple tables. Replace synchronous session.execute with async session.execute (using AsyncSession). Use asyncio.gather to run independent queries concurrently.