Промты для генерации Python кода: от скриптов до FastAPI

{
"title": "15 Prompts for Python Code Generation: From Simple Scripts to Production FastAPI Apps",
"content": "## Introduction\n\nPython remains one of the most versatile programming languages in 2026, powering everything from quick automation scripts to enterprise-grade APIs. But writing high-quality Python code consistently takes time and expertise. That’s where prompt engineering for code generation comes in. By crafting precise prompts for large language models (LLMs) like GPT-4, Claude, or specialized code assistants, you can generate Python code that is not only functional but also follows best practices — including type hints, async patterns, and proper error handling.\n\nThis article provides 15 carefully designed prompts organized into three levels: basic scripts, advanced patterns, and expert-level FastAPI production code. Each prompt includes the task, the exact prompt text, and an example result with code. Whether you’re a data analyst automating Excel reports or a backend engineer building a microservice, these prompts will save you hours of boilerplate and debugging.\n\n## Why Prompt Engineering Matters for Python\n\nAccording to a 2025 Stack Overflow survey, over 48% of developers now use AI coding assistants regularly. But the quality of generated code heavily depends on how you phrase your request. Vague prompts like “write a Python script to parse CSV” often produce code without error handling, type annotations, or performance considerations. A well-structured prompt includes:\n\n- Context: what the code should do and its environment\n- Constraints: Python version, dependencies, style guide\n- Expectations: input/output format, edge cases\n- Examples: a sample input and expected output\n\nThis approach, known as “chain-of-thought prompting” in the LLM literature (Wei et al., 2022), significantly improves code correctness and readability.\n\n## Basic Prompts: Quick Scripts for Everyday Tasks\n\nThese prompts are ideal for data analysts, system administrators, and anyone automating repetitive tasks. They focus on clarity, error handling, and minimal dependencies.\n\n### 1. CSV to JSON Converter with Validation\n\nTask: Generate a script that reads a CSV file, validates required columns, and outputs JSON.\n\nPrompt:\n\nWrite a Python 3.11 script that:\n1. Reads a CSV file from a given path (use argparse for the file path)\n2. Validates that the CSV contains at least these columns: 'id', 'name', 'email'\n3. If validation fails, print a clear error message with missing columns\n4. If validation passes, convert each row to a JSON object and write to an output file (also specified via argparse)\n5. Use csv.DictReader for reading\n6. Add type hints for all functions\n7. Handle FileNotFoundError gracefully\n\nExample input CSV:\nid,name,email\n1,Alice,alice@example.com\n2,Bob,bob@example.com\n\nExpected output JSON:\n[\n {\"id\": \"1\", \"name\": \"Alice\", \"email\": \"alice@example.com\"},\n {\"id\": \"2\", \"name\": \"Bob\", \"email\": \"bob@example.com\"}\n]\n\n\nExample result:\npython\nimport argparse\nimport csv\nimport json\nfrom pathlib import Path\nfrom typing import List, Dict\n\ndef validate_columns(headers: List[str], required: List[str]) -> bool:\n missing = [col for col in required if col not in headers]\n if missing:\n print(f\"Validation error: missing columns {missing}\")\n return False\n return True\n\ndef convert_csv_to_json(input_path: Path, output_path: Path) -> None:\n try:\n with open(input_path, 'r', newline='') as csvfile:\n reader = csv.DictReader(csvfile)\n if not validate_columns(reader.fieldnames, ['id', 'name', 'email']):\n return\n rows: List[Dict[str, str]] = [row for row in reader]\n with open(output_path, 'w') as jsonfile:\n json.dump(rows, jsonfile, indent=2)\n print(f\"Successfully wrote {len(rows)} records to {output_path}\")\n except FileNotFoundError:\n print(f\"Error: file {input_path} not found\")\n except Exception as e:\n print(f\"Unexpected error: {e}\")\n\nif __name__ == \"__main__\":\n parser = argparse.ArgumentParser(description=\"Convert CSV to JSON\")\n parser.add_argument(\"input\", type=Path, help=\"Path to input CSV\")\n parser.add_argument(\"output\", type=Path, help=\"Path to output JSON\")\n args = parser.parse_args()\n convert_csv_to_json(args.input, args.output)\n\n\n### 2. Web Scraper with Rate Limiting\n\nTask: Build a script that scrapes product prices from a list of URLs, respects robots.txt, and adds random delays.\n\nPrompt:\n\nCreate a Python 3.11 script that:\n1. Takes a file with one URL per line as input\n2. For each URL, fetches the page using httpx (not requests)\n3. Checks robots.txt using urllib.robotparser; skip URL if disallowed\n4. Extracts all text inside <h1> and <p class=\"price\"> using BeautifulSoup4\n5. Prints URL, title, and first price found\n6. Waits a random time between 2 and 5 seconds between requests\n7. Uses asyncio and httpx.AsyncClient for concurrent fetching (max 3 concurrent)\n8. Catches HTTP errors and timeouts gracefully\n\nProvide full code with imports and main block.\n\n\nExample result:\npython\nimport asyncio\nimport random\nimport httpx\nfrom bs4 import BeautifulSoup\nfrom urllib.robotparser import RobotFileParser\nfrom urllib.parse import urlparse\nfrom typing import List, Tuple, Optional\n\nasync def check_robots(url: str) -> bool:\n parsed = urlparse(url)\n robots_url = f\"{parsed.scheme}://{parsed.netloc}/robots.txt\"\n rp = RobotFileParser()\n try:\n rp.set_url(robots_url)\n rp.read()\n return rp.can_fetch(\"MyScraper/1.0\", url)\n except Exception:\n return True # assume allowed on error\n\nasync def fetch_price(client: httpx.AsyncClient, url: str) -> Tuple[str, Optional[str], Optional[str]]:\n try:\n resp = await client.get(url, timeout=10.0)\n resp.raise_for_status()\n soup = BeautifulSoup(resp.text, 'html.parser')\n title = soup.find('h1').get_text(strip=True) if soup.find('h1') else None\n price_tag = soup.find('p', class_='price')\n price = price_tag.get_text(strip=True) if price_tag else None\n return url, title, price\n except Exception as e:\n return url, None, f\"Error: {e}\"\n\nasync def scrape_urls(urls: List[str]) -> None:\n allowed_urls = []\n for u in urls:\n if await check_robots(u):\n allowed_urls.append(u)\n sem = asyncio.Semaphore(3)\n async def bounded_fetch(url: str) -> Tuple[str, Optional[str], Optional[str]]:\n async with sem:\n async with httpx.AsyncClient() as client:\n result = await fetch_price(client, url)\n await asyncio.sleep(random.uniform(2, 5))\n return result\n tasks = [bounded_fetch(url) for url in allowed_urls]\n results = await asyncio.gather(*tasks)\n for url, title, price in results:\n print(f\"URL: {url}\")\n print(f\"Title: {title}\")\n print(f\"Price: {price}\")\n print(\"---\")\n\nif __name__ == \"__main__\":\n import sys\n with open(sys.argv[1], 'r') as f:\n urls = [line.strip() for line in f if line.strip()]\n asyncio.run(scrape_urls(urls))\n\n\n### 3. Data Cleaner for Pandas\n\nTask: Generate a script that cleans a DataFrame: removes duplicates, fills missing values, and normalizes text columns.\n\nPrompt:\n\nWrite a Python script using pandas 2.0+ that:\n1. Reads a CSV file (path from command line)\n2. Drops duplicate rows based on a 'user_id' column\n3. For numeric columns, fill NaN with median\n4. For string columns, strip whitespace, convert to lowercase, and replace empty strings with 'unknown'\n5. Adds a column 'cleaned_at' with current timestamp\n6. Saves cleaned DataFrame to a new CSV file with suffix '_cleaned'\n7. Prints summary: original rows, removed duplicates, remaining rows\n\nInclude type hints and docstrings.\n\n\nExample result:\npython\nimport pandas as pd\nfrom pathlib import Path\nfrom datetime import datetime\nfrom typing import List\n\ndef clean_dataframe(df: pd.DataFrame) -> pd.DataFrame:\n \"\"\"Clean a DataFrame: remove duplicates, fill NaN, normalize strings.\"\"\"\n original_count = len(df)\n df = df.drop_duplicates(subset=['user_id'])\n dup_count = original_count - len(df)\n \n numeric_cols: List[str] = df.select_dtypes(include='number').columns.tolist()\n for col in numeric_cols:\n df[col] = df[col].fillna(df[col].median())\n \n string_cols: List[str] = df.select_dtypes(include='object').columns.tolist()\n for col in string_cols:\n df[col] = df[col].astype(str).str.strip().str.lower()\n df[col] = df[col].replace('', 'unknown')\n df[col] = df[col].replace('nan', 'unknown')\n \n df['cleaned_at'] = datetime.now().isoformat()\n print(f\"Original rows: {original_count}\")\n print(f\"Duplicates removed: {dup_count}\")\n print(f\"Remaining rows: {len(df)}\")\n return df\n\ndef main(input_path: Path) -> None:\n df = pd.read_csv(input_path)\n cleaned = clean_dataframe(df)\n output_path = input_path.with_stem(input_path.stem + '_cleaned')\n cleaned.to_csv(output_path, index=False)\n print(f\"Saved to {output_path}\")\n\nif __name__ == \"__main__\":\n import sys\n main(Path(sys.argv[1]))\n\n\n## Advanced Prompts: APIs, Async, and Production Patterns\n\nThese prompts target developers building backend services, microservices, or data pipelines with Python. They emphasize async/await, dependency injection, and testing.\n\n### 4. FastAPI CRUD with SQLAlchemy Async\n\nTask: Generate a FastAPI app with async SQLAlchemy for a simple User model.\n\nPrompt:\n\nGenerate a FastAPI application (Python 3.11+) with:\n1. SQLAlchemy 2.0 async session using asyncpg for PostgreSQL\n2. User model: id (UUID primary key), username (unique, indexed), email (unique), created_at (datetime, server default)\n3. CRUD endpoints:\n - POST /users (create user, return 201 with user)\n - GET /users/{id} (return user or 404)\n - GET /users (with pagination: skip and limit query params, default 10)\n - DELETE /users/{id} (return 204)\n4. Pydantic v2 schemas for request/response\n5. Dependency injection for session\n6. Proper error handling with HTTPException\n7. A main.py file that can run with uvicorn\n\nUse SQLAlchemy 2.0 style: async with async_session_maker() as session.\n\nDo not include Alembic migrations, just the model and CRUD.\n\n\nExample result (abbreviated):\npython\n# models.py\nimport uuid\nfrom datetime import datetime\nfrom sqlalchemy import Column, String, DateTime\nfrom sqlalchemy.dialects.postgresql import UUID\nfrom sqlalchemy.orm import DeclarativeBase\n\nclass Base(DeclarativeBase):\n pass\n\nclass User(Base):\n __tablename__ = \"users\"\n id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)\n username = Column(String(50), unique=True, index=True, nullable=False)\n email = Column(String(120), unique=True, nullable=False)\n created_at = Column(DateTime, server_default=func.now())\n\n# schemas.py\nfrom pydantic import BaseModel, EmailStr\nfrom uuid import UUID\nfrom datetime import datetime\n\nclass UserCreate(BaseModel):\n username: str\n email: EmailStr\n\nclass UserResponse(BaseModel):\n id: UUID\n username: str\n email: str\n created_at: datetime\n model_config = {\"from_attributes\": True}\n\n# main.py\nfrom fastapi import FastAPI, Depends, HTTPException, status\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker\nfrom typing import List\n\nDATABASE_URL = \"postgresql+asyncpg://user:pass@localhost/dbname\"\nengine = create_async_engine(DATABASE_URL)\nAsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)\n\napp = FastAPI()\n\nasync def get_db() -> AsyncSession:\n async with AsyncSessionLocal() as session:\n yield session\n\n@app.post(\"/users\", response_model=UserResponse, status_code=status.HTTP_201_CREATED)\nasync def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):\n new_user = User(**user.model_dump())\n db.add(new_user)\n await db.commit()\n await db.refresh(new_user)\n return new_user\n\n# ... other endpoints follow similar pattern\n\n\nNote: ASI Biont supports connecting to PostgreSQL databases via SQLAlchemy for building data integration pipelines — see details at asibiont.com/courses.\n\n### 5. Background Task Queue with Celery and Redis\n\nTask: Create a Celery worker that processes image resizing tasks with retries.\n\nPrompt:\n\nWrite a Python project with:\n1. Celery 5.3+ configured with Redis as broker\n2. A task called 'resize_image' that takes a source path and target size (tuple of width, height)\n3. Uses Pillow to resize the image, saves to a 'processed' subfolder\n4. Task has retry policy: max_retries=3, countdown=10 seconds\n5. Include a Celery beat schedule to clean up old files (older than 1 hour) every 30 minutes\n6. Provide a separate client.py that calls resize_image.delay()\n\nUse Python 3.11 and include requirements.txt.\n\n\nExample result:\npython\n# tasks.py\nfrom celery import Celery\nfrom PIL import Image\nfrom pathlib import Path\nimport os\nfrom datetime import datetime, timedelta\n\napp = Celery('tasks', broker='redis://localhost:6379/0')\n\napp.conf.beat_schedule = {\n 'cleanup-old-files': {\n 'task': 'tasks.cleanup_old_files',\n 'schedule': timedelta(minutes=30),\n },\n}\n\n@app.task(bind=True, max_retries=3, default_retry_delay=10)\ndef resize_image(self, source_path: str, target_size: tuple):\n try:\n img = Image.open(source_path)\n img_resized = img.resize(target_size)\n output_dir = Path('processed')\n output_dir.mkdir(exist_ok=True)\n output_path = output_dir / Path(source_path).name\n img_resized.save(output_path)\n return str(output_path)\n except Exception as exc:\n raise self.retry(exc=exc)\n\n@app.task\ndef cleanup_old_files():\n cutoff = datetime.now() - timedelta(hours=1)\n for f in Path('processed').glob('*'):\n if f.is_file() and datetime.fromtimestamp(f.stat().st_mtime) < cutoff:\n f.unlink()\n\n\n### 6. Async WebSocket Chat Server\n\nTask: Build a WebSocket chat server using FastAPI that broadcasts messages to all connected clients.\n\nPrompt:\n\nCreate a FastAPI application with WebSocket support for a simple chat room:\n1. On connect, server asks for username (first message from client)\n2. Server stores username in a connection manager\n3. Any subsequent message from a client is broadcast to all other connected clients in format \"{username}: {message}\"\n4. On disconnect, remove user and broadcast \"{username} left\"\n5. Connection manager should be thread-safe using asyncio.Lock\n6. Include a simple HTML client page served at GET / that connects to ws://host/chat\n\nUse Python 3.11, FastAPI, and websockets.\n\n\nExample result (manager class):\npython\nfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect\nfrom typing import Dict\nimport asyncio\n\napp = FastAPI()\n\nclass ConnectionManager:\n def __init__(self):\n self.active_connections: Dict[str, WebSocket] = {}\n self.lock = asyncio.Lock()\n\n async def connect(self, username: str, websocket: WebSocket):\n await websocket.accept()\n async with self.lock:\n self.active_connections[username] = websocket\n\n async def disconnect(self, username: str):\n async with self.lock:\n self.active_connections.pop(username, None)\n\n async def broadcast(self, message: str, exclude: str = None):\n async with self.lock:\n for user, ws in self.active_connections.items():\n if user != exclude:\n await ws.send_text(message)\n\nmanager = ConnectionManager()\n\n@app.websocket(\"/chat\")\nasync def chat_endpoint(websocket: WebSocket):\n username = await websocket.receive_text()\n await manager.connect(username, websocket)\n await manager.broadcast(f\"{username} joined\", exclude=username)\n try:\n while True:\n data = await websocket.receive_text()\n await manager.broadcast(f\"{username}: {data}\", exclude=username)\n except WebSocketDisconnect:\n await manager.disconnect(username)\n await manager.broadcast(f\"{username} left\")\n\n\n### 7. Unit Tests with Pytest and Mocking\n\nTask: Generate a comprehensive test suite for a simple calculator class.\n\nPrompt:\n\nWrite pytest tests for a Calculator class with methods: add, subtract, multiply, divide, power.\nThe Calculator class is in calculator.py. Requirements:\n1. Test normal cases\n2. Test edge cases: division by zero, negative numbers, large numbers\n3. Use pytest fixtures to create a calculator instance\n4. Use pytest.mark.parametrize for multiple test cases\n5. Mock external dependency: if divide method logs to a logger, mock the logger\n6. Include a conftest.py with shared fixtures\n7. Achieve >90% coverage\n\nProvide calculator.py first, then tests.\n\n\nExample result:\npython\n# calculator.py\nimport logging\n\nlogger = logging.getLogger(__name__)\n\nclass Calculator:\n def add(self, a: float, b: float) -> float:\n return a + b\n\n def divide(self, a: float, b: float) -> float:\n if b == 0:\n logger.error(\"Division by zero attempted\")\n raise ValueError(\"Cannot divide by zero\")\n return a / b\n\n# test_calculator.py\nimport pytest\nfrom calculator import Calculator\nfrom unittest.mock import patch\n\n@pytest.fixture\ndef calc():\n return Calculator()\n\n@pytest.mark.parametrize(\"a,b,expected\", [\n (1, 2, 3),\n (-1, 1, 0),\n (0, 0, 0),\n (1000000, 1, 1000001),\n])\ndef test_add(calc, a, b, expected):\n assert calc.add(a, b) == expected\n\n@patch('calculator.logger')\ndef test_divide_by_zero(mock_logger, calc):\n with pytest.raises(ValueError, match=\"Cannot divide by zero\"):\n calc.divide(10, 0)\n mock_logger.error.assert_called_once()\n\n\n## Expert Prompts: Production-Grade FastAPI and System Design\n\nThese prompts are for senior engineers building scalable, secure, and observable microservices.\n\n### 8. FastAPI with JWT Auth, Role-Based Access, and Rate Limiting\n\nTask: Generate a FastAPI app with full authentication flow.\n\nPrompt:\n\nBuild a FastAPI application with:\n1. User registration (POST /auth/register) with password hashing (bcrypt)\n2. Login endpoint (POST /auth/login) returning JWT access token and refresh token\n3. JWT token validation middleware using python-jose\n4. Role-based access: 'admin' and 'user' roles\n5. Admin-only endpoint DELETE /users/{id} (only admin can delete)\n6. Rate limiting using slowapi: 100 requests per minute per IP for regular endpoints, 20 for auth\n7. Refresh token endpoint (POST /auth/refresh)\n8. Store users in SQLite (for simplicity) using SQLAlchemy\n9. Include a dependency get_current_user that returns user from token\n\nUse Python 3.11, FastAPI 0.110+, and pydantic v2.\n\n\nExample result (abbreviated):\npython\n# auth.py\nfrom datetime import datetime, timedelta\nfrom jose import JWTError, jwt\nfrom passlib.context import CryptContext\nfrom fastapi import Depends, HTTPException, status\nfrom fastapi.security import OAuth2PasswordBearer\n\nSECRET_KEY = \"your-secret-key\"\nALGORITHM = \"HS256\"\nACCESS_TOKEN_EXPIRE = timedelta(minutes=30)\nREFRESH_TOKEN_EXPIRE = timedelta(days=7)\n\npwd_context = CryptContext(schemes=[\"bcrypt\"], deprecated=\"auto\")\noauth2_scheme = OAuth2PasswordBearer(tokenUrl=\"/auth/login\")\n\ndef create_access_token(data: dict):\n to_encode = data.copy()\n expire = datetime.utcnow() + ACCESS_TOKEN_EXPIRE\n to_encode.update({\"exp\": expire})\n return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)\n\ndef create_refresh_token(data: dict):\n to_encode = data.copy()\n expire = datetime.utcnow() + REFRESH_TOKEN_EXPIRE\n to_encode.update({\"exp\": expire, \"type\": \"refresh\"})\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=\"Could not validate credentials\",\n )\n try:\n payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])\n username: str = payload.get(\"sub\")\n if username is None:\n raise credentials_exception\n except JWTError:\n raise credentials_exception\n user = await get_user_by_username(username) # async DB call\n if user is None:\n raise credentials_exception\n return user\n\ndef require_role(required_role: str):\n async def role_checker(current_user = Depends(get_current_user)):\n if current_user.role != required_role:\n raise HTTPException(status_code=403, detail=\"Insufficient permissions\")\n return current_user\n return role_checker\n\n\n### 9. Async Data Pipeline with Kafka and FastStream\n\nTask: Create a Kafka consumer that processes events and writes to PostgreSQL.\n\nPrompt:\n\nBuild an async data pipeline using FastStream (Kafka) and SQLAlchemy async:\n1. Consumer reads JSON messages from Kafka topic 'user_events'\n2. Each message has fields: user_id, event_type, timestamp, payload (dict)\n3. Validate message with Pydantic v2 schema\n4. Store event in PostgreSQL table 'events' with columns: id (serial), user_id, event_type, event_data (JSONB), created_at\n5. If event_type is 'purchase', also update a 'purchase_summary' materialized view (just execute REFRESH MATERIALIZED VIEW)\n6. Handle deserialization errors: log and send to dead-letter topic 'user_events_dlq'\n7. Use dependency injection for database session\n\nProvide full code with FastStream app, model, and schema.\n\n\nExample result (consumer):\npython\nfrom faststream import FastStream\nfrom faststream.kafka import KafkaBroker\nfrom pydantic import BaseModel\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession\nfrom sqlalchemy import text\nfrom typing import Any\nimport json\n\nbroker = KafkaBroker(\"localhost:9092\")\napp = FastStream(broker)\n\nDATABASE_URL = \"postgresql+asyncpg://user:pass@localhost/db\"\nengine = create_async_engine(DATABASE_URL)\n\nclass UserEvent(BaseModel):\n user_id: str\n event_type: str\n timestamp: str\n payload: dict\n\n@broker.subscriber(\"user_events\")\nasync def handle_event(msg: Any):\n try:\n event = UserEvent(**json.loads(msg.value.decode()))\n except Exception as e:\n await broker.publish(msg.value, topic=\"user_events_dlq\")\n return\n async with AsyncSession(engine) as session:\n async with session.begin():\n await session.execute(\n text(\"INSERT INTO events (user_id, event_type, event_data) VALUES (:uid, :etype, :edata)\"),\n {\"uid\": event.user_id, \"etype\": event.event_type, \"edata\": json.dumps(event.payload)}\n )\n if event.event_type == \"purchase\":\n await session.execute(text(\"REFRESH MATERIALIZED VIEW purchase_summary\"))\n\n\n### 10. FastAPI with OpenTelemetry, Prometheus Metrics, and Structured Logging\n\nTask: Add observability to a FastAPI service.\n\nPrompt:\n\nEnhance an existing FastAPI app with:\n1. OpenTelemetry instrumentation: automatic tracing for HTTP requests, SQLAlchemy queries, and Redis calls\n2. Export traces to Jaeger (OTLP gRPC)\n3. Prometheus metrics: request count, latency histogram (with endpoint label), error counter\n4. Structured JSON logging using structlog with correlation ID in each log entry\n5. A middleware that adds a unique request_id to each request and includes it in logs and traces\n6. A health check endpoint GET /health that returns 200 and checks DB connectivity\n\nUse opentelemetry-instrumentation-fastapi, opentelemetry-exporter-otlp, prometheus_client, structlog.\n\nAssume FastAPI app is in app.py.\n\n\nExample result (middleware and instrumentation):\npython\n# instrumentation.py\nfrom opentelemetry import trace\nfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter\nfrom opentelemetry.instrumentation.fastapi import FastAPIInstrumentor\nfrom opentelemetry.sdk.trace import TracerProvider\nfrom opentelemetry.sdk.trace.export import BatchSpanProcessor\nfrom prometheus_client import Counter, Histogram, generate_latest\nfrom fastapi import Request, Response\nimport structlog\nimport uuid\nimport time\n\n# Tracing setup\ntracer_provider = TracerProvider()\nspan_processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=\"localhost:4317\", insecure=True))\ntracer_provider.add_span_processor(span_processor)\ntrace.set_tracer_provider(tracer_provider)\n\n# Metrics\nREQUEST_COUNT = Counter('http_requests_total', 'Total requests', ['method', 'endpoint'])\nREQUEST_LATENCY = Histogram('http_request_duration_seconds', 'Request latency', ['method', 'endpoint'])\nERROR_COUNT = Counter('http_errors_total', 'Total errors', ['method', 'endpoint'])\n\n# Structured logging\nstructlog.configure(\n processors=[\n structlog.stdlib.add_log_level,\n structlog.processors.TimeStamper(fmt=\"iso\"),\n structlog.processors.JSONRenderer()\n ],\n context_class=dict,\n)\n\ndef setup_instrumentation(app):\n FastAPIInstrumentor.instrument_app(app)\n\n @app.middleware(\"http\")\n async def add_observability(request: Request, call_next):\n request_id = str(uuid.uuid4())\n request.state.request_id = request_id\n start_time = time.time()\n response = await call_next(request)\n duration = time.time() - start_time\n REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path).inc()\n REQUEST_LATENCY.labels(method=request.method, endpoint=request.url.path).observe(duration)\n if response.status_code >= 400:\n ERROR_COUNT.labels(method=request.method, endpoint=request.url.path).inc()\n response.headers[\"X-Request-ID\"] = request_id\n return response\n\n @app.get(\"/metrics\")\n async def metrics():\n return Response(content=generate_latest(), media_type=\"text/plain\")\n\n\n### 11. Database Migration Script with Alembic and Async\n\nTask: Generate Alembic migration configuration for an async SQLAlchemy setup.\n\nPrompt:\n\nCreate Alembic migration files for an async SQLAlchemy 2.0 project:\n1. Write env.py that uses async engine (asyncpg) and run_async\n2. Create an initial migration that adds a 'products' table with columns: id (UUID), name (string 100), price (numeric 10,2), created_at (timestamp)\n3. Write a second migration that adds a 'category' column (string 50) and a unique constraint on (name, category)\n4. Provide the alembic.ini with correct database URL placeholder\n5. Use Alembic 1.13+\n\nAssume models are in app.models.\n\n\nExample result (env.py snippet):\npython\nfrom alembic import context\nfrom sqlalchemy.ext.asyncio import create_async_engine\nfrom app.models import Base\n\nconfig = context.config\n\ndef run_migrations_online():\n connectable = create_async_engine(config.get_main_option(\"sqlalchemy.url\"))\n\n async def do_run_migrations(connection):\n context.configure(connection=connection, target_metadata=Base.metadata)\n with context.begin_transaction():\n context.run_migrations()\n\n async def run_async():\n async with connectable.connect() as connection:\n await do_run_migrations(connection)\n\n import asyncio\n asyncio.run(run_async())\n\n\n### 12. Dockerized FastAPI with Multi-Stage Build\n\nTask: Create a production-ready Dockerfile for a FastAPI app.\n\nPrompt:\n\nWrite a multi-stage Dockerfile for a FastAPI application:\n1. Stage 1 (builder): Python 3.11-slim, install Poetry, copy pyproject.toml, poetry.lock, run poetry install --no-dev\n2. Stage 2 (runtime): Python 3.11-slim, copy site-packages from builder, copy app code\n3. Expose port 8000\n4. Run with gunicorn + uvicorn workers (uvicorn.workers.UvicornWorker) with 4 workers\n5. Healthcheck: curl --fail http://localhost:8000/health\n6. Use non-root user 'appuser'\n7. Set PYTHONDONTWRITEBYTECODE=1 and PYTHONUNBUFFERED=1\n\nProvide complete Dockerfile and a docker-compose.yml that also starts a PostgreSQL and Redis.\n\n\nExample result:\ndockerfile\n# Dockerfile\nFROM python:3.11-slim AS builder\nWORKDIR /app\nRUN pip install poetry\nCOPY pyproject.toml poetry.lock ./\nRUN poetry config virtualenvs.create false && poetry install --no-dev\n\nFROM python:3.11-slim AS runtime\nWORKDIR /app\nCOPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages\nCOPY --from=builder /usr/local/bin /usr/local/bin\nCOPY app/ ./app/\nRUN adduser --disabled-password --gecos '' appuser\nUSER appuser\nENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1\nEXPOSE 8000\nHEALTHCHECK --interval=30s --timeout=3s CMD curl --fail http://localhost:8000/health || exit 1\nCMD [\"gunicorn\", \"app.main:app\", \"--worker-class\", \"uvicorn.workers.UvicornWorker\", \"--bind\", \"0.0.0.0:8000\", \"--workers\", \"4\"]\n\n\n### 13. FastAPI with CQRS and Event Sourcing Patterns\n\nTask: Implement a simple CQRS pattern with separate read and write models.\n\nPrompt:\n\nBuild a FastAPI application for a simple inventory system using CQRS:\n1. Write model (PostgreSQL): events table stores all changes as JSON (event sourcing)\n2. Read model (PostgreSQL materialized view or separate table): current inventory snapshot\n3. POST /inventory/items creates an item (write side) and publishes an event\n4. GET /inventory/items returns current state (read side)\n5. A background process (using asyncio.create_task) updates the read model after each write\n6. Use Pydantic for event schemas\n7. Include a GET /events/{item_id} to replay history\n\nThis is a simplified example; in production you'd use a message broker.\n\n\nExample result (event store and read model update):\n```python\n# events.py\nfrom datetime import datetime\nfrom pydantic import