20 Expert Prompts for ChatGPT and GPT-4: From Code Generation to System Architecture

You are a senior backend engineer. Write a Python module called `rate_limiter` that implements a token bucket algorithm for API rate limiting. Requirements:
- Support configurable rate (requests per second) and burst size.
- Thread-safe using `threading.Lock`.
- Include a `wait_for_token()` method that blocks until a token is available.
- Handle edge cases: negative rates, zero burst, high concurrency.
- Provide 3 unit tests using `pytest` covering normal usage, edge cases, and concurrent access.
- Add a docstring for each public method and a module-level docstring.
- Use type hints.

import time
import threading
from typing import Optional

class TokenBucket:
    """Implements a token bucket rate limiter."""

    def __init__(self, rate: float, burst: int) -> None:
        if rate <= 0:
            raise ValueError("Rate must be positive")
        if burst <= 0:
            raise ValueError("Burst must be positive")
        self.rate = rate
        self.burst = burst
        self.tokens = float(burst)
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
        self.last_refill = now

    def wait_for_token(self, timeout: Optional[float] = None) -> bool:
        """Block until a token is available. Returns True if acquired."""
        deadline = time.monotonic() + timeout if timeout else None
        while True:
            with self.lock:
                self._refill()
                if self.tokens >= 1:
                    self.tokens -= 1
                    return True
            if deadline and time.monotonic() >= deadline:
                return False
            time.sleep(0.001)

Act as a data analyst. Write a PostgreSQL query for an e-commerce database with tables: `orders`, `order_items`, `products`, `customers`. The query must:
- Return monthly revenue for the last 12 months, grouped by product category.
- Include columns: month, category, total_revenue, order_count, average_order_value.
- Exclude cancelled orders (status = 'cancelled').
- Use `DATE_TRUNC` for monthly grouping.
- Handle categories with zero revenue in a month (use `COALESCE`).
- Optimize for large datasets: use indexes hints if applicable.
- Add comments explaining each CTE or subquery.

WITH monthly_orders AS (
    SELECT
        o.id AS order_id,
        DATE_TRUNC('month', o.created_at) AS month,
        p.category,
        oi.quantity,
        oi.unit_price
    FROM orders o
    JOIN order_items oi ON o.id = oi.order_id
    JOIN products p ON oi.product_id = p.id
    WHERE o.status != 'cancelled'
        AND o.created_at >= CURRENT_DATE - INTERVAL '12 months'
),
monthly_agg AS (
    SELECT
        month,
        category,
        SUM(quantity * unit_price) AS total_revenue,
        COUNT(DISTINCT order_id) AS order_count,
        AVG(quantity * unit_price) AS avg_order_value
    FROM monthly_orders
    GROUP BY month, category
),
all_months AS (
    SELECT GENERATE_SERIES(
        DATE_TRUNC('month', CURRENT_DATE - INTERVAL '11 months'),
        DATE_TRUNC('month', CURRENT_DATE),
        '1 month'
    ) AS month
),
categories AS (
    SELECT DISTINCT category FROM products
)
SELECT
    am.month,
    c.category,
    COALESCE(ma.total_revenue, 0) AS total_revenue,
    COALESCE(ma.order_count, 0) AS order_count,
    COALESCE(ma.avg_order_value, 0) AS avg_order_value
FROM all_months am
CROSS JOIN categories c
LEFT JOIN monthly_agg ma ON am.month = ma.month AND c.category = ma.category
ORDER BY am.month DESC, c.category;

You are a concurrency expert. Review the following Python code that uses threading to process a list of URLs. It has a bug: sometimes results are duplicated or missing. Find the root cause and provide a corrected version.

```python
import threading
import requests

results = []
def fetch(url):
    response = requests.get(url)
    results.append(response.status_code)

urls = ['https://example.com'] * 10
threads = [threading.Thread(target=fetch, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(results)

**Example Result:**
**Root cause:** The `results` list is shared without synchronization. Multiple threads may call `append` concurrently, leading to lost updates or corrupted list state (CPython's GIL protects single operations, but `append` is atomic; however, the real issue here is that all threads fetch the same URL, but the output can still have missing entries due to race conditions in list resizing under high load).

**Fixed version:**
```python
import threading
import requests

results = []
lock = threading.Lock()
def fetch(url):
    response = requests.get(url)
    with lock:
        results.append(response.status_code)

urls = ['https://example.com'] * 10
threads = [threading.Thread(target=fetch, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(len(results))  # Should be 10

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def fetch(url):
    return requests.get(url).status_code

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    results = [future.result() for future in as_completed(futures)]
print(len(results))

You are a software architect. I have a Django monolith with 200+ models, 500+ views, and business logic scattered across views and signals. The app is 5 years old, with no test coverage. Propose a step-by-step refactoring plan that:
- Extracts core business logic into service layer.
- Introduces repository pattern for database access.
- Gradually isolates modules for future microservices extraction.
- Includes testing strategy (start with integration tests for critical paths).
- Lists tools and libraries (e.g., pytest, django-test-migrations, structlog).
- Estimates effort in person-weeks for a team of 3.
- Warns about common pitfalls (e.g., breaking signals, circular imports).

Design a real-time notification delivery system for a social media platform with 10 million daily active users. Requirements:
- Support push (APNS/FCM), email (SMTP), SMS (Twilio-like provider).
- Delivery latency < 5 seconds for push, < 1 minute for email.
- Handle 100,000 notifications per second peak.
- Exactly-once delivery semantics (at least once with idempotency).
- Retry with exponential backoff for failed deliveries.
- Prioritize notifications (critical > transactional > promotional).
- Provide a high-level diagram description, technology choices (message queue, database, etc.), and a scaling strategy.

Generate an OpenAPI 3.0.3 specification for a `PaymentService` microservice with the following endpoints:
- POST /payments - initiate payment (request body: amount, currency, source_id, description).
- GET /payments/{id} - retrieve payment status.
- POST /payments/{id}/refund - initiate refund (request body: amount).
- GET /health - health check.
Include:
- Proper error responses (400, 404, 500) with JSON body containing `error_code` and `message`.
- Security scheme using API key in header `X-API-Key`.
- Example values for all request/response bodies.
- Use `$ref` for reusable schemas (e.g., `PaymentResponse`, `ErrorResponse`).
- Markdown formatting not required, just valid YAML.

openapi: "3.0.3"
info:
  title: Payment Service API
  version: "1.0.0"
paths:
  /payments:
    post:
      summary: Initiate a payment
      security:
        - ApiKeyAuth: []
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreatePaymentRequest'
      responses:
        '201':
          description: Payment created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/PaymentResponse'
        '400':
          description: Validation error
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ErrorResponse'
        '401':
          description: Unauthorized
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/ErrorResponse'
components:
  securitySchemes:
    ApiKeyAuth:
      type: apiKey
      in: header
      name: X-API-Key
  schemas:
    CreatePaymentRequest:
      type: object
      required: [amount, currency, source_id]
      properties:
        amount:
          type: number
          example: 19.99
        currency:
          type: string
          example: USD
        source_id:
          type: string
          example: tok_visa_4242
        description:
          type: string
          example: Order #12345
    PaymentResponse:
      type: object
      properties:
        id:
          type: string
          example: pay_abc123
        status:
          type: string
          enum: [pending, succeeded, failed]

Write a comprehensive test suite for the following Python function using `pytest` with parameterization:

```python
import re

def is_valid_email(email: str) -> bool:
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

**Example Result:**
```python
import pytest

def is_valid_email(email: str) -> bool:
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

@pytest.mark.parametrize("email,expected", [
    ("user@example.com", True),
    ("user.name+tag@example.co.uk", True),
    ("a@b.cd", True),
    ("", False),
    ("user@.com", False),
    ("@example.com", False),
    ("user@com", False),
    ("user name@example.com", False),
    ("a" * 256 + "@example.com", True),  # regex doesn't limit length
])
def test_is_valid_email(email, expected):
    assert is_valid_email(email) == expected

Create a multi-stage Dockerfile for a FastAPI application that:
- Stage 1: Install dependencies using `pip` with a `requirements.txt` file, use `--no-cache-dir`.
- Stage 2: Copy only the application code and installed packages from stage 1, use `python:3.12-slim` as base.
- Run as non-root user (create user `appuser`).
- Set `WORKDIR` to `/app`.
- Expose port 8000.
- Use `uvicorn` as the server with `CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]`.
- Include healthcheck using `curl` (install `curl` in stage 2).
- Minimize image size.

FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.12-slim
RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
RUN groupadd -r appuser && useradd -r -g appuser appuser
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY . .
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Write a professional README.md for a Python client library called `weather-client` that wraps the OpenWeatherMap API. Include:
- Badges (Python version, license, CI status).
- Installation via pip.
- Quick start example (get current weather by city name).
- API reference for main classes (`WeatherClient`, `Forecast`).
- Configuration (API key via environment variable).
- Error handling example.
- Link to full documentation on ReadTheDocs.
- Contribution guidelines.

# Weather Client

[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue)](https://python.org)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![CI](https://github.com/example/weather-client/actions/workflows/ci.yml/badge.svg)](https://github.com/example/weather-client/actions)

A Python client for the OpenWeatherMap API.

## Installation

```bash
pip install weather-client

import os
from weather_client import WeatherClient

client = WeatherClient(api_key=os.environ["OPENWEATHER_API_KEY"])
weather = client.get_current_weather(city="London")
print(f"Temperature: {weather.temperature}°C")

from weather_client import WeatherClient, APIError

try:
    weather = client.get_current_weather(city="InvalidCity")
except APIError as e:
    print(f"API error: {e.status_code} - {e.message}")