10 Essential Prompts for Python Code Generation: From Scripts to FastAPI

Write a Python script that:
- Takes an input CSV file path and an output CSV file path as command-line arguments.
- Reads the CSV file using the csv module.
- Filters rows where the 'status' column equals 'active'.
- Writes the filtered rows to the output CSV file, preserving the header.
- Includes error handling for missing files or invalid CSV format.

import csv
import sys

def filter_active_csv(input_path, output_path):
    try:
        with open(input_path, 'r', newline='') as infile:
            reader = csv.DictReader(infile)
            fieldnames = reader.fieldnames
            if 'status' not in fieldnames:
                raise ValueError("CSV missing 'status' column")
            rows = [row for row in reader if row['status'] == 'active']
        with open(output_path, 'w', newline='') as outfile:
            writer = csv.DictWriter(outfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(rows)
        print(f"Filtered {len(rows)} rows written to {output_path}")
    except FileNotFoundError:
        print(f"Error: File {input_path} not found.")
    except ValueError as e:
        print(f"Error: {e}")

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print("Usage: python filter_csv.py <input.csv> <output.csv>")
        sys.exit(1)
    filter_active_csv(sys.argv[1], sys.argv[2])

Create a Python script using the requests library that:
- Fetches data from https://api.example.com/items (replace with actual URL).
- Implements retry logic with exponential backoff (max 3 retries) for HTTP 429 and 5xx errors.
- Parses the JSON response and saves it to 'data.json'.
- Prints the number of items fetched.

import requests
import json
import time

def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            wait = 2 ** attempt
            print(f"Rate limited. Waiting {wait}s...")
            time.sleep(wait)
        elif response.status_code >= 500:
            wait = 2 ** attempt
            print(f"Server error {response.status_code}. Retrying in {wait}s...")
            time.sleep(wait)
        else:
            response.raise_for_status()
    raise Exception(f"Failed after {max_retries} attempts")

data = fetch_with_retry('https://api.example.com/items')
with open('data.json', 'w') as f:
    json.dump(data, f, indent=2)
print(f"Fetched {len(data)} items.")

Write a Python module called 'pipeline.py' with the following functions, all using type hints:
- load_data(path: str) -> list[dict]: Loads JSON data from a file.
- clean_data(data: list[dict]) -> list[dict]: Removes entries with missing 'id' or 'value' keys, and converts 'value' to float.
- transform_data(data: list[dict]) -> list[dict]: Adds a 'value_squared' key to each entry.
- aggregate_data(data: list[dict]) -> dict: Returns the sum, mean, and count of 'value' fields.
Include a main block that runs the pipeline on 'input.json' and prints the aggregation result.

import json
from typing import List, Dict

def load_data(path: str) -> List[Dict]:
    with open(path, 'r') as f:
        return json.load(f)

def clean_data(data: List[Dict]) -> List[Dict]:
    cleaned = []
    for item in data:
        if 'id' in item and 'value' in item:
            try:
                item['value'] = float(item['value'])
                cleaned.append(item)
            except (ValueError, TypeError):
                continue
    return cleaned

def transform_data(data: List[Dict]) -> List[Dict]:
    for item in data:
        item['value_squared'] = item['value'] ** 2
    return data

def aggregate_data(data: List[Dict]) -> Dict:
    values = [item['value'] for item in data]
    return {
        'sum': sum(values),
        'mean': sum(values) / len(values) if values else 0,
        'count': len(values)
    }

if __name__ == '__main__':
    raw = load_data('input.json')
    clean = clean_data(raw)
    transformed = transform_data(clean)
    result = aggregate_data(transformed)
    print(result)

Write a pytest test file 'test_pipeline.py' that tests all functions in 'pipeline.py' using a sample dataset. Include:
- A fixture that creates a temporary JSON file with test data.
- Tests for clean_data: ensure missing keys are removed and 'value' is converted to float.
- Tests for transform_data: verify 'value_squared' is correct.
- Tests for aggregate_data: verify sum, mean, count for known values.

import pytest
import json
import tempfile
import os
from pipeline import load_data, clean_data, transform_data, aggregate_data

@pytest.fixture
def sample_data():
    return [
        {'id': 1, 'value': '10.5'},
        {'id': 2, 'value': '20.3'},
        {'id': 3},  # missing value
        {'value': '30.1'},  # missing id
        {'id': 4, 'value': 'abc'}  # invalid value
    ]

@pytest.fixture
def temp_json_file(sample_data):
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
        json.dump(sample_data, f)
        fname = f.name
    yield fname
    os.unlink(fname)

def test_load_data(temp_json_file):
    data = load_data(temp_json_file)
    assert len(data) == 5

def test_clean_data(sample_data):
    cleaned = clean_data(sample_data)
    assert len(cleaned) == 2  # only first two valid
    assert cleaned[0]['value'] == 10.5
    assert isinstance(cleaned[0]['value'], float)

def test_transform_data():
    data = [{'id': 1, 'value': 3.0}]
    transformed = transform_data(data)
    assert transformed[0]['value_squared'] == 9.0

def test_aggregate_data():
    data = [{'id': 1, 'value': 10.0}, {'id': 2, 'value': 20.0}]
    agg = aggregate_data(data)
    assert agg['sum'] == 30.0
    assert agg['mean'] == 15.0
    assert agg['count'] == 2

Create a FastAPI application with SQLAlchemy that:
- Defines a 'Item' model with fields: id (int, primary key), name (str), description (str, optional), price (float), created_at (datetime, default now).
- Implements endpoints: GET /items (list all), GET /items/{id} (get one), POST /items (create), PUT /items/{id} (update), DELETE /items/{id} (delete).
- Uses Pydantic schemas for request and response validation.
- Includes database session management with dependency injection.
- Runs on SQLite for simplicity.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from datetime import datetime
from typing import Optional

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    price = Column(Float)
    created_at = Column(DateTime, default=datetime.utcnow)

Base.metadata.create_all(bind=engine)

# Pydantic schemas
class ItemCreate(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

class ItemResponse(ItemCreate):
    id: int
    created_at: datetime
    class Config:
        orm_mode = True

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.get("/items", response_model=list[ItemResponse])
def list_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

@app.get("/items/{item_id}", response_model=ItemResponse)
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.post("/items", response_model=ItemResponse, status_code=201)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.put("/items/{item_id}", response_model=ItemResponse)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    for key, value in item.dict().items():
        setattr(db_item, key, value)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(Item).filter(Item.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    db.delete(db_item)
    db.commit()
    return None

Write a FastAPI dependency that:
- Extracts the JWT token from the Authorization header (Bearer scheme).
- Validates the token using PyJWT (python-jose) with a secret key and HS256 algorithm.
- If valid, extracts the 'sub' (user ID) claim and returns it.
- If invalid or missing, raises HTTP 401.
- Include an endpoint /protected that uses this dependency and returns a welcome message with the user ID.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

app = FastAPI()
security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return user_id
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/protected")
def protected_route(user_id: str = Depends(get_current_user)):
    return {"message": f"Hello, user {user_id}!"}

Create a FastAPI app that:
- Uses Celery with Redis as a message broker.
- Defines a background task that simulates a long computation (sleep 10 seconds) and returns a result.
- An endpoint /start-task that initiates the task and returns a task ID.
- An endpoint /task-status/{task_id} that returns the current status and result if ready.

# tasks.py
from celery import Celery
import time

celery_app = Celery('tasks', broker='redis://localhost:6379/0')

@celery_app.task(bind=True)
def long_computation(self, x: int, y: int):
    time.sleep(10)
    return x + y

# main.py
from fastapi import FastAPI
from celery.result import AsyncResult
from tasks import long_computation, celery_app

app = FastAPI()

@app.post("/start-task")
def start_task(x: int, y: int):
    task = long_computation.delay(x, y)
    return {"task_id": task.id}

@app.get("/task-status/{task_id}")
def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    return {
        "task_id": task_id,
        "status": result.state,
        "result": result.result if result.ready() else None
    }

Write an Alembic migration script that:
- Adds a column 'is_active' (Boolean, default=True) to the 'items' table.
- Includes both upgrade() and downgrade() functions.
- Assumes the table already exists and the revision prefix is 'abc123'.

"""Add is_active column to items table

Revision ID: abc123_add_is_active
Revises: previous_revision
Create Date: 2026-07-12 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123_add_is_active'
down_revision = 'previous_revision'
branch_labels = None
depends_on = None

def upgrade():
    op.add_column('items', sa.Column('is_active', sa.Boolean(), server_default='1', nullable=False))

def downgrade():
    op.drop_column('items', 'is_active')

Create a FastAPI WebSocket endpoint /ws/chat that:
- Accepts a query parameter 'room' (string).
- Maintains a dictionary of connected clients per room.
- Broadcasts incoming messages to all other clients in the same room.
- Handles disconnection gracefully.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        self.rooms[room].remove(websocket)
        if not self.rooms[room]:
            del self.rooms[room]

    async def broadcast(self, message: str, sender: WebSocket, room: str):
        for client in self.rooms.get(room, []):
            if client != sender:
                await client.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket, room: str = "default"):
    await manager.connect(websocket, room)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"User says: {data}", websocket, room)
    except WebSocketDisconnect:
        manager.disconnect(websocket, room)
        await manager.broadcast("A user left the chat", websocket, room)

Write a docker-compose.yml file that:
- Defines three services: api (FastAPI), redis (Redis), and worker (Celery).
- The api service builds from ./app, exposes port 8000, and depends on redis.
- The worker service builds from ./app, runs the Celery worker command, and depends on redis.
- All services use a common network.

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    restart: unless-stopped

  api:
    build: ./app
    command: uvicorn main:app --host 0.0.0.0 --port 8000
    ports:
      - "8000:8000"
    depends_on:
      - redis
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    restart: unless-stopped

  worker:
    build: ./app
    command: celery -A tasks worker --loglevel=info
    depends_on:
      - redis
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
    restart: unless-stopped