12 Prompts for Generating Python Code: From Simple Scripts to FastAPI

Write a Python script that renames all .jpg files in the current directory. Add the current date (YYYY-MM-DD) as a prefix to each filename, separated by an underscore. Skip files that already have the prefix. Use pathlib and datetime.

from pathlib import Path
from datetime import date

today = date.today().isoformat()
for file in Path('.').glob('*.jpg'):
    if not file.name.startswith(today):
        new_name = f'{today}_{file.name}'
        file.rename(file.with_name(new_name))

Write a Python function that takes a CSV file path as input. It should remove duplicate rows based on all columns, parse the 'date' column (if present) into YYYY-MM-DD format, and save the cleaned data to a new CSV. Use pandas.

import pandas as pd

def clean_csv(input_path: str, output_path: str = 'cleaned.csv'):
    df = pd.read_csv(input_path)
    df = df.drop_duplicates()
    if 'date' in df.columns:
        df['date'] = pd.to_datetime(df['date'], errors='coerce').dt.strftime('%Y-%m-%d')
    df.to_csv(output_path, index=False)

Write a Python script that fetches https://example.com/blog, finds all <h2> elements with class 'post-title', and prints each title and its <a> href. Use requests and BeautifulSoup.

import requests
from bs4 import BeautifulSoup

response = requests.get('https://example.com/blog')
soup = BeautifulSoup(response.text, 'html.parser')
for h2 in soup.select('h2.post-title'):
    link = h2.find('a')
    if link:
        print(f'{link.text.strip()} -> {link.get("href")}')

Write a script that reads 'config.json' and writes its content to 'config.yaml' preserving nested structures. Use Python's json and yaml libraries.

import json
import yaml

with open('config.json') as f:
    data = json.load(f)

with open('config.yaml', 'w') as f:
    yaml.dump(data, f, default_flow_style=False)

Create a Python CLI tool using argparse with three subcommands: add (takes a task string), list (prints all tasks), and done (takes a task ID). Store tasks in 'tasks.json' with fields: id, text, done (boolean). Use a simple monotonic counter for IDs.

import argparse
import json
from pathlib import Path

TASKS_FILE = Path('tasks.json')

def load_tasks():
    if TASKS_FILE.exists():
        return json.loads(TASKS_FILE.read_text())
    return []

def save_tasks(tasks):
    TASKS_FILE.write_text(json.dumps(tasks, indent=2))

def add_task(text):
    tasks = load_tasks()
    new_id = max((t['id'] for t in tasks), default=0) + 1
    tasks.append({'id': new_id, 'text': text, 'done': False})
    save_tasks(tasks)

def list_tasks():
    tasks = load_tasks()
    for t in tasks:
        status = '[x]' if t['done'] else '[ ]'
        print(f"{t['id']}: {status} {t['text']}")

def done_task(task_id):
    tasks = load_tasks()
    for t in tasks:
        if t['id'] == task_id:
            t['done'] = True
            break
    save_tasks(tasks)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='command')

    add_parser = subparsers.add_parser('add')
    add_parser.add_argument('task')

    subparsers.add_parser('list')

    done_parser = subparsers.add_parser('done')
    done_parser.add_argument('id', type=int)

    args = parser.parse_args()
    if args.command == 'add':
        add_task(args.task)
    elif args.command == 'list':
        list_tasks()
    elif args.command == 'done':
        done_task(args.id)

Write a Python script that reads 'access.log' (Apache Common Log Format). Count unique IPs, status codes, and top 5 requested paths. Print the results. Use only standard library modules (re, collections).

import re
from collections import Counter

pattern = r'(\S+) .*? \[.*?\] ".*? (\S+).*?" (\d+)'
ips = []
paths = []
statuses = []

with open('access.log') as f:
    for line in f:
        match = re.search(pattern, line)
        if match:
            ips.append(match.group(1))
            paths.append(match.group(2))
            statuses.append(match.group(3))

print(f'Unique IPs: {len(set(ips))}')
print(f'Status codes: {dict(Counter(statuses))}')
print(f'Top 5 paths: {Counter(paths).most_common(5)}')

Write an async Python script using aiohttp that takes a list of URLs from a file 'urls.txt', fetches each concurrently (max 5 simultaneous), and saves the content to files named after the URL's last path segment. Handle timeouts (10 seconds) and non-200 status codes gracefully.

import asyncio
import aiohttp
from urllib.parse import urlparse

semaphore = asyncio.Semaphore(5)

async def fetch_and_save(session, url):
    async with semaphore:
        try:
            async with session.get(url, timeout=10) as resp:
                if resp.status == 200:
                    content = await resp.text()
                    filename = urlparse(url).path.split('/')[-1] or 'index.html'
                    with open(filename, 'w') as f:
                        f.write(content)
                    print(f'Saved {filename}')
                else:
                    print(f'Failed {url}: {resp.status}')
        except Exception as e:
            print(f'Error {url}: {e}')

async def main():
    with open('urls.txt') as f:
        urls = [line.strip() for line in f if line.strip()]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_and_save(session, u) for u in urls]
        await asyncio.gather(*tasks)

asyncio.run(main())

Write a script that reads 'sales.csv' (columns: date, product, amount) and inserts rows into a SQLite database 'sales.db'. Create the table if it doesn't exist. Use csv and sqlite3 standard libraries.

import csv
import sqlite3

conn = sqlite3.connect('sales.db')
cursor = conn.cursor()
cursor.execute('''
    CREATE TABLE IF NOT EXISTS sales (
        date TEXT,
        product TEXT,
        amount REAL
    )
''')

with open('sales.csv') as f:
    reader = csv.DictReader(f)
    for row in reader:
        cursor.execute(
            'INSERT INTO sales (date, product, amount) VALUES (?, ?, ?)',
            (row['date'], row['product'], float(row['amount']))
        )
conn.commit()
conn.close()

Create a FastAPI application with two models: Post (id, title, content, created_at) and Comment (id, post_id, author, text). Implement CRUD endpoints for both using async SQLAlchemy 2.0 with SQLite (aiosqlite). Include proper dependency injection for the database session.

# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime

class Base(DeclarativeBase):
    pass

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)

class Comment(Base):
    __tablename__ = 'comments'
    id = Column(Integer, primary_key=True)
    post_id = Column(Integer, ForeignKey('posts.id'))
    author = Column(String(50))
    text = Column(Text)

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

DATABASE_URL = 'sqlite+aiosqlite:///./blog.db'
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.post('/posts/')
async def create_post(title: str, content: str, db: AsyncSession = Depends(get_db)):
    post = Post(title=title, content=content)
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

@app.get('/posts/')
async def get_posts(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Post))
    return result.scalars().all()

Extend a FastAPI app with user authentication. Create a User model (id, username, hashed_password). Add endpoints: POST /register (hash password with passlib bcrypt), POST /login (verify password, return JWT token using python-jose). Protect a GET /me endpoint that returns the current user from the token.

from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta

pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
SECRET_KEY = 'your-secret-key'
ALGORITHM = 'HS256'

def create_access_token(data: dict, expires_delta: timedelta = timedelta(hours=24)):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({'exp': expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post('/register')
async def register(username: str, password: str, db: AsyncSession = Depends(get_db)):
    hashed = pwd_context.hash(password)
    user = User(username=username, hashed_password=hashed)
    db.add(user)
    await db.commit()
    return {'msg': 'created'}

@app.post('/login')
async def login(username: str, password: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.username == username))
    user = result.scalar_one_or_none()
    if not user or not pwd_context.verify(password, user.hashed_password):
        raise HTTPException(status_code=401, detail='Invalid credentials')
    token = create_access_token({'sub': user.username})
    return {'access_token': token, 'token_type': 'bearer'}

Create a FastAPI endpoint POST /resize that accepts an image URL and target width. Submit a Celery task that downloads the image, resizes it using Pillow, and saves the result to 'processed/' folder. Return the task ID immediately. Use Redis as broker.

# tasks.py
from celery import Celery
from PIL import Image
import requests
from io import BytesIO

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

@app.task
def resize_image(url: str, width: int):
    response = requests.get(url)
    img = Image.open(BytesIO(response.content))
    ratio = width / img.width
    height = int(img.height * ratio)
    img = img.resize((width, height))
    filename = url.split('/')[-1]
    img.save(f'processed/{filename}')
    return filename

# main.py
from fastapi import FastAPI
from tasks import resize_image

app = FastAPI()

@app.post('/resize')
async def resize(url: str, width: int = 300):
    task = resize_image.delay(url, width)
    return {'task_id': task.id, 'status': 'processing'}

Create a FastAPI WebSocket endpoint at /ws/{room_name}. Clients can join a room by connecting. When a client sends a message, broadcast it to all other clients in the same room. Include a simple HTML page for testing. Use a dictionary to store room connections.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
rooms = {}

@app.websocket('/ws/{room_name}')
async def websocket_endpoint(websocket: WebSocket, room_name: str):
    await websocket.accept()
    if room_name not in rooms:
        rooms[room_name] = []
    rooms[room_name].append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for client in rooms[room_name]:
                if client != websocket:
                    await client.send_text(f'User: {data}')
    except WebSocketDisconnect:
        rooms[room_name].remove(websocket)