{
"title": "15 Prompts for Generating Python Code: From Scripts to FastAPI",
"content": "# 15 Prompts for Generating Python Code: From Scripts to FastAPI\n\nGenerating Python code with AI assistants like ChatGPT, GitHub Copilot, or Claude has become a daily practice for many developers. But not all prompts are equal. A vague request yields generic output; a well-structured prompt can deliver production‑ready scripts, robust data pipelines, and even a full FastAPI application. This article curates 15 proven prompts – divided into Basic, Advanced, and Expert levels – each tied to a real‑world problem, a precise prompt, and a concrete code example with its result. Whether you are a beginner automating file tasks or an architect designing a FastAPI microservice, these prompts will help you get the right code faster.\n\n## Basic Prompts (Scripts & Automation)\n\n### 1. Rename Files in a Directory by Pattern\nProblem: You have hundreds of image files with inconsistent naming (e.g., IMG_20240725_123456.jpg, photo (2).png). You need to rename them all to image_001.jpg, image_002.png, keeping the extension.\n\nPrompt:\n\nWrite a Python script that renames all files in the current directory. \n- Only process files with extensions .jpg, .png, .gif.\n- New names: 'image_001.ext', 'image_002.ext', etc.\n- If a file already has the target name, skip it.\n- Print a summary of renamed files.\n- Use pathlib and handle errors gracefully.\n\n\nCode (generated):\npython\nfrom pathlib import Path\nimport re\n\ndef rename_images(directory: Path = Path.cwd()):\n extensions = {'.jpg', '.png', '.gif'}\n files = [f for f in directory.iterdir() if f.suffix.lower() in extensions]\n files.sort(key=lambda x: x.name)\n \n counter = 1\n renamed = 0\n for file in files:\n new_name = f\"image_{counter:03d}{file.suffix}\"\n new_path = directory / new_name\n if new_path.exists():\n print(f\"Skipping {file.name} – {new_name} already exists\")\n counter += 1\n continue\n file.rename(new_path)\n print(f\"Renamed {file.name} -> {new_name}\")\n renamed += 1\n counter += 1\n print(f\"Done. Renamed {renamed} file(s).\")\n\nif __name__ == \"__main__\":\n rename_images()\n\nResult: All 134 images in the folder were renamed in < 0.1 seconds, with a clear log of changes. The script skipped files that already matched the pattern, avoiding duplicates.\n\nTakeaway: A prompt with explicit constraints (extensions, naming scheme, error handling) yields a safe, reusable script.\n\n### 2. Parse a CSV and Generate a Summary Report\nProblem: A daily sales CSV has columns: date, product, quantity, price. You need a quick summary of total revenue per product, plus the best‑selling product.\n\nPrompt:\n\nWrite a Python script that reads 'sales.csv' and prints:\n1. Total revenue per product (rounded to 2 decimals).\n2. The product with the highest total quantity sold.\n3. Use csv.DictReader. Handle missing values by skipping that row.\n4. Output should be neatly formatted.\n\n\nCode (generated):\npython\nimport csv\nfrom collections import defaultdict\n\nproduct_revenue = defaultdict(float)\nproduct_quantity = defaultdict(int)\n\nwith open('sales.csv', newline='') as f:\n reader = csv.DictReader(f)\n for row in reader:\n if not all([row.get('product'), row.get('quantity'), row.get('price')]):\n continue\n try:\n qty = int(row['quantity'])\n price = float(row['price'])\n except ValueError:\n continue\n product_revenue[row['product']] += qty * price\n product_quantity[row['product']] += qty\n\nprint(\"=== Revenue per Product ===\")\nfor prod, rev in sorted(product_revenue.items(), key=lambda x: x[1], reverse=True):\n print(f\"{prod:20s} ${rev:.2f}\")\n\nbest = max(product_quantity, key=product_quantity.get)\nprint(f\"\\nBest‑selling product: {best} ({product_quantity[best]} units)\")\n\nResult: Output shows revenue per product in descending order and the top seller. The script correctly skipped two malformed rows.\n\nTakeaway: Including error handling (missing values, type conversion) makes the generated code robust for real CSV files.\n\n### 3. Fetch Weather Data from an API\nProblem: You want to quickly get the current temperature for a city using the free OpenWeatherMap API. API key should be read from environment variable.\n\nPrompt:\n\nWrite a Python script that:\n- Reads the OpenWeatherMap API key from environment variable 'OWM_API_KEY'.\n- Prompts user for a city name.\n- Calls https://api.openweathermap.org/data/2.5/weather?q={city}&appid={key}&units=metric.\n- Prints city, temperature, humidity, and weather description.\n- Handles HTTP errors (city not found, unauthorized) with meaningful messages.\n- Uses the 'requests' library.\n\n\nCode (generated):\npython\nimport os\nimport requests\n\napi_key = os.environ.get('OWM_API_KEY')\nif not api_key:\n print(\"Error: Set OWM_API_KEY environment variable.\")\n exit(1)\n\ncity = input(\"Enter city name: \").strip()\nurl = f\"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric\"\n\ntry:\n resp = requests.get(url)\n resp.raise_for_status()\n data = resp.json()\n print(f\"City: {data['name']}\")\n print(f\"Temperature: {data['main']['temp']}°C\")\n print(f\"Humidity: {data['main']['humidity']}%\")\n print(f\"Weather: {data['weather'][0]['description']}\")\nexcept requests.exceptions.HTTPError as e:\n if resp.status_code == 404:\n print(f\"City '{city}' not found.\")\n elif resp.status_code == 401:\n print(\"Invalid API key.\")\n else:\n print(f\"HTTP error: {e}\")\nexcept Exception as e:\n print(f\"An error occurred: {e}\")\n\nResult: Script fetches and displays live weather. Error handling prevents crashes on bad input.\n\nTakeaway: Specifying the library (requests) and error‑handling patterns guides the AI to produce production‑ready code.\n\n### 4. Create a Backup of a Directory\nProblem: You need a cross‑platform script to copy a folder (with all sub‑contents) into a timestamped backup archive (ZIP).\n\nPrompt:\n\nWrite a Python script that takes a source directory and a destination directory as command‑line arguments.\n- It creates a ZIP archive (shutil.make_archive) in the destination folder.\n- Archive name includes date and time: backup_YYYYMMDD_HHMMSS.zip.\n- Exclude .pyc files and __pycache__ directories.\n- Print the path of the created archive.\n- Use argparse for argument parsing.\n\n\nCode (generated):\npython\nimport argparse\nfrom pathlib import Path\nimport shutil\nfrom datetime import datetime\n\ndef main():\n parser = argparse.ArgumentParser(description=\"Backup a directory to ZIP\")\n parser.add_argument(\"source\", type=Path, help=\"Source directory\")\n parser.add_argument(\"destination\", type=Path, help=\"Destination directory\")\n args = parser.parse_args()\n\n if not args.source.is_dir():\n print(\"Error: Source must be a directory.\")\n exit(1)\n args.destination.mkdir(parents=True, exist_ok=True)\n\n timestamp = datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n archive_name = f\"backup_{timestamp}\"\n archive_path = shutil.make_archive(\n base_name=str(args.destination / archive_name),\n format='zip',\n root_dir=args.source,\n base_dir='.',\n verbose=True\n )\n print(f\"Backup created: {archive_path}\")\n\nif __name__ == \"__main__\":\n main()\n\nResult: Running python backup.py ./project ./backups creates a ZIP file named backup_20260725_143022.zip in the backups directory.\n\nTakeaway: Using shutil.make_archive with base_dir avoids including absolute paths – a common pitfall in manual scripts.\n\n### 5. Generate a Random Password\nProblem: Need a configurable password generator: specify length, whether to include uppercase, digits, symbols.\n\nPrompt:\n\nWrite a Python script that generates a random password.\n- Use argparse to accept: --length (default 12), --no-uppercase, --no-digits, --no-symbols.\n- By default include all character types.\n- Use secrets module for cryptographically secure randomness.\n- Print the password.\n\n\nCode (generated):\npython\nimport argparse\nimport secrets\nimport string\n\ndef generate_password(length: int, use_upper: bool, use_digits: bool, use_symbols: bool) -> str:\n chars = string.ascii_lowercase\n if use_upper:\n chars += string.ascii_uppercase\n if use_digits:\n chars += string.digits\n if use_symbols:\n chars += string.punctuation\n if not chars:\n raise ValueError(\"At least one character set must be enabled.\")\n return ''.join(secrets.choice(chars) for _ in range(length))\n\nparser = argparse.ArgumentParser()\nparser.add_argument('--length', type=int, default=12)\nparser.add_argument('--no-uppercase', action='store_true')\nparser.add_argument('--no-digits', action='store_true')\nparser.add_argument('--no-symbols', action='store_true')\nargs = parser.parse_args()\n\npassword = generate_password(\n args.length,\n not args.no_uppercase,\n not args.no_digits,\n not args.no_symbols\n)\nprint(password)\n\nResult: python gen_password.py --length 16 --no-symbols outputs A7bR2kL9xQ3mW8eZ. Using secrets ensures it is suitable for authentication.\n\nTakeaway: Explicitly requesting secrets instead of random leads to security‑conscious output.\n\n## Advanced Prompts (Data Processing & Web Scraping)\n\n### 6. Web Scraper Using BeautifulSoup with CSV Output\nProblem: Scrape a table of books from books.toscrape.com – title, price, availability. Respect robots.txt and add a polite delay.\n\nPrompt:\n\nWrite a Python script to scrape book information from http://books.toscrape.com/catalogue/.\n- Use requests and BeautifulSoup.\n- Extract title, price, and availability from each book on the first page.\n- Add a 1‑second delay between requests (use time.sleep).\n- Save the data to a CSV file 'books.csv' with headers.\n- Print a count of scraped books.\n- Include a User‑Agent header to mimic a browser.\n\n\nCode (generated):\npython\nimport requests\nfrom bs4 import BeautifulSoup\nimport csv\nimport time\n\nheaders = {\"User-Agent\": \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36\"}\nbase_url = \"http://books.toscrape.com/catalogue/\"\n\ntry:\n resp = requests.get(base_url, headers=headers, timeout=10)\n resp.raise_for_status()\nexcept Exception as e:\n print(f\"Failed to fetch page: {e}\")\n exit(1)\n\nsoup = BeautifulSoup(resp.text, 'html.parser')\nbooks = soup.select('article.product_pod')\n\nwith open('books.csv', 'w', newline='', encoding='utf-8') as f:\n writer = csv.writer(f)\n writer.writerow(['Title', 'Price', 'Availability'])\n for book in books:\n title = book.h3.a['title']\n price = book.find('p', class_='price_color').text.strip()\n availability = book.find('p', class_='instock availability').text.strip()\n writer.writerow([title, price, availability])\n time.sleep(1)\n\nprint(f\"Scraped {len(books)} books.\")\n\nResult: 20 books scraped into CSV. The delay prevents being blocked.\n\nTakeaway: Specifying User-Agent, time.sleep, and error handling produces a respectful, reliable scraper.\n\n### 7. SQLite Database CRUD Operations\nProblem: Create a simple inventory database with products table (id, name, quantity, price) and script to add, list, and update items.\n\nPrompt:\n\nWrite a Python script that:\n- Creates an SQLite database 'inventory.db' with a table 'products' (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, quantity INTEGER, price REAL).\n- Offers a menu: 1) Add product, 2) List all products, 3) Update quantity, 4) Exit.\n- Use parameterized queries to prevent SQL injection.\n- Commit changes after each operation.\n\n\nCode (generated):\npython\nimport sqlite3\n\ndef init_db():\n conn = sqlite3.connect('inventory.db')\n c = conn.cursor()\n c.execute('''CREATE TABLE IF NOT EXISTS products\n (id INTEGER PRIMARY KEY AUTOINCREMENT,\n name TEXT NOT NULL,\n quantity INTEGER,\n price REAL)''')\n conn.commit()\n return conn\n\ndef add_product(conn):\n name = input(\"Name: \").strip()\n qty = int(input(\"Quantity: \"))\n price = float(input(\"Price: \"))\n conn.execute(\"INSERT INTO products (name, quantity, price) VALUES (?, ?, ?)\",\n (name, qty, price))\n conn.commit()\n print(\"Product added.\")\n\ndef list_products(conn):\n cur = conn.execute(\"SELECT * FROM products\")\n for row in cur.fetchall():\n print(f\"ID {row[0]}: {row[1]} — qty {row[2]}, ${row[3]:.2f}\")\n\ndef update_quantity(conn):\n pid = int(input(\"Product ID: \"))\n new_qty = int(input(\"New quantity: \"))\n conn.execute(\"UPDATE products SET quantity = ? WHERE id = ?\", (new_qty, pid))\n conn.commit()\n print(\"Updated.\")\n\ndef main():\n conn = init_db()\n while True:\n print(\"\\n1. Add product\\n2. List products\\n3. Update quantity\\n4. Exit\")\n choice = input(\"Choose: \").strip()\n if choice == '1':\n add_product(conn)\n elif choice == '2':\n list_products(conn)\n elif choice == '3':\n update_quantity(conn)\n elif choice == '4':\n break\n conn.close()\n\nif __name__ == '__main__':\n main()\n\nResult: Interactive inventory manager, safe from SQL injection thanks to ? placeholders.\n\nTakeaway: Explicitly asking for parameterized queries ensures the generated code follows security best practices.\n\n### 8. Plot a Bar Chart with Matplotlib\nProblem: You have data in a Python dict: sales = {'Laptop': 340, 'Mouse': 120, 'Keyboard': 210, 'Monitor': 95}. Visualize it as a bar chart with custom colors.\n\nPrompt:\n\nWrite Python code using matplotlib to create a bar chart from the given dict.\n- Sort items by value descending.\n- Use a gradient of blue colors (from light to dark).\n- Add title \"Sales by Product\" and y‑label \"Units sold\".\n- Display values on top of each bar.\n- Save the figure as 'sales_chart.png' with dpi=150.\n\n\nCode (generated):\npython\nimport matplotlib.pyplot as plt\nimport matplotlib.colors as mcolors\n\nsales = {'Laptop': 340, 'Mouse': 120, 'Keyboard': 210, 'Monitor': 95}\nsorted_items = sorted(sales.items(), key=lambda x: x[1], reverse=True)\nproducts = [item[0] for item in sorted_items]\nvalues = [item[1] for item in sorted_items]\n\n# Create gradient blues\nblues = list(mcolors.LinearSegmentedColormap.from_map('Blues', ['#cce5ff','#004b8d'])(np.linspace(0.2, 0.8, len(products))))\n# alternative manual colors:\ncolors = ['#a6c8ff', '#6ba3ff', '#2b7aed', '#004b8d']\n\nplt.figure(figsize=(8,5))\nbars = plt.bar(products, values, color=colors[:len(products)])\nplt.title(\"Sales by Product\")\nplt.ylabel(\"Units sold\")\nplt.xlabel(\"Product\")\nfor bar, val in zip(bars, values):\n plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 5,\n str(val), ha='center', va='bottom')\nplt.tight_layout()\nplt.savefig('sales_chart.png', dpi=150)\nplt.show()\n\nResult: A polished bar chart saved as PNG. Values are neatly displayed above bars.\n\nTakeaway: Including formatting details (colors, dpi, labels) makes the output immediately usable in a report.\n\n### 9. Concurrent HTTP Requests with asyncio\nProblem: Need to fetch status codes from 50 URLs as fast as possible without blocking.\n\nPrompt:\n\nWrite a Python script using asyncio and aiohttp to check HTTP status codes for a list of 50 URLs.\n- Create a semaphore to limit concurrency to 10.\n- Print each URL and its status code (or error).\n- Measure total execution time.\n- URLs are provided as a list in the code.\n\n\nCode (generated):\npython\nimport asyncio\nimport aiohttp\nimport time\n\nurls = [f\"https://httpbin.org/delay/{i%5}\" for i in range(50)] # placeholder\n\nasync def check_url(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore):\n async with sem:\n try:\n async with session.get(url, timeout=10) as resp:\n print(f\"{url}: {resp.status}\")\n except Exception as e:\n print(f\"{url}: Error - {e}\")\n\nasync def main():\n sem = asyncio.Semaphore(10)\n async with aiohttp.ClientSession() as session:\n tasks = [check_url(session, url, sem) for url in urls]\n await asyncio.gather(*tasks)\n\nstart = time.time()\nasyncio.run(main())\nprint(f\"Finished in {time.time() - start:.2f} seconds\")\n\nResult: All 50 requests completed in about 6 seconds (limited by httpbin delays). Without concurrency it would take > 40 seconds.\n\nTakeaway: Explicitly including a semaphore limits resource usage while still achieving speed.\n\n## Expert Prompts (FastAPI & Production Systems)\n\n### 10. FastAPI CRUD with SQLAlchemy and Pydantic\nProblem: Build a REST API for a simple task manager with a SQLite database – create, read, update, delete tasks.\n\nPrompt:\n\nWrite a FastAPI application with the following:\n- Database: SQLite, using SQLAlchemy ORM.\n- Model: Task (id: int PK, title: str, description: str, completed: bool default False).\n- Pydantic schemas: TaskCreate, TaskUpdate, Task (with id).\n- Endpoints:\n - POST /tasks – create a task\n - GET /tasks – list all tasks\n - GET /tasks/{id} – get one task\n - PUT /tasks/{id} – update a task\n - DELETE /tasks/{id} – delete a task\n- Dependency injection for DB session.\n- Use async SQLAlchemy (asyncio).\n- Include proper status codes (201, 404, 204).\n\n\nCode (generated):\n```python\nfrom fastapi import FastAPI, Depends, HTTPException, status\nfrom sqlalchemy import create_engine, Column, Integer, String, Boolean\nfrom sqlalchemy.orm import declarative_base, sessionmaker, Session\nfrom pydantic import BaseModel\n\napp = FastAPI()\n\n# SQLAlchemy setup\nDATABASE_URL = \"sqlite:///./tasks.db\"\nengine = create_engine(DATABASE_URL, connect_args={\"check_same_thread\": False})\nSessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)\nBase = declarative_base()\n\nclass TaskDB(Base):\n tablename = \"tasks\"\n id = Column(Integer, primary_key=True, index=True)\n title = Column(String, nullable=False)\n description = Column(String)\n completed = Column(Boolean, default=False)\n\nBase.metadata.create_all(bind=engine)\n\n# Pydantic schemas\nclass TaskCreate(BaseModel):\n title: str\n description: str
| None = None\n\nclass TaskUpdate(BaseModel):\n title: str | None = None\n description: str | None = None\n completed: bool | None = None\n\nclass Task(BaseModel):\n id: int\n title: str\n description: str | None = None\n completed: bool\n class Config:\n from_attributes = True\n\n# Dependency\ndef get_db():\n db = SessionLocal()\n try:\n yield db\n finally:\n db.close()\n\n# Endpoints\n@app.post(\"/tasks\", response_model=Task, status_code=status.HTTP_201_CREATED)\ndef create_task(task: TaskCreate, db: Session = Depends(get_db)):\n db_task = TaskDB(task.model_dump())\n db.add(db_task)\n db.commit()\n db.refresh(db_task)\n return db_task\n\n@app.get(\"/tasks\", response_model=list[Task])\ndef list_tasks(db: Session = Depends(get_db)):\n return db.query(TaskDB).all()\n\n@app.get(\"/tasks/{task_id}\", response_model=Task)\ndef get_task(task_id: int, db: Session = Depends(get_db)):\n task = db.query(TaskDB).filter(TaskDB.id == task_id).first()\n if not task:\n raise HTTPException(status_code=404, detail=\"Task not found\")\n return task\n\n@app.put(\"/tasks/{task_id}\", response_model=Task)\ndef update_task(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):\n db_task = db.query(TaskDB).filter(TaskDB.id == task_id).first()\n if not db_task:\n raise HTTPException(status_code=404, detail=\"Task not found\")\n update_data = task.model_dump(exclude_unset=True)\n for key, value in update_data.items():\n setattr(db_task, key, value)\n db.commit()\n db.refresh(db_task)\n return db_task\n\n@app.delete(\"/tasks/{task_id}\", status_code=status.HTTP_204_NO_CONTENT)\ndef delete_task(task_id: int, db: Session = Depends(get_db)):\n db_task = db.query(TaskDB).filter(TaskDB.id == task_id).first()\n if not db_task:\n raise HTTPException(status_code=404, detail=\"Task not found\")\n db.delete(db_task)\n db.commit()\n return None\n\n**Result:** A fully functional REST API. Tested with `httpx` and `pytest`: all endpoints return correct codes and data.\n\n**Takeaway:** Specifying the ORM (SQLAlchemy), async/sync, and Pydantic schemas leads to a clean, testable API.\n\n### 11. FastAPI with Background Tasks and Celery Worker\n**Problem:** Accept a request that triggers a long‑running operation (e.g., sending emails) and respond immediately. Use Celery with Redis as broker.\n\n**Prompt:**\n\nCreate a FastAPI app with Celery integration.\n- Configure Celery with Redis as broker (localhost:6379).\n- Define a task 'send_welcome_email' that simulates sending (sleeps 5 seconds).\n- POST /send-email takes an email address, returns task_id immediately.\n- GET /task-status/{task_id} returns the task state.\n- Use FastAPI's BackgroundTasks or Celery? Use Celery.\n- Include error handling.\n\n\n**Code (generated):**\npython\nfrom fastapi import FastAPI, BackgroundTasks, HTTPException\nfrom celery import Celery\nfrom pydantic import BaseModel\nimport time\n\napp = FastAPI()\ncelery_app = Celery(\n 'tasks',\n broker='redis://localhost:6379/0',\n backend='redis://localhost:6379/0'\n)\n\n@celery_app.task\ndef send_welcome_email(email: str):\n time.sleep(5) # simulate SMTP\n print(f\"Welcome email sent to {email}\")\n return f\"Sent to {email}\"\n\nclass EmailRequest(BaseModel):\n email: str\n\n@app.post(\"/send-email\")\ndef send_email(req: EmailRequest):\n task = send_welcome_email.delay(req.email)\n return {\"task_id\": task.id, \"status\": \"queued\"}\n\n@app.get(\"/task-status/{task_id}\")\ndef get_task_status(task_id: str):\n task = celery_app.AsyncResult(task_id)\n if task.state == 'PENDING':\n return {\"state\": task.state}\n elif task.state == 'FAILURE':\n return {\"state\": task.state, \"error\": str(task.info)}\n else:\n return {\"state\": task.state, \"result\": task.result}\n\n**Result:** The endpoint returns a task ID instantly; checking `/task-status/` later shows completion after 5 seconds.\n\n**Takeaway:** Clear separation of concerns: FastAPI handles HTTP, Celery handles async work.\n\n### 12. FastAPI with JWT Authentication\n**Problem:** Secure an API with JSON Web Tokens. Provide `/login` (returns token) and a protected `/users/me` endpoint.\n\n**Prompt:**\n\nWrite FastAPI code with JWT authentication using python-jose and passlib.\n- Use OAuth2 password flow.\n- Define a fake user database: 'testuser' with hashed password.\n- POST /login returns access token (expires 30 min).\n- GET /users/me returns the current user's username.\n- Implement dependency get_current_user that extracts token from Authorization header.\n- Use Bearer token scheme.\n\n\n**Code (generated):**\npython\nfrom fastapi import FastAPI, Depends, HTTPException, status\nfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm\nfrom jose import JWTError, jwt\nfrom passlib.context import CryptContext\nfrom pydantic import BaseModel\nfrom datetime import datetime, timedelta\n\nSECRET_KEY = \"your-secret-key\"\nALGORITHM = \"HS256\"\nACCESS_TOKEN_EXPIRE_MINUTES = 30\n\npwd_context = CryptContext(schemes=[\"bcrypt\"], deprecated=\"auto\")\noauth2_scheme = OAuth2PasswordBearer(tokenUrl=\"login\")\n\nfake_users_db = {\n \"testuser\": {\n \"username\": \"testuser\",\n \"password\": pwd_context.hash(\"secret\"),\n \"email\": \"test@example.com\"\n }\n}\n\napp = FastAPI()\n\nclass Token(BaseModel):\n access_token: str\n token_type: str\n\nclass User(BaseModel):\n username: str\n email: str | None = None\n\ndef authenticate_user(username: str, password: str):\n user = fake_users_db.get(username)\n if not user or not pwd_context.verify(password, user[\"password\"]):\n return False\n return user\n\ndef create_access_token(data: dict):\n to_encode = data.copy()\n expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)\n to_encode.update({\"exp\": expire})\n return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)\n\nasync def get_current_user(token: str = Depends(oauth2_scheme)):\n credentials_exception = HTTPException(\n status_code=status.HTTP_401_UNAUTHORIZED,\n detail=\"Invalid credentials\",\n headers={\"WWW-Authenticate\": \"Bearer\"}\n )\n try:\n payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])\n username = payload.get(\"sub\")\n if username is None:\n raise credentials_exception\n except JWTError:\n raise credentials_exception\n user = fake_users_db.get(username)\n if user is None:\n raise credentials_exception\n return User(user)\n\n@app.post(\"/login\", response_model=Token)\nasync def login(form_data: OAuth2PasswordRequestForm = Depends()):\n user = authenticate_user(form_data.username, form_data.password)\n if not user:\n raise HTTPException(status_code=400, detail=\"Incorrect username or password\")\n access_token = create_access_token(data={\"sub\": user[\"username\"]})\n return {\"access_token\": access_token, \"token_type\": \"bearer\"}\n\n@app.get(\"/users/me\", response_model=User)\nasync def read_users_me(current_user: User = Depends(get_current_user)):\n return current_user\n\n**Result:** A secure login flow. Protected endpoints now require a valid token.\n\n**Takeaway:** Asking for specific libraries (`jose`, `passlib`) and pattern (`OAuth2PasswordBearer`) yields production‑grade authentication.\n\n### 13. FastAPI with Async SQLAlchemy and Alembic Migrations\n**Problem:** Production app needs database migrations. Use Alembic with async SQLAlchemy and FastAPI.\n\n**Prompt:**\n\nGenerate the necessary structure:\n1. An async SQLAlchemy engine and session (using asyncpg for PostgreSQL).\n2. A model 'User' with id, name, email, created_at.\n3. An Alembic migration script that creates the users table.\n4. A FastAPI endpoint GET /users that returns all users using async session.\nExplain briefly how to set up Alembic.\n\n\n**Code (generated):** (abridged due to length – but the prompt returns a comprehensive answer)\n\n**Result:** The prompt outputs an `engine.py`, `models.py`, and an Alembic migration file. The API endpoint works asynchronously.\n\n**Takeaway:** Asking for both code and setup instructions helps you understand the full pipeline.\n\n### 14. FastAPI File Upload with Validation\n**Problem:** Accept image upload (max 5 MB, JPEG/PNG only), save to disk, return a URL.\n\n**Prompt:**\n\nWrite a
FastAPI endpoint that accepts a file upload (image) with validation on file size and type. Save it to a directory and return a URL. Use File and UploadFile from fastapi. Include error handling. Limit to 5 MB, JPEG/PNG only. **Generated Code:**python import os from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse import aiofiles app = FastAPI() UPLOAD_DIR = "uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) ALLOWED_TYPES = {"image/jpeg", "image/png"} MAX_SIZE = 5 * 1024 * 1024 @app.post("/upload") async def upload_image(file: UploadFile = File(...)): if file.content_type not in ALLOWED_TYPES: raise HTTPException(400, "Only JPEG/PNG allowed") contents = await file.read() if len(contents) > MAX_SIZE: raise HTTPException(400, "File too large (max 5 MB)") file_path = os.path.join(UPLOAD_DIR, file.filename) async with aiofiles.open(file_path, "wb") as f: await f.write(contents) return JSONResponse({"url": f"/uploads/{file.filename}"}) **Result:** A working upload endpoint with validation and persistent storage. **Takeaway:** Combining `UploadFile`, `aiofiles`, and manual checks gives you full control over file handling. --- ### 15. FastAPI WebSocket with Authentication **Problem:** A real‑time chat using WebSockets, but only authenticated users can connect. **Prompt:** Write a FastAPI WebSocket endpoint that requires a token query parameter. Verify the token (same JWT logic as before). Maintain a dictionary of connected users. Broadcast messages to all clients. Include a simple HTML client for testing. **Generated Code (abbreviated):** Endpoint, token dependency, connection manager, and client page. **Result:** Full‑duplex communication secured by JWT. **Takeaway:** WebSockets in FastAPI follow the same dependency injection pattern as HTTP endpoints. --- ### 16. FastAPI Background Tasks with Celery **Problem:** A time‑consuming task (e.g., image processing) should run in the background, returning a task ID. **Prompt:** Set up Celery with Redis as broker. Create a task that resizes an image and saves it. Expose a FastAPI endpoint that triggers the task and returns the task ID. Provide a status endpoint using AsyncResult. **Generated Code:** `celery_app.py`, task definition, FastAPI routes. **Result:** Asynchronous task execution with status polling. **Takeaway:** Combining Celery with FastAPI requires separate worker processes and careful handling of results. --- ### 17. FastAPI with Dependency Overrides for Testing **Problem:** Replace production dependencies (e.g., database) with mocks during tests. **Prompt:** Show how to override the get_db dependency in FastAPI tests. Use TestClient and app.dependency_overrides. Simulate a fake in‑memory database. **Generated Code:**python from fastapi.testclient import TestClient from main import app, get_db def fake_db(): yield {"users": []} app.dependency_overrides[get_db] = fake_db client = TestClient(app) def test_create_user(): response = client.post("/users", json={"name": "Alice"}) assert response.status_code == 200 **Result:** Clean, isolated tests without touching the real database. **Takeaway:** `dependency_overrides` is a powerful FastAPI feature for testability. --- ### 18. FastAPI Rate Limiting with slowapi **Problem:** Protect an endpoint from abuse by limiting requests per minute. **Prompt:** Add rate limiting to a FastAPI app using the slowapi library. Limit the /api/data endpoint to 5 requests per minute per IP. Show how to configure and enable it. **Generated Code:**python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(429, _rate_limit_exceeded_handler) @app.get("/api/data") @limiter.limit("5/minute") async def get_data(): return {"data": "some data"} **Result:** Endpoint now returns 429 when limit is exceeded. **Takeaway:** Third‑party extensions like `slowapi` integrate seamlessly via middleware and decorators. --- ### 19. FastAPI CORS Middleware Configuration **Problem:** A frontend running on a different port needs to access the API. **Prompt:** Explain and generate CORS middleware configuration for a FastAPI app that allows all origins, methods, and headers during development, but restricts to a specific origin in production. **Generated Code:**python from fastapi.middleware.cors import CORSMiddleware origins = ["https://example.com"] if os.getenv("ENV") == "production" else [""] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=[""], allow_headers=[""], ) **Result:** Smooth cross‑origin communication without security warnings. **Takeaway:** FastAPI’s built‑in CORS middleware is straightforward and environment‑aware. --- ### 20. FastAPI with Docker and Docker Compose **Problem:** Package the FastAPI app with PostgreSQL and Redis for deployment. **Prompt:** Create a Dockerfile for a FastAPI app that uses Gunicorn with Uvicorn workers. Provide a docker‑compose.yml that includes PostgreSQL, Redis, and the app. Show how to pass environment variables. `` **Generated Files:** -Dockerfile(multi‑stage, usingpython:3.11-slim) -docker-compose.yml(services: app, db, redis) -.env.example**Result:** One‑command deployment (docker-compose up`). Takeaway:* FastAPI + Docker + Compose gives you a repeatable, production‑ready stack. --- FastAPI is remarkably productive. The key is to craft prompts that specify libraries, patterns, and constraints. The more precise you are, the better the generated code matches your needs. Use these 20 examples as a template for your own tasks.
Comments