10 Prompts for GPT-4: From Code Snippets to System Architecture

You are an expert Python developer. Write a function `load_csv_with_validation(filepath: str, required_columns: list[str]) -> list[dict]` that:
- Opens the CSV file using the `csv` module.
- Checks that all `required_columns` exist in the header. If any are missing, raise a `ValueError` with a descriptive message listing the missing columns.
- Returns a list of dictionaries, each representing a row.
- Uses type hints and includes a docstring in Google style.
- Handles `FileNotFoundError` gracefully by re-raising it with a user-friendly message.

import csv
from typing import List, Dict

def load_csv_with_validation(filepath: str, required_columns: List[str]) -> List[Dict[str, str]]:
    """Load a CSV file and validate required columns.

    Args:
        filepath: Path to the CSV file.
        required_columns: List of column names that must exist.

    Returns:
        List of dictionaries where keys are column names.

    Raises:
        FileNotFoundError: If the file does not exist.
        ValueError: If required columns are missing.
    """
    try:
        with open(filepath, 'r', newline='') as f:
            reader = csv.DictReader(f)
            if not reader.fieldnames:
                raise ValueError("CSV file is empty or has no header.")
            missing = [col for col in required_columns if col not in reader.fieldnames]
            if missing:
                raise ValueError(f"Missing required columns: {', '.join(missing)}")
            return [row for row in reader]
    except FileNotFoundError:
        raise FileNotFoundError(f"CSV file not found at: {filepath}")

Given the following Python function:

def factorial(n: int) -> int:
    if n < 0:
        raise ValueError("Factorial not defined for negative numbers")
    if n == 0:
        return 1
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result

Write a pytest test suite that:
- Tests edge cases: n=0, n=1, n=5, n=10.
- Tests that a negative input raises ValueError.
- Uses parameterized tests to avoid duplication.
- Includes a test for large input (n=100) to ensure performance (timeout of 1 second).

import pytest
from math_functions import factorial

@pytest.mark.parametrize("n, expected", [
    (0, 1),
    (1, 1),
    (5, 120),
    (10, 3628800),
])
def test_factorial_known_values(n, expected):
    assert factorial(n) == expected

def test_factorial_negative():
    with pytest.raises(ValueError, match="not defined for negative"):
        factorial(-1)

@pytest.mark.timeout(1)
def test_factorial_large():
    result = factorial(100)
    assert result > 0
    assert isinstance(result, int)

Convert the following JavaScript function to TypeScript. Add TypeScript interfaces for the API response and function parameters. Use async/await and handle HTTP errors (non-200 status codes).

// Original JS
function fetchUserData(userId) {
    return fetch(`https://api.example.com/users/${userId}`)
        .then(response => response.json())
        .then(data => console.log(data.name));
}

interface User {
    id: number;
    name: string;
    email: string;
}

interface ApiResponse<T> {
    data: T;
    error?: string;
}

async function fetchUserData(userId: number): Promise<User> {
    const response = await fetch(`https://api.example.com/users/${userId}`);
    if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
    }
    const result: ApiResponse<User> = await response.json();
    if (result.error) {
        throw new Error(result.error);
    }
    console.log(result.data.name);
    return result.data;
}

I have a Python script that uses asyncio and aiohttp to scrape multiple URLs concurrently. The results are stored in a global list. Sometimes the list is missing items or has duplicates. Here is the code:

import asyncio
import aiohttp

results = []

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            results.append(len(data))

async def main():
    urls = ["https://example.com"] * 10
    tasks = [fetch(url) for url in urls]
    await asyncio.gather(*tasks)
    print(len(results))

asyncio.run(main())

Explain the race condition, provide a fix using a thread-safe data structure or a lock, and show the corrected code.

The race condition occurs because `results.append()` is not an atomic operation at the Python level. Under asyncio, multiple coroutines can interleave during the `append` method, causing lost updates. The fix is to use `asyncio.Lock` or switch to `asyncio.Queue` for producer-consumer pattern. Here's the corrected version using a lock:

import asyncio
import aiohttp

results = []
lock = asyncio.Lock()

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            async with lock:
                results.append(len(data))

async def main():
    urls = ["https://example.com"] * 10
    tasks = [fetch(url) for url in urls]
    await asyncio.gather(*tasks)
    print(len(results))  # Always 10

asyncio.run(main())

Refactor the following monolithic function into smaller, single-responsibility functions. Each function should have a clear purpose, be testable in isolation, and use dependency injection for external services (database, email, analytics). Provide the refactored code in Python.

# Original monolith
def signup_user(email, password, name):
    if len(password) < 8:
        return {"error": "Password too short"}
    import re
    if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
        return {"error": "Invalid email"}
    # Check database for existing user
    import sqlite3
    conn = sqlite3.connect("users.db")
    cursor = conn.cursor()
    cursor.execute("SELECT id FROM users WHERE email = ?", (email,))
    if cursor.fetchone():
        conn.close()
        return {"error": "Email already exists"}
    # Insert user
    import hashlib
    hashed_pw = hashlib.sha256(password.encode()).hexdigest()
    cursor.execute("INSERT INTO users (email, password, name) VALUES (?, ?, ?)", (email, hashed_pw, name))
    conn.commit()
    conn.close()
    # Send welcome email
    import smtplib
    with smtplib.SMTP("smtp.example.com") as server:
        server.sendmail("noreply@example.com", email, "Welcome!")
    # Log analytics event
    print(f"Analytics: user {email} signed up")
    return {"success": True}

# validation.py
import re

def validate_password(password: str) -> str | None:
    if len(password) < 8:
        return "Password too short"
    return None

def validate_email(email: str) -> str | None:
    if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
        return "Invalid email"
    return None

# repository.py
class UserRepository:
    def __init__(self, connection):
        self.conn = connection

    def find_by_email(self, email):
        cursor = self.conn.cursor()
        cursor.execute("SELECT id FROM users WHERE email = ?", (email,))
        return cursor.fetchone()

    def create_user(self, email, hashed_password, name):
        cursor = self.conn.cursor()
        cursor.execute("INSERT INTO users (email, password, name) VALUES (?, ?, ?)",
                       (email, hashed_password, name))
        self.conn.commit()

# email_service.py
class EmailService:
    def __init__(self, smtp_server):
        self.smtp_server = smtp_server

    def send_welcome_email(self, to_email):
        import smtplib
        with smtplib.SMTP(self.smtp_server) as server:
            server.sendmail("noreply@example.com", to_email, "Welcome!")

# analytics.py
class AnalyticsLogger:
    def log_event(self, event_name, user_email):
        print(f"Analytics: {event_name} for {user_email}")

# main_signup.py
from validation import validate_password, validate_email

def signup_user(email, password, name, user_repo, email_service, analytics):
    err = validate_email(email)
    if err:
        return {"error": err}
    err = validate_password(password)
    if err:
        return {"error": err}
    if user_repo.find_by_email(email):
        return {"error": "Email already exists"}
    hashed_pw = hashlib.sha256(password.encode()).hexdigest()
    user_repo.create_user(email, hashed_pw, name)
    email_service.send_welcome_email(email)
    analytics.log_event("signup", email)
    return {"success": True}

Explain the A* pathfinding algorithm step by step using a 5x5 grid. The start is at (0,0) and goal at (4,4). Obstacles are at (1,1), (2,2), (3,3). Use Manhattan distance as heuristic. Show the open set, closed set, and f, g, h values after each iteration until the path is found. Provide the final path coordinates.

Grid representation:
  0 1 2 3 4
0 S . . . .
1 . X . . .
2 . . X . .
3 . . . X .
4 . . . . G

Initial state:
- Open set: [(0,0)] with f=8 (g=0, h=8)
- Closed set: []

Iteration 1: Expand (0,0). Neighbors: (0,1), (1,0).
- (0,1): g=1, h=7, f=8
- (1,0): g=1, h=7, f=8
Open set: [(0,1), (1,0)] both f=8. Open set sorted by f, then h.

Iteration 2: Expand (0,1) (lower h). Neighbors: (0,2), (1,1) blocked.
- (0,2): g=2, h=6, f=8
Open set: [(1,0), (0,2)] both f=8.

... (continues until goal is reached)

Final path: (0,0) -> (0,1) -> (0,2) -> (0,3) -> (0,4) -> (1,4) -> (2,4) -> (3,4) -> (4,4)

Act as a senior software architect. Design a notification microservice in Python (FastAPI) that supports three channels: email (via SMTP), SMS (via Twilio API), and push notifications (via Firebase Cloud Messaging). Requirements:
- Each notification must be idempotent (use a unique idempotency key).
- Retry failed deliveries up to 3 times with exponential backoff.
- Store delivery status in a PostgreSQL database.
- Expose a REST API: POST /notifications (body: channel, recipient, content, idempotency_key).
- Provide a high-level architecture diagram in text (ASCII or Mermaid).
- Include code for the main router, the retry logic, and the database schema (SQLAlchemy models).

flowchart TD
Client -->

|POST /notifications| API[FastAPI Router]
    API --> IdempotencyCheck[Check idempotency_key in DB]
IdempotencyCheck -->

|Exists| ReturnStatus[Return existing status]
IdempotencyCheck -->

|New| ChannelRouter[Route to channel handler]
    ChannelRouter --> EmailHandler
    ChannelRouter --> SMSHandler
    ChannelRouter --> PushHandler
    EachHandler --> RetryQueue[Retry with exponential backoff]
    RetryQueue --> DB[PostgreSQL]

# models.py
from sqlalchemy import Column, String, Integer, DateTime, Enum
from sqlalchemy.ext.declarative import declarative_base
import enum

Base = declarative_base()

class NotificationStatus(enum.Enum):
    pending = "pending"
    sent = "sent"
    failed = "failed"

class Notification(Base):
    __tablename__ = "notifications"
    id = Column(Integer, primary_key=True)
    idempotency_key = Column(String, unique=True, nullable=False)
    channel = Column(String, nullable=False)
    recipient = Column(String, nullable=False)
    content = Column(String, nullable=False)
    status = Column(Enum(NotificationStatus), default=NotificationStatus.pending)
    retry_count = Column(Integer, default=0)
    created_at = Column(DateTime, server_default=func.now())

You are a database architect. A SaaS platform currently uses a single PostgreSQL database for all tenants. As the number of tenants grew, write-heavy queries on the users table are causing contention. Propose a strategy to migrate to a sharded database with 4 shards, using a consistent hashing approach on tenant_id. The migration must have zero downtime. Provide:
- The sharding key selection rationale.
- A step-by-step migration plan (e.g., dual-writes, backfill, cutover).
- Example Python code for the consistent hashing ring.
- Rollback plan.

Sharding key: tenant_id (mod 4). Consistent hashing reduces reshuffling when adding shards.

Migration steps:
1. Add a new 'shard_map' table mapping tenant_id to shard_id.
2. Write to both old DB and new shard (dual-write) for all INSERT/UPDATE on users table.
3. Backfill historical data: run a batch job that reads from old DB in chunks and writes to the correct shard.
4. Verify consistency with a diff tool (compare row counts and checksums per tenant).
5. Cutover: update application config to read from shard first, fall back to old DB. After a week, remove fallback.
6. Rollback: keep old DB online for 30 days; if issues arise, flip the read config back to old DB and stop dual-writes.

Consistent hashing example:
import hashlib

def get_shard(tenant_id: str, num_shards: int = 4) -> int:
    hash_val = int(hashlib.md5(tenant_id.encode()).hexdigest(), 16)
    return hash_val % num_shards

Review the following pull request diff for a FastAPI file upload endpoint. Focus on security vulnerabilities (e.g., path traversal, file size limits, content type validation) and performance issues (e.g., loading entire file into memory). Suggest concrete fixes.

# PR diff
@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    with open(f"/uploads/{file.filename}", "wb") as f:
        f.write(contents)
    return {"filename": file.filename}

Issues identified:
1. **Path traversal**: `file.filename` could contain "../" to overwrite system files. Fix: sanitize filename using `os.path.basename` and generate a UUID-based filename.
2. **No file size limit**: A malicious client could upload a multi-GB file, exhausting memory. Fix: use `UploadFile`’s `file` attribute (a SpooledTemporaryFile) and stream it in chunks.
3. **No content type validation**: An attacker could upload an executable. Fix: validate `file.content_type` against a whitelist and/or use `python-magic` for MIME detection.
4. **Synchronous file write in async endpoint**: Blocks the event loop. Fix: use `aiofiles` for async write.

Fixed code:
import os
import uuid
import aiofiles
from fastapi import HTTPException

ALLOWED_CONTENT_TYPES = {"image/png", "image/jpeg", "application/pdf"}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(400, "Unsupported file type")
    # Validate size by reading in chunks
    size = 0
    async with aiofiles.open(f"/uploads/{uuid.uuid4()}", "wb") as f:
        while chunk := await file.read(1024 * 1024):
            size += len(chunk)
            if size > MAX_FILE_SIZE:
                raise HTTPException(413, "File too large")
            await f.write(chunk)
    return {"filename": file.filename}

Compare Event Sourcing (ES) and traditional CRUD for implementing an order management system in an e-commerce platform. Provide:
- A pros/cons table.
- A recommendation based on the following constraints: the system must support audit trails, handle high write throughput (1000 orders/sec), and require eventual consistency.
- Example code for saving an order in both approaches (Python with SQLAlchemy for CRUD, and with an event store for ES).

# CRUD
order = Order(user_id=123, total=49.99)
db.session.add(order)
db.session.commit()

# ES - store event
event = OrderCreatedEvent(order_id=uuid4(), user_id=123, total=49.99, timestamp=datetime.utcnow())
event_store.append(event)
# Projection updates asynchronously
projection.apply(event)  # updates a read model