10 Prompts for ChatGPT and GPT-4: From Code Writing to System Architecture

You are a senior backend engineer. Write a Python Flask endpoint for user registration. Requirements:
- Accept JSON payload: username (string, 3-30 chars), email (valid format), password (min 8 chars, at least one number and one special character).
- Return 201 with user ID on success.
- Return 400 with specific error messages for each validation failure.
- Include a simple in-memory store (dictionary) for users.
- Add type hints and a docstring.

from flask import Flask, request, jsonify
import re
from typing import Dict, Any

app = Flask(__name__)
users: Dict[str, Any] = {}
user_id_counter = 1

def validate_registration(data: Dict[str, Any]) -> tuple:
    errors = {}
    if not data.get('username') or not (3 <= len(data['username']) <= 30):
        errors['username'] = 'Must be 3-30 characters'
    if not re.match(r'[^@]+@[^@]+\.[^@]+', data.get('email', '')):
        errors['email'] = 'Invalid email format'
    password = data.get('password', '')
    if len(password) < 8 or not re.search(r'\d', password) or not re.search(r'[!@#$%^&*]', password):
        errors['password'] = 'Must be 8+ chars with a number and special char'
    return errors

@app.route('/register', methods=['POST'])
def register():
    data = request.get_json(force=True)
    errors = validate_registration(data)
    if errors:
        return jsonify({'errors': errors}), 400
    global user_id_counter
    user_id = user_id_counter
    user_id_counter += 1
    users[user_id] = {'username': data['username'], 'email': data['email']}
    return jsonify({'user_id': user_id}), 201

Refactor this legacy JavaScript code to use modern ES6+ patterns (arrow functions, destructuring, async/await, modules). The original code handles user authentication and profile updates in one function. Break it into separate modules: auth.js, profile.js, and utils.js. Add JSDoc comments.

Legacy code:
function handleUserAction(action, userData, callback) {
  if (action === 'login') {
    var username = userData.username;
    var password = userData.password;
    // ... 50 lines of mixed logic
  }
  callback(null, result);
}

// auth.js
/**
 * Authenticates a user with username and password.
 * @param {string} username
 * @param {string} password
 * @returns {Promise<Object>}
 */
export const login = async ({ username, password }) => {
  const response = await fetch('/api/login', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ username, password }),
  });
  if (!response.ok) throw new Error('Authentication failed');
  return response.json();
};

Analyze this async Python code for race conditions. The scraper uses a shared list to store results, but sometimes misses URLs. Explain the bug, then provide a fixed version using asyncio.Lock or an alternative pattern.

Buggy code:
import asyncio
results = []
async def fetch(url):
    data = await some_http_call(url)
    results.append(data)
async def main():
    urls = ['url1', 'url2', 'url3']
    await asyncio.gather(*[fetch(u) for u in urls])
    print(len(results))

The bug: `results.append(data)` is a non-atomic operation. Multiple coroutines can interleave, causing data loss or inconsistent state. In CPython, list.append is thread-safe due to GIL, but await points allow context switches. The fix: use asyncio.Lock.

import asyncio
results = []
lock = asyncio.Lock()

async def fetch(url):
    data = await some_http_call(url)
    async with lock:
        results.append(data)

async def main():
    urls = ['url1', 'url2', 'url3']
    await asyncio.gather(*[fetch(u) for u in urls])
    print(len(results))  # Always 3

Design a PostgreSQL schema for a multi-tenant SaaS platform. Requirements:
- Tables: organizations, users, roles, subscriptions, invoices.
- Each user belongs to one organization, but can have multiple roles.
- Subscriptions track plan type and renewal date.
- Add indexes for common queries (e.g., find users by email, invoices by org).
- Include foreign keys with ON DELETE CASCADE where appropriate.
- Write the DDL as a single script.

CREATE TABLE organizations (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    organization_id INTEGER REFERENCES organizations(id) ON DELETE CASCADE,
    created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_users_email ON users(email);

CREATE TABLE roles (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) NOT NULL
);

CREATE TABLE user_roles (
    user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
    role_id INTEGER REFERENCES roles(id) ON DELETE CASCADE,
    PRIMARY KEY (user_id, role_id)
);

CREATE TABLE subscriptions (
    id SERIAL PRIMARY KEY,
    organization_id INTEGER REFERENCES organizations(id) ON DELETE CASCADE,
    plan_type VARCHAR(50) NOT NULL,
    renewal_date DATE NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_subscriptions_org ON subscriptions(organization_id);

CREATE TABLE invoices (
    id SERIAL PRIMARY KEY,
    organization_id INTEGER REFERENCES organizations(id) ON DELETE CASCADE,
    amount DECIMAL(10, 2) NOT NULL,
    due_date DATE NOT NULL,
    paid BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_invoices_org ON invoices(organization_id);

Write a Bash script for a CI/CD pipeline that:
- Builds a Docker image from the current directory.
- Tags it with the commit SHA and 'latest'.
- Pushes to a private Docker registry at registry.example.com.
- SSHes into a staging server and pulls the new image.
- Stops and removes the old container, then starts a new one with port mapping 3000:3000.
- Includes error handling (exit on failure) and logging.

#!/bin/bash
set -e

COMMIT_SHA=$(git rev-parse --short HEAD)
IMAGE_NAME="registry.example.com/myapp:$COMMIT_SHA"
LATEST_TAG="registry.example.com/myapp:latest"

echo "Building image..."
docker build -t $IMAGE_NAME .
docker tag $IMAGE_NAME $LATEST_TAG

echo "Pushing to registry..."
docker push $IMAGE_NAME
docker push $LATEST_TAG

echo "Deploying to staging..."
ssh deploy@staging.example.com << EOF
  docker pull $IMAGE_NAME
  docker stop myapp || true
  docker rm myapp || true
  docker run -d --name myapp -p 3000:3000 $IMAGE_NAME
EOF

echo "Deployment complete."

Write pytest tests for this function that fetches weather data. Use unittest.mock to simulate the API response. Test these scenarios: successful response, HTTP 404, network timeout, and malformed JSON.

Function:
import requests
def get_weather(city: str) -> dict:
    url = f"https://api.weather.com/v1/{city}"
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

import pytest
from unittest.mock import patch, Mock
from weather import get_weather

def test_success():
    mock_response = Mock()
    mock_response.json.return_value = {'temp': 25}
    mock_response.raise_for_status.return_value = None
    with patch('weather.requests.get', return_value=mock_response):
        result = get_weather('London')
        assert result == {'temp': 25}

def test_404():
    mock_response = Mock()
    mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError('404')
    with patch('weather.requests.get', return_value=mock_response):
        with pytest.raises(requests.exceptions.HTTPError):
            get_weather('UnknownCity')

def test_timeout():
    with patch('weather.requests.get', side_effect=requests.exceptions.Timeout):
        with pytest.raises(requests.exceptions.Timeout):
            get_weather('London')

Explain the A* pathfinding algorithm as if to a junior developer. Include:
- The intuition (why it works).
- The heuristic function (Euclidean distance).
- A Python implementation that finds the shortest path in a 2D grid with obstacles.
- An example grid and the resulting path.

import heapq

def heuristic(a, b):
    return ((a[0] - b[0])**2 + (a[1] - b[1])**2) ** 0.5

def a_star(grid, start, goal):
    rows, cols = len(grid), len(grid[0])
    open_set = [(0, start)]
    came_from = {}
    g_score = {start: 0}
    f_score = {start: heuristic(start, goal)}

    while open_set:
        _, current = heapq.heappop(open_set)
        if current == goal:
            path = []
            while current in came_from:
                path.append(current)
                current = came_from[current]
            path.append(start)
            return path[::-1]

        for dx, dy in [(-1,0),(1,0),(0,-1),(0,1)]:
            neighbor = (current[0]+dx, current[1]+dy)
            if 0 <= neighbor[0] < rows and 0 <= neighbor[1] < cols and grid[neighbor[0]][neighbor[1]] == 0:
                tentative_g = g_score[current] + 1
                if neighbor not in g_score or tentative_g < g_score[neighbor]:
                    came_from[neighbor] = current
                    g_score[neighbor] = tentative_g
                    f_score[neighbor] = tentative_g + heuristic(neighbor, goal)
                    heapq.heappush(open_set, (f_score[neighbor], neighbor))
    return None

grid = [
    [0, 0, 0, 0, 0],
    [0, 1, 1, 1, 0],
    [0, 0, 0, 1, 0],
    [0, 1, 0, 0, 0],
    [0, 0, 0, 0, 0]
]
path = a_star(grid, (0, 0), (4, 4))
print(path)  # [(0, 0), (0, 1), (0, 2), (0, 3), (0, 4), (1, 4), (2, 4), (3, 4), (4, 4)]

This query takes 12 seconds on a table with 500k rows. Explain why it's slow, then rewrite it with proper indexes and a more efficient join order. Use EXPLAIN ANALYZE output in your explanation.

SELECT u.name, o.order_date, p.product_name, pay.amount
FROM users u
JOIN orders o ON u.id = o.user_id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
JOIN payments pay ON o.id = pay.order_id
WHERE u.created_at > '2025-01-01'
ORDER BY o.order_date DESC;

-- Problem: Full table scans on all tables due to missing indexes.
-- Fix: Add indexes on foreign keys and the WHERE column.

CREATE INDEX idx_users_created_at ON users(created_at);
CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_order_items_order_id ON order_items(order_id);
CREATE INDEX idx_order_items_product_id ON order_items(product_id);
CREATE INDEX idx_payments_order_id ON payments(order_id);

-- Rewritten query (same logic, but now uses indexes):
SELECT u.name, o.order_date, p.product_name, pay.amount
FROM users u
JOIN orders o ON u.id = o.user_id AND u.created_at > '2025-01-01'
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
JOIN payments pay ON o.id = pay.order_id
ORDER BY o.order_date DESC;

Create a docker-compose.yml for a microservices app with:
- Service A (Flask API) on port 5000.
- Service B (Node.js worker) that consumes messages from Redis.
- Redis as message broker.
- PostgreSQL as the main database.
- All services should restart on failure.
- Use environment variables for configuration.
- Include a healthcheck for PostgreSQL.

version: '3.8'
services:
  api:
    build: ./api
    ports:
      - "5000:5000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    restart: always

  worker:
    build: ./worker
    environment:
      - REDIS_URL=redis://redis:6379/0
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
    depends_on:
      - db
      - redis
    restart: always

  redis:
    image: redis:7-alpine
    restart: always

  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: always
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Review this Flask code for security vulnerabilities. List each issue with severity (low/medium/high), then provide a fixed version.

@app.route('/search')
def search():
    query = request.args.get('q')
    result = db.execute(f"SELECT * FROM items WHERE name = '{query}'")
    return render_template('results.html', items=result.fetchall())

1. SQL Injection (High): f-string interpolation allows arbitrary SQL. Fix: use parameterized queries.
2. XSS (Medium): If item names contain HTML, they will be rendered unsafely. Fix: escape output in template.

Fixed code:
@app.route('/search')
def search():
    query = request.args.get('q')
    result = db.execute("SELECT * FROM items WHERE name = %s", (query,))
    return render_template('results.html', items=result.fetchall())

In template:
{% for item in items %}
  <p>{{ item.name | e }}</p>
{% endfor %}