#!/usr/bin/env python3
"""
Expand Python dataset from 120 to 200 entries.
Adds: SQLAlchemy, Advanced Type Hints, API Patterns, CLI/Packaging,
Advanced Logging, Error Handling, Threading/Multiprocessing, Configuration.
"""
import json
import os
import random
from pathlib import Path
BASE = Path(__file__).parent.parent / "data" / "processed"
def g(issue, bad, good):
"""Generate a training entry."""
return {
"messages": [
{
"role": "user",
"content": f"Review this Python code:\n\n```python\n{bad}\n```"
},
{
"role": "assistant",
"content": f"**Issue: {issue}**\n\n**Fix:**\n```python\n{good}\n```"
}
],
"source": "python_training_expansion",
"tags": ["python", issue.lower().replace(" ", "_")]
}
entries = []
# =============================================================================
# SQLALCHEMY / ORM PATTERNS (15 entries)
# =============================================================================
entries.append(g(
"SQLAlchemy N+1 Query",
"users = session.query(User).all()\nfor user in users:\n print(user.posts) # Lazy loads each post",
"from sqlalchemy.orm import joinedload\nusers = session.query(User).options(joinedload(User.posts)).all()\n# Single query with JOIN"
))
entries.append(g(
"SQLAlchemy Session Leak",
"def get_user(user_id):\n session = Session()\n return session.query(User).get(user_id)\n # Session never closed",
"from contextlib import contextmanager\n\n@contextmanager\ndef session_scope():\n session = Session()\n try:\n yield session\n session.commit()\n except Exception:\n session.rollback()\n raise\n finally:\n session.close()\n\n# Or use dependency injection container"
))
entries.append(g(
"SQLAlchemy Eager Loading Missing",
"# Template accesses post.author.name for each post\nposts = session.query(Post).filter_by(category='tech').all()",
"from sqlalchemy.orm import selectinload\nposts = session.query(Post).options(\n selectinload(Post.author)\n).filter_by(category='tech').all()\n# One query for posts, one for authors"
))
entries.append(g(
"SQLAlchemy Raw SQL Injection",
"users = session.query(User).filter(\n User.name.like(f\"%{user_input}%\")\n).all()",
"users = session.query(User).filter(\n User.name.like(f\"%{user_input}%\")\n).all()\n# SQLAlchemy parameterizes automatically\n# But be careful with .text() and .from_statement()\nusers = session.execute(\n text(\"SELECT * FROM users WHERE name LIKE :name\"),\n {\"name\": f\"%{user_input}%\"}\n)"
))
entries.append(g(
"SQLAlchemy Missing Relationship Cascade",
"class User(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n posts = relationship('Post')\n # Deleting User leaves orphaned Posts",
"class User(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n posts = relationship('Post', cascade='all, delete-orphan')\n # Deleting User also deletes their posts"
))
entries.append(g(
"SQLAlchemy Query Chaining Without Limit",
"all_posts = session.query(Post).all()\n# Table has millions of rows",
"page = request.args.get('page', 1, type=int)\nper_page = request.args.get('per_page', 50, type=int)\nposts = session.query(Post).paginate(\n page=page, per_page=per_page, error_out=False\n)\n# Or use limit/offset directly"
))
entries.append(g(
"SQLAlchemy Bulk Operations",
"for item in items:\n new_record = Model(data=item)\n session.add(new_record)\nsession.commit()\n# 10K inserts = 10K round trips",
"session.bulk_save_objects([Model(data=item) for item in items])\n# Or even faster:\nsession.execute(\n Model.__table__.insert(),\n [{\"data\": item} for item in items]\n)"
))
entries.append(g(
"SQLAlchemy Column vs Property Confusion",
"class User(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n name = Column(String(100))\n # Accidentally using property as column\n def active(self):\n return self.last_login > datetime.now() - timedelta(days=30)",
"from sqlalchemy.ext.hybrid import hybrid_property\n\nclass User(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n name = Column(String(100))\n last_login = Column(DateTime)\n\n @hybrid_property\n def active(self):\n return self.last_login > datetime.now() - timedelta(days=30)\n\n @active.expression\n def active(cls):\n return cls.last_login > datetime.now() - timedelta(days=30)\n # Can now filter: session.query(User).filter(User.active.is_(True))"
))
entries.append(g(
"SQLAlchemy Migration Missing",
"# Adding new column to model but forgetting migration\nclass User(Base):\n __tablename__ = 'users'\n id = Column(Integer, primary_key=True)\n name = Column(String(100))\n email = Column(String(255)) # NEW COLUMN",
"# Always generate migration for schema changes\n# Terminal command:\n# alembic revision --autogenerate -m \"add user email\"\n# alembic upgrade head\n\n# Migration file auto-generated:\n# op.add_column('users', sa.Column('email', sa.String(255), nullable=True))"
))
entries.append(g(
"SQLAlchemy Detached Instance Error",
"def process_user(user_id):\n user = session.query(User).get(user_id)\n session.close()\n return user.email # DetachedInstanceError",
"def process_user(user_id):\n user = session.query(User).get(user_id)\n email = user.email # Access before session close\n session.close()\n return email\n\n# Or make a DTO:\n return {\"email\": user.email}"
))
entries.append(g(
"SQLAlchemy Lazy Loading in Production",
"class User(Base):\n posts = relationship('Post', lazy='dynamic')\n # Each template access triggers query",
"class User(Base):\n posts = relationship('Post', lazy='joined')\n # Or better: use explicit options in queries\n # Avoid lazy='subquery' or 'select' in production templates"
))
entries.append(g(
"SQLAlchemy Connection Pool Exhaustion",
"# Default pool_size=5, max_overflow=10\n# Under load, all connections used up\nengine = create_engine('postgresql://user:pass@host/db')",
"engine = create_engine(\n 'postgresql://user:pass@host/db',\n pool_size=20,\n max_overflow=10,\n pool_timeout=30,\n pool_recycle=1800, # Recycle connections every 30min\n pool_pre_ping=True # Test connections before use\n)"
))
entries.append(g(
"SQLAlchemy Scalar Query",
"result = session.query(func.count(User.id)).all()\n# Returns [(42,)] - tuple, not int",
"count = session.query(func.count(User.id)).scalar()\n# Returns 42 directly\n\n# Or with newer SQLAlchemy:\nresult = session.execute(select(func.count(User.id)))\ncount = result.scalar()"
))
entries.append(g(
"SQLAlchemy Select vs Filter",
"users = session.query(User).select(User.id, User.name).all()\n# Returns tuples, not User objects",
"users = session.query(User).filter(User.active == True).all()\n# Returns User objects\n\n# If you only need specific columns:\nresult = session.query(User.id, User.name).all()\n# Returns [(1, 'Alice'), (2, 'Bob')] - use when you don't need full objects"
))
entries.append(g(
"SQLAlchemy Transaction Isolation",
"def transfer(from_acc, to_acc, amount):\n # Without isolation, concurrent transfers can corrupt balances\n from_balance = session.query(Account).get(from_acc).balance\n to_balance = session.query(Account).get(to_acc).balance",
"from sqlalchemy import event\nfrom sqlalchemy.orm import sessionmaker\n\n# Set isolation level\nengine = create_engine(\n 'postgresql://...',\n isolation_level='SERIALIZABLE' # Prevents phantom reads\n)\n\n# Or use pessimistic locking:\nfrom_acc = session.query(Account).filter_by(id=from_acc).with_for_update().first()\nto_acc = session.query(Account).filter_by(id=to_acc).with_for_update().first()"
))
# =============================================================================
# ADVANCED TYPE HINTS (10 entries)
# =============================================================================
entries.append(g(
"Protocol for Structural Subtyping",
"def process(duck):\n return duck.quack() + duck.waddle()\n# No type checking, any object with these methods works",
"from typing import Protocol\n\nclass Quackable(Protocol):\n def quack(self) -> str: ...\n def waddle(self) -> str: ...\n\ndef process(duck: Quackable) -> str:\n return duck.quack() + duck.waddle()\n# Type checker verifies structural compatibility"
))
entries.append(g(
"TypeGuard for Narrowing",
"def process(data: Union[str, list, dict]):\n if isinstance(data, str):\n return data.upper()\n # Type checker still sees Union[str, list, dict]",
"from typing import TypeGuard, Union\n\ndef is_string(data: object) -> TypeGuard[str]:\n return isinstance(data, str)\n\ndef process(data: Union[str, list, dict]):\n if is_string(data):\n return data.upper() # Type narrowed to str\n elif isinstance(data, list):\n return [str(x) for x in data]\n return dict(data)"
))
entries.append(g(
"TypedDict for Structured Data",
"config: dict = {\"host\": \"localhost\", \"port\": 8080}\n# No type checking on keys/values",
"from typing import TypedDict\n\nclass ServerConfig(TypedDict):\n host: str\n port: int\n timeout: float\n\nconfig: ServerConfig = {\"host\": \"localhost\", \"port\": 8080}\n# Type checker ensures all required keys are present"
))
entries.append(g(
"Overload for Multiple Signatures",
"def get_item(key: str) -> Union[str, dict]:\n # Returns str for single key, dict for multiple\n pass\n# Caller can't know which type they'll get",
"from typing import overload, Union\n\n@overload\ndef get_item(key: str) -> str: ...\n@overload\ndef get_item(key: list[str]) -> dict[str, str]: ...\ndef get_item(key):\n if isinstance(key, str):\n return db.get(key)\n return {k: db.get(k) for k in key}\n# Type checker infers return type based on input"
))
entries.append(g(
"Self Type for Fluent Interfaces",
"class QueryBuilder:\n def filter(self, **kwargs):\n self._filters.update(kwargs)\n return self # Type is QueryBuilder, not self\n def order_by(self, field):\n self._order = field\n return self\n# Subclasses lose type info on method chains",
"from typing import Self\n\nclass QueryBuilder:\n def filter(self, **kwargs) -> Self:\n self._filters.update(kwargs)\n return self\n def order_by(self, field) -> Self:\n self._order = field\n return self\n# Subclasses maintain their type in chains"
))
entries.append(g(
"Literal Types for Restricted Values",
"def set_status(status: str):\n # Accepts any string, including typos\n db.set(\"status\", status)",
"from typing import Literal\n\ndef set_status(status: Literal['active', 'pending', 'closed']):\n db.set(\"status\", status)\n# Type checker restricts to exact values\n\nset_status('active') # OK\nset_status('pending') # OK\nset_status('closed') # OK\nset_status('unknown') # TYPE ERROR"
))
entries.append(g(
"Generic Type with TypeVar",
"def first_item(items):\n return items[0]\n# No type info on return",
"from typing import TypeVar, Sequence\n\nT = TypeVar('T')\n\ndef first_item(items: Sequence[T]) -> T:\n return items[0]\n# Return type matches element type of sequence\n\nresult = first_item([1, 2, 3]) # inferred as int\nresult = first_item(['a', 'b']) # inferred as str"
))
entries.append(g(
"NewType for Distinct Types",
"user_id: str = \"abc123\"\ndb_id: str = \"42\"\n# Both are str, can be confused",
"from typing import NewType\n\nUserId = NewType('UserId', str)\nDbId = NewType('DbId', int)\n\nuser_id: UserId = UserId(\"abc123\")\ndb_id: DbId = DbId(42)\n# Type checker prevents mixing UserId and DbId"
))
entries.append(g(
"ParamSpec for Decorator Type Safety",
"def retry(func):\n def wrapper(*args, **kwargs):\n for _ in range(3):\n try:\n return func(*args, **kwargs)\n except Exception:\n pass\n return wrapper\n# Loses original function signature for type checkers",
"from typing import ParamSpec, Callable, TypeVar\n\nP = ParamSpec('P')\nR = TypeVar('R')\n\ndef retry(func: Callable[P, R]) -> Callable[P, R]:\n def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:\n for _ in range(3):\n try:\n return func(*args, **kwargs)\n except Exception:\n pass\n return wrapper\n# Preserves original function signature"
))
entries.append(g(
"TypeAlias for Complex Types",
"def process(data: dict[str, list[tuple[int, float, str]]]):\n # Hard to read and maintain\n pass",
"from typing import TypeAlias\n\nPoint: TypeAlias = tuple[int, float, str]\nDataset: TypeAlias = dict[str, list[Point]]\n\ndef process(data: Dataset):\n # Much more readable\n pass"
))
# =============================================================================
# API PATTERNS (10 entries)
# =============================================================================
entries.append(g(
"Missing Rate Limiting",
"@app.route('/api/search')\ndef search():\n query = request.args.get('q')\n return jsonify(search_db(query))\n# No protection against abuse",
"from flask_limiter import Limiter\nfrom flask_limiter.util import get_remote_address\n\nlimiter = Limiter(\n app=app,\n key_func=get_remote_address,\n default_limits=[\"200 per day\", \"50 per hour\"]\n)\n\n@app.route('/api/search')\n@limiter.limit(\"10 per minute\")\ndef search():\n query = request.args.get('q')\n return jsonify(search_db(query))"
))
entries.append(g(
"Missing API Versioning",
"@app.route('/api/users')\ndef get_users():\n return jsonify(User.query.all())\n# Breaking changes affect all clients",
"# URL versioning\n@app.route('/api/v1/users')\ndef get_users_v1():\n return jsonify(User.query.all())\n\n# Or header versioning\n@app.route('/api/users')\ndef get_users():\n version = request.headers.get('API-Version', 'v1')\n if version == 'v1':\n return get_users_v1()\n elif version == 'v2':\n return get_users_v2()"
))
entries.append(g(
"Missing API Response Envelope",
"@app.route('/api/users')\ndef get_users():\n users = User.query.all()\n return jsonify(users)\n# Inconsistent response format",
"def success_response(data, status=200):\n return {\n \"status\": \"success\",\n \"data\": data,\n \"meta\": {\"timestamp\": datetime.utcnow().isoformat()}\n }, status\n\ndef error_response(message, status=400):\n return {\n \"status\": \"error\",\n \"message\": message,\n \"meta\": {\"timestamp\": datetime.utcnow().isoformat()}\n }, status\n\n@app.route('/api/users')\ndef get_users():\n try:\n users = User.query.all()\n return success_response(users)\n except Exception as e:\n return error_response(\"Failed to fetch users\", 500)"
))
entries.append(g(
"Missing API Pagination",
"@app.route('/api/items')\ndef get_items():\n return jsonify(Item.query.all())\n# Returns all items - memory issue on large tables",
"@app.route('/api/items')\ndef get_items():\n page = request.args.get('page', 1, type=int)\n per_page = min(request.args.get('per_page', 50, type=int), 100) # Cap at 100\n \n pagination = Item.query.paginate(\n page=page, per_page=per_page, error_out=False\n )\n \n return jsonify({\n \"items\": [item.to_dict() for item in pagination.items],\n \"total\": pagination.total,\n \"pages\": pagination.pages,\n \"current_page\": page,\n \"per_page\": per_page\n })"
))
entries.append(g(
"Missing API Caching",
"@app.route('/api/categories')\ndef get_categories():\n # Rarely changes, but queried every request\n categories = Category.query.all()\n return jsonify([c.to_dict() for c in categories])",
"from flask_caching import Cache\n\ncache = Cache(app, config={'CACHE_TYPE': 'RedisCache', 'CACHE_DEFAULT_TIMEOUT': 300})\n\n@app.route('/api/categories')\n@cache.cached(timeout=300, unless=lambda: request.args.get('force_refresh'))\ndef get_categories():\n categories = Category.query.all()\n return jsonify([c.to_dict() for c in categories])\n # Cached for 5 minutes"
))
entries.append(g(
"Missing API Error Handling",
"@app.route('/api/users/<int:user_id>')\ndef get_user(user_id):\n user = User.query.get(user_id)\n if not user:\n return None # Returns 500 on None jsonify\n return jsonify(user.to_dict())",
"@app.errorhandler(404)\ndef not_found(error):\n return jsonify({\"error\": \"Not found\"}), 404\n\n@app.errorhandler(500)\ndef internal_error(error):\n return jsonify({\"error\": \"Internal server error\"}), 500\n\n@app.route('/api/users/<int:user_id>')\ndef get_user(user_id):\n user = User.query.get(user_id)\n if not user:\n abort(404)\n return jsonify(user.to_dict())"
))
entries.append(g(
"Missing API Authentication",
"@app.route('/api/admin/stats')\ndef get_stats():\n return jsonify(get_admin_stats())\n# No auth check - anyone can access",
"from functools import wraps\nfrom flask import request, jsonify\n\ndef require_auth(f):\n @wraps(f)\n def decorated(*args, **kwargs):\n token = request.headers.get('Authorization')\n if not token or not verify_token(token):\n return jsonify({\"error\": \"Unauthorized\"}), 401\n return f(*args, **kwargs)\n return decorated\n\n@app.route('/api/admin/stats')\n@require_auth\ndef get_stats():\n return jsonify(get_admin_stats())"
))
entries.append(g(
"Missing API Input Validation",
"@app.route('/api/users', methods=['POST'])\ndef create_user():\n data = request.get_json()\n user = User(\n name=data['name'],\n email=data['email']\n )\n db.session.add(user)\n return jsonify(user.to_dict())\n# Fails with KeyError if fields missing",
"from flask import request, jsonify\nfrom marshmallow import Schema, fields, validate\n\nclass UserSchema(Schema):\n name = fields.String(required=True, validate=validate.Length(min=1, max=100))\n email = fields.Email(required=True)\n age = fields.Integer(validate=validate.Range(min=0, max=150))\n\nschema = UserSchema()\n\n@app.route('/api/users', methods=['POST'])\ndef create_user():\n data = request.get_json()\n errors = schema.validate(data)\n if errors:\n return jsonify({\"errors\": errors}), 400\n \n validated_data = schema.load(data)\n user = User(**validated_data)\n db.session.add(user)\n db.session.commit()\n return jsonify(user.to_dict()), 201"
))
entries.append(g(
"Missing API Health Check",
"# No health endpoint - orchestration can't monitor service",
"@app.route('/health')\ndef health_check():\n health = {\n \"status\": \"healthy\",\n \"timestamp\": datetime.utcnow().isoformat(),\n \"version\": app.config.get('VERSION', 'unknown')\n }\n \n # Check database connectivity\n try:\n db.session.execute(text(\"SELECT 1\"))\n health[\"database\"] = \"connected\"\n except Exception:\n health[\"database\"] = \"disconnected\"\n health[\"status\"] = \"degraded\"\n \n status_code = 200 if health[\"status\"] == \"healthy\" else 503\n return jsonify(health), status_code"
))
entries.append(g(
"Missing API CORS Configuration",
"# Frontend on different origin gets CORS errors",
"from flask_cors import CORS\n\napp = Flask(__name__)\nCORS(app, resources={\n r\"/api/*\": {\n \"origins\": [\"https://frontend.example.com\"],\n \"methods\": [\"GET\", \"POST\", \"PUT\", \"DELETE\"],\n \"allow_headers\": [\"Content-Type\", \"Authorization\"],\n \"max_age\": 600 # Pre-flight cache for 10 minutes\n }\n})"
))
# =============================================================================
# ADVANCED LOGGING (8 entries)
# =============================================================================
entries.append(g(
"Missing Structured Logging",
"logger.info(f\"User {user_id} logged in from {ip_address}\")\n# Hard to parse and search in logs",
"import logging\nimport json\n\nclass JSONFormatter(logging.Formatter):\n def format(self, record):\n log_data = {\n \"timestamp\": self.formatTime(record),\n \"level\": record.levelname,\n \"message\": record.getMessage(),\n \"module\": record.module,\n \"function\": record.funcName,\n \"line\": record.lineno\n }\n # Add extra fields\n if hasattr(record, 'user_id'):\n log_data['user_id'] = record.user_id\n if hasattr(record, 'ip_address'):\n log_data['ip_address'] = record.ip_address\n return json.dumps(log_data)\n\nformatter = JSONFormatter()\nhandler = logging.StreamHandler()\nhandler.setFormatter(formatter)\nlogger = logging.getLogger(__name__)\nlogger.addHandler(handler)\n\n# Usage:\nextra = {'user_id': user_id, 'ip_address': ip_address}\nlogger.info(\"User logged in\", extra=extra)"
))
entries.append(g(
"Missing Log Levels",
"logger.info(\"Processing user input\") # Too noisy for production\nlogger.info(\"User request completed\") # Should be debug level",
"import logging\n\n# Development: DEBUG\n# Production: INFO\nlogger.setLevel(logging.DEBUG if debug else logging.INFO)\n\nlogger.debug(\"Processing user input: %s\", sanitized_input)\nlogger.info(\"User request completed in %.2fms\", duration)\nlogger.warning(\"Rate limit approaching: %d/%d requests\", current, limit)\nlogger.error(\"Database connection failed: %s\", str(e))\nlogger.critical(\"Service unavailable - all replicas down\")"
))
entries.append(g(
"Missing Log Redaction",
"logger.info(\"User auth attempt: %s\", f\"user={username}, pass={password}\")\n# Password in logs!",
"import logging\nimport re\n\nclass RedactingFilter(logging.Filter):\n SENSITIVE_PATTERNS = [\n re.compile(r'(?i)(password|passwd|pwd)[\s=]*[\\w]+'),\n re.compile(r'(?i)(api[_-]?key|secret|token)[\s=]*[\\w]+'),\n re.compile(r'(?i)(credit[_-]?card|cc)[\s=]*\\d{13,19}'),\n ]\n \n def filter(self, record):\n record.msg = re.sub(\n self.SENSITIVE_PATTERNS,\n lambda m: m.group(0).replace(m.group(0)[10:], '[REDACTED]'),\n str(record.msg)\n )\n return True\n\nlogger.addFilter(RedactingFilter())"
))
entries.append(g(
"Missing Correlation IDs",
"# Distributed system - can't trace requests across services",
"import logging\nimport uuid\nimport contextvars\n\nrequest_id_ctx = contextvars.ContextVar('request_id', default=None)\n\nclass CorrelationIdFilter(logging.Filter):\n def filter(self, record):\n record.request_id = request_id_ctx.get() or 'unknown'\n return True\n\n# In middleware:\n@app.before_request\ndef set_request_id():\n request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())\n request_id_ctx.set(request_id)\n\n# In logging config:\nlogger.addFilter(CorrelationIdFilter())\nformatter = logging.Formatter('%(asctime)s %(request_id)s %(levelname)s %(message)s')"
))
entries.append(g(
"Missing Log Rotation",
"logging.basicConfig(filename='app.log', level=logging.INFO)\n# Single log file grows forever",
"from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler\n\n# Rotate by size (10MB, keep 5 backups)\nhandler = RotatingFileHandler(\n 'app.log',\n maxBytes=10*1024*1024,\n backupCount=5\n)\n\n# Or rotate by time (daily, keep 30 days)\nhandler = TimedRotatingFileHandler(\n 'app.log',\n when='midnight',\n interval=1,\n backupCount=30\n)\n\nlogger.addHandler(handler)"
))
entries.append(g(
"Missing Log Aggregation",
"# Logs only go to stdout - hard to search across services",
"# Docker logging driver\n# docker-compose.yml:\nservices:\n app:\n logging:\n driver: fluentd\n options:\n fluentd-address: \"localhost:24224\"\n tag: \"app.%{service_name}%\"\n\n# Or use ELK stack:\nimport logging.handlers\n\n# Syslog handler for centralization\nsyslog_handler = logging.handlers.SysLogHandler(\n address=('log-server', 514),\n facility=logging.handlers.SysLogHandler.LOG_USER\n)"
))
entries.append(g(
"Missing Performance Logging",
"def process_data():\n # No timing info - hard to identify slow operations",
"import time\nimport logging\n\nlogger = logging.getLogger(__name__)\n\ndef log_performance(func):\n def wrapper(*args, **kwargs):\n start = time.perf_counter()\n try:\n result = func(*args, **kwargs)\n return result\n finally:\n elapsed = time.perf_counter() - start\n if elapsed > 0.1: # Log if > 100ms\n logger.warning(\n \"%s took %.3fs\", func.__name__, elapsed\n )\n return wrapper\n\n@log_performance\ndef process_data():\n # Auto-logged if slow\n pass"
))
entries.append(g(
"Missing Log Context",
"logger.info(\"Payment processed\")\n# No context - which payment? which user?",
"import logging\n\nlogger = logging.getLogger(__name__)\n\n# Add context to logger\ncontext_logger = logger.getChild('payments')\n\n# Usage with context:\ndef process_payment(user_id, amount):\n logger.info(\n \"Processing payment\",\n extra={\n 'user_id': user_id,\n 'amount': amount,\n 'payment_type': 'credit_card'\n }\n )\n # Structured logging with context for easier debugging"
))
# =============================================================================
# ADVANCED ERROR HANDLING (10 entries)
# =============================================================================
entries.append(g(
"Missing Exception Hierarchy",
"raise ValueError(\"Invalid user input\")\nraise ValueError(\"Database connection failed\")\nraise ValueError(\"API rate limit exceeded\")\n# All ValueError - hard to handle specifically",
"class AppError(Exception):\n \"\"\"Base application exception\"\"\"\n def __init__(self, message, code=None):\n super().__init__(message)\n self.code = code\n\nclass ValidationError(AppError):\n \"\"\"Input validation failed\"\"\"\n pass\n\nclass DatabaseError(AppError):\n \"\"\"Database operation failed\"\"\"\n pass\n\nclass RateLimitError(AppError):\n \"\"\"API rate limit exceeded\"\"\"\n pass\n\n# Usage:\ntry:\n validate_input(data)\nexcept ValidationError as e:\n handle_validation_error(e)"
))
entries.append(g(
"Missing Exception Chaining",
"try:\n result = process_data()\nexcept ProcessError as e:\n raise RuntimeError(\"Failed to process\") # Loses original traceback",
"try:\n result = process_data()\nexcept ProcessError as e:\n raise RuntimeError(\"Failed to process data\") from e # Preserves chain\n\n# Original traceback is preserved in __cause__\n# Better for debugging: you see both errors"
))
entries.append(g(
"Missing Retry Pattern",
"result = requests.get(url)\n# Fails on transient network errors",
"from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type\n\n@retry(\n stop=stop_after_attempt(3),\n wait=wait_exponential(multiplier=1, min=1, max=10),\n retry=retry_if_exception_type((requests.RequestException, TimeoutError)),\n reraise=True\n)\ndef fetch_data(url):\n response = requests.get(url, timeout=5)\n response.raise_for_status()\n return response.json()"
))
entries.append(g(
"Missing Sentinel Values",
"def get_config(key):\n try:\n return config[key]\n except KeyError:\n return None # Can't distinguish missing from actual None value",
"# Sentinel pattern\n_MISSING = object()\n\ndef get_config(key):\n result = config.get(key, _MISSING)\n if result is _MISSING:\n raise KeyError(f\"Config key not found: {key}\")\n return result # Can be None, False, 0, etc.\n\n# Alternative: use exceptions\ntry:\n value = config[key]\nexcept KeyError:\n # Handle missing key specifically\n raise ConfigError(f\"Missing required config: {key}\")"
))
entries.append(g(
"Missing Context Manager for Resources",
"file = open('data.txt')\ndata = file.read()\n# File left open if exception occurs",
"from contextlib import contextmanager\n\n@contextmanager\ndef managed_resource(resource):\n try:\n resource.acquire()\n yield resource\n finally:\n resource.release()\n\n# Usage:\nwith managed_resource(database_connection) as conn:\n conn.execute(\"SELECT * FROM users\")\n # Connection always closed, even on exception"
))
entries.append(g(
"Missing Error Recovery",
"def process_payments(payments):\n for payment in payments:\n process(payment) # One failure stops all",
"from collections import defaultdict\n\ndef process_payments(payments):\n results = defaultdict(list)\n \n for payment in payments:\n try:\n process(payment)\n results['success'].append(payment.id)\n except ValidationError:\n results['validation_errors'].append(payment.id)\n except DatabaseError:\n results['db_errors'].append(payment.id)\n except Exception as e:\n results['unknown_errors'].append({\n 'id': payment.id,\n 'error': str(e)\n })\n \n return results\n # Continue processing, report all failures"
))
entries.append(g(
"Missing Error Context",
"def divide(a, b):\n if b == 0:\n raise ValueError(\"Division by zero\")\n return a / b\n # Error message doesn't include the actual values",
"def divide(a, b):\n if b == 0:\n raise ValueError(f\"Division by zero: {a} / {b}\")\n return a / b\n # Better error messages include context for debugging\n\n# Even better: custom exception with context\nclass CalculationError(Exception):\n def __init__(self, operation, operand1, operand2):\n self.operation = operation\n self.operand1 = operand1\n self.operand2 = operand2\n super().__init__(f\"{operation}({operand1}, {operand2}) failed\")"
))
entries.append(g(
"Missing Async Exception Handling",
"async def fetch_data():\n result = await api_call()\n # Exception swallowed if not awaited",
"async def fetch_data():\n try:\n result = await asyncio.wait_for(\n api_call(),\n timeout=5.0\n )\n return result\n except asyncio.TimeoutError:\n logger.warning(\"API call timed out\")\n return default_value\n except aiohttp.ClientError as e:\n logger.error(\"API error: %s\", e)\n raise\n except Exception as e:\n logger.exception(\"Unexpected error in fetch_data\")\n raise"
))
entries.append(g(
"Missing Error Reporting",
"try:\n process_payment()\nexcept Exception:\n pass # Silent failure!",
"import sentry_sdk # or rollbar, datadog, etc.\n\ntry:\n process_payment()\nexcept ValidationError as e:\n # Expected error - log but don't alert\n logger.warning(\"Payment validation failed: %s\", e)\n return {\"error\": \"Invalid payment data\"}\nexcept PaymentError as e:\n # Business error - track but don't panic\n logger.error(\"Payment failed: %s\", e)\n sentry_sdk.capture_exception(e)\n return {\"error\": \"Payment processing failed\"}\nexcept Exception as e:\n # Unexpected error - full alert\n logger.exception(\"Unexpected payment error\")\n sentry_sdk.capture_exception(e)\n raise"
))
entries.append(g(
"Missing Error Boundaries",
"async def serve_request(request):\n # Any exception crashes the whole request handler",
"async def serve_request(request):\n try:\n return await handle_request(request)\n except ValidationError as e:\n return json_response({\"error\": str(e)}, 400)\n except AuthenticationError:\n return json_response({\"error\": \"Unauthorized\"}, 401)\n except AuthorizationError:\n return json_response({\"error\": \"Forbidden\"}, 403)\n except NotFoundError:\n return json_response({\"error\": \"Not found\"}, 404)\n except RateLimitError:\n return json_response({\"error\": \"Rate limit exceeded\"}, 429)\n except Exception as e:\n logger.exception(\"Unhandled error in request\")\n return json_response({\"error\": \"Internal server error\"}, 500)\n # All errors caught, service stays up"
))
# =============================================================================
# THREADING / MULTIPROCESSING (8 entries)
# =============================================================================
entries.append(g(
"Thread Safety Missing",
"counter = 0\ndef increment():\n global counter\n counter += 1 # Not atomic!",
"import threading\n\ncounter = 0\nlock = threading.Lock()\n\ndef increment():\n global counter\n with lock: # Thread-safe\n counter += 1"
))
entries.append(g(
"Deadlock Risk",
"lock_a = threading.Lock()\nlock_b = threading.Lock()\n\ndef thread1():\n with lock_a:\n with lock_b:\n process()\n\ndef thread2():\n with lock_b:\n with lock_a:\n process()\n# Deadlock! Each thread holds one lock and waits for the other",
"lock_a = threading.Lock()\nlock_b = threading.Lock()\n\ndef thread1():\n with lock_a:\n with lock_b:\n process()\n\ndef thread2():\n with lock_a: # Same order as thread1\n with lock_b:\n process()\n# Always acquire locks in consistent order\n# Or use a single lock if they protect related data"
))
entries.append(g(
"Missing ThreadPoolExecutor",
"results = []\nfor url in urls:\n result = requests.get(url)\n results.append(result)\n# Sequential - slow for I/O bound work",
"from concurrent.futures import ThreadPoolExecutor\n\ndef fetch(url):\n return requests.get(url).json()\n\nwith ThreadPoolExecutor(max_workers=10) as executor:\n results = list(executor.map(fetch, urls))\n# Parallel I/O - much faster for network requests"
))
entries.append(g(
"Missing ProcessPoolExecutor",
"results = []\nfor data in large_dataset:\n result = heavy_computation(data)\n results.append(result)\n# Sequential - doesn't use multiple cores",
"from concurrent.futures import ProcessPoolExecutor\n\ndef heavy_computation(data):\n # CPU-bound work\n return process(data)\n\nwith ProcessPoolExecutor(max_workers=4) as executor:\n results = list(executor.map(heavy_computation, large_dataset))\n# Uses multiple CPU cores - bypasses GIL"
))
entries.append(g(
"Thread Local Storage Missing",
"current_user = None # Global - shared across threads!\n\ndef middleware(request):\n global current_user\n current_user = authenticate(request)\n return handle_request(request)",
"import threading\n\nthread_local = threading.local()\n\ndef middleware(request):\n thread_local.current_user = authenticate(request)\n return handle_request(request)\n\n# Each thread has its own current_user\n# Safe for multi-threaded servers"
))
entries.append(g(
"Missing Queue for Producer-Consumer",
"# Unbounded list for task queue - memory leak risk\ntasks = []\ndef producer():\n while True:\n tasks.append(generate_task())\n\ndef consumer():\n while tasks:\n process(tasks.pop(0))",
"from queue import Queue\nimport threading\n\ntask_queue = Queue(maxsize=1000) # Bounded queue\n\ndef producer():\n while True:\n task_queue.put(generate_task()) # Blocks if full\n\ndef consumer():\n while True:\n task = task_queue.get() # Blocks if empty\n process(task)\n task_queue.task_done()\n\n# Start consumer threads\nfor _ in range(5):\n t = threading.Thread(target=consumer, daemon=True)\n t.start()"
))
entries.append(g(
"Missing Barrier for Coordination",
"# Threads need to sync at checkpoint - using sleep() is fragile\nimport time\ntime.sleep(1) # Wait for other threads",
"import threading\n\nnum_threads = 4\nbarrier = threading.Barrier(num_threads)\n\ndef worker(thread_id):\n # Phase 1\n do_phase1(thread_id)\n \n barrier.wait() # All threads reach here before continuing\n \n # Phase 2\n do_phase2(thread_id)\n\n# Start threads\nthreads = [threading.Thread(target=worker, args=(i,)) for i in range(num_threads)]\nfor t in threads:\n t.start()\nfor t in threads:\n t.join()"
))
entries.append(g(
"Missing Timeout on Thread Join",
"t = threading.Thread(target=long_running_task)\nt.start()\nt.join() # Blocks forever if task hangs",
"t = threading.Thread(target=long_running_task)\nt.start()\nt.join(timeout=30) # Wait max 30 seconds\n\nif t.is_alive():\n logger.warning(\"Task didn't complete in 30s\")\n # Handle timeout - cancel task or retry\n handle_timeout()"
))
# =============================================================================
# CLI / PACKAGING (8 entries)
# =============================================================================
entries.append(g(
"Missing CLI Framework",
"import sys\nimport argparse\n\nparser = argparse.ArgumentParser()\nparser.add_argument('--input', required=True)\nparser.add_argument('--output', default='out.txt')\nparser.add_argument('--verbose', action='store_true')\nargs = parser.parse_args()\n# Works but verbose",
"import click\n\n@click.command()\n@click.option('--input', '-i', required=True, help='Input file')\n@click.option('--output', '-o', default='out.txt', help='Output file')\n@click.option('--verbose', '-v', is_flag=True, help='Verbose output')\ndef cli(input, output, verbose):\n '''Process input file'''\n process(input, output, verbose)\n\nif __name__ == '__main__':\n cli()"
))
entries.append(g(
"Missing pyproject.toml",
"# setup.py\nfrom setuptools import setup, find_packages\nsetup(\n name='mypackage',\n version='0.1.0',\n packages=find_packages(),\n install_requires=['requests>=2.28'],\n)",
"# pyproject.toml\n[build-system]\nrequires = [\"hatchling\"]\nbuild-backend = \"hatchling.build\"\n\n[project]\nname = \"mypackage\"\nversion = \"0.1.0\"\ndescription = \"My package\"\nreadme = \"README.md\"\nrequires-python = \">=3.8\"\ndependencies = [\n \"requests>=2.28\",\n]\n\n[project.scripts]\nmycli = \"mypackage.cli:main\"\n\n[tool.hatch.build.targets.wheel]\npackages = [\"src/mypackage\"]"
))
entries.append(g(
"Missing CLI Error Handling",
"def main():\n if not os.path.exists(args.input):\n print(f\"Error: {args.input} not found\")\n sys.exit(1)\n# Inconsistent error handling",
"import click\nimport sys\n\nclass CliError(click.ClickException):\n def __init__(self, message, exit_code=1):\n super().__init__(message)\n self.exit_code = exit_code\n\nclass CliExit:\n @staticmethod\n def exit(message, code=1):\n click.echo(f\"Error: {message}\", err=True)\n sys.exit(code)\n\n@click.command()\ndef main(input_file):\n if not os.path.exists(input_file):\n raise CliError(f\"Input file not found: {input_file}\")\n process(input_file)"
))
entries.append(g(
"Missing CLI Progress Bar",
"for i, item in enumerate(items):\n process(item)\n if i % 100 == 0:\n print(f\"Processed {i}/{len(items)}\")",
"from rich.progress import Progress, SpinnerColumn, TextColumn\n\nwith Progress(\n SpinnerColumn(),\n TextColumn(\"[progress.description]{task.description}\"),\n \"•\",\n \"[progress.percentage]{task.percentage:>3.0f}%\",\n transient=True\n) as progress:\n task = progress.add_task(\"Processing...\", total=len(items))\n for item in items:\n process(item)\n progress.update(task, advance=1)"
))
entries.append(g(
"Missing CLI Configuration",
"config = {\n 'host': 'localhost',\n 'port': 8080,\n 'debug': False\n}\n# Hardcoded defaults",
"from pathlib import Path\nimport yaml\n\nDEFAULT_CONFIG = {\n 'host': 'localhost',\n 'port': 8080,\n 'debug': False\n}\n\ndef load_config(config_path: str | None = None) -> dict:\n # Priority: CLI args > config file > defaults\n config = DEFAULT_CONFIG.copy()\n \n # Load from config file if exists\n if config_path:\n path = Path(config_path)\n else:\n path = Path.home() / '.config' / 'myapp' / 'config.yaml'\n \n if path.exists():\n with open(path) as f:\n file_config = yaml.safe_load(f) or {}\n config.update(file_config)\n \n return config"
))
entries.append(g(
"Missing CLI Help Text",
"@click.command()\n@click.option('--input', '-i')\n@click.option('--output', '-o')\ndef cli(input, output):\n process(input, output)",
"@click.command()\n@click.option('--input', '-i', required=True,\n help='Input file path. Supports .json, .yaml, .csv formats.')\n@click.option('--output', '-o', default='output.json',\n help='Output file path. Format inferred from extension.')\n@click.option('--verbose', '-v', is_flag=True,\n help='Enable verbose output with debug information.')\n@click.version_option(version='1.0.0')\ndef cli(input, output, verbose):\n '''Process data files and convert between formats.\n \n Examples:\\n\n\\n \\b\n myapp -i input.json -o output.yaml\n myapp -i data.csv -o data.json --verbose\n '''\n process(input, output, verbose)"
))
entries.append(g(
"Missing CLI Validation",
"@click.command()\n@click.option('--port', default=8080)\ndef start(port):\n # No validation - accepts invalid ports\n run_server(port)",
"@click.command()\n@click.option('--port', default=8080,\n type=click.IntRange(1, 65535))\n@click.option('--workers', default=4,\n type=click.IntRange(1, 32))\ndef start(port, workers):\n # Click validates ranges automatically\n run_server(port, workers)"
))
entries.append(g(
"Missing CLI Subcommands",
"# All commands in one script - hard to maintain\nif args.command == 'serve':\n serve()\nelif args.command == 'build':\n build()\nelif args.command == 'test':\n test()",
"import click\n\n@click.group()\ndef cli():\n '''MyApp CLI - Manage your application'''\n pass\n\n@cli.command()\n@click.option('--port', default=8080)\ndef serve(port):\n '''Start the development server'''\n run_server(port)\n\n@cli.command()\n@click.option('--target', default='production')\ndef build(target):\n '''Build the application'''\n build_app(target)\n\n@cli.command()\n@click.option('--coverage', is_flag=True)\ndef test(coverage):\n '''Run tests'''\n run_tests(coverage=coverage)\n\nif __name__ == '__main__':\n cli()"
))
# =============================================================================
# CONFIGURATION (6 entries)
# =============================================================================
entries.append(g(
"Missing Configuration Validation",
"config = {\n 'host': os.environ.get('DB_HOST', 'localhost'),\n 'port': int(os.environ.get('DB_PORT', '5432')),\n 'timeout': int(os.environ.get('DB_TIMEOUT', '30'))\n}",
"from pydantic import BaseModel, validator\nfrom typing import Optional\n\nclass DatabaseConfig(BaseModel):\n host: str = 'localhost'\n port: int = 5432\n timeout: int = 30\n ssl: bool = False\n \n @validator('port')\n def validate_port(cls, v):\n if not (1 <= v <= 65535):\n raise ValueError('Port must be 1-65535')\n return v\n \n @validator('timeout')\n def validate_timeout(cls, v):\n if v < 1:\n raise ValueError('Timeout must be positive')\n return v\n\n# Load from environment\nconfig = DatabaseConfig(**{\n k.replace('DB_', '').lower(): v\n for k, v in os.environ.items()\n if k.startswith('DB_')\n})"
))
entries.append(g(
"Missing Environment-Specific Config",
"if os.environ.get('ENV') == 'production':\n DEBUG = False\n DB_URL = os.environ['PROD_DB']\nelif os.environ.get('ENV') == 'staging':\n DEBUG = True\n DB_URL = os.environ['STAGING_DB']\nelse:\n DEBUG = True\n DB_URL = 'sqlite:///:memory:'\n# Grows unmaintainable",
"from pydantic import BaseModel\nfrom typing import Optional\n\nclass BaseConfig(BaseModel):\n app_name: str = 'myapp'\n log_level: str = 'INFO'\n\nclass DevConfig(BaseConfig):\n debug: bool = True\n db_url: str = 'sqlite:///:memory:'\n test_mode: bool = True\n\nclass StagingConfig(BaseConfig):\n debug: bool = True\n db_url: str = os.environ['STAGING_DB']\n\nclass ProdConfig(BaseConfig):\n debug: bool = False\n db_url: str = os.environ['PROD_DB']\n sentry_dsn: str = os.environ['SENTRY_DSN']\n\n# Factory\nENV_MAP = {\n 'dev': DevConfig,\n 'staging': StagingConfig,\n 'production': ProdConfig,\n}\n\ndef get_config() -> BaseConfig:\n env = os.environ.get('ENV', 'dev')\n config_class = ENV_MAP[env]\n return config_class()"
))
entries.append(g(
"Missing Configuration Immutability",
"config = {\n 'host': 'localhost',\n 'port': 8080\n}\n# Can be accidentally modified\nconfig['port'] = 9999",
"from types import MappingProxyType\n\nconfig_dict = {\n 'host': 'localhost',\n 'port': 8080,\n 'timeout': 30\n}\n\n# Immutable view - prevents accidental modification\nconfig = MappingProxyType(config_dict)\n\n# config['port'] = 9999 # TypeError: 'mappingproxy' object does not support item assignment"
))
entries.append(g(
"Missing Configuration Documentation",
"# No docs on what config values mean",
"# config.py\n\"\"\"Configuration for MyApp.\n\nAll config values can be set via environment variables:\n - APP_HOST: Server hostname (default: localhost)\n - APP_PORT: Server port (default: 8080)\n - DB_URL: Database connection string\n - REDIS_URL: Redis cache URL\n - DEBUG: Enable debug mode (default: false)\n - LOG_LEVEL: Logging level (default: INFO)\n - SENTRY_DSN: Sentry error tracking DSN\n \nExample .env file:\n APP_HOST=0.0.0.0\n APP_PORT=8080\n DB_URL=postgresql://user:pass@localhost/db\n DEBUG=true\n\"\"\"\nfrom pydantic_settings import BaseSettings\n\nclass Settings(BaseSettings):\n app_host: str = 'localhost'\n app_port: int = 8080\n db_url: str\n redis_url: str | None = None\n debug: bool = False\n log_level: str = 'INFO'\n sentry_dsn: str | None = None\n \n class Config:\n env_file = '.env'"
))
entries.append(g(
"Missing Configuration Hot Reload",
"# Config only loaded at startup - requires restart for changes",
"import os\nfrom pathlib import Path\nimport yaml\n\nclass HotReloadConfig:\n def __init__(self, config_path: str):\n self.path = Path(config_path)\n self._config = self._load()\n self._mtime = self.path.stat().st_mtime\n\n def _load(self) -> dict:\n with open(self.path) as f:\n return yaml.safe_load(f) or {}\n\n def get(self, key: str, default=None):\n # Check for config changes\n if self.path.stat().st_mtime > self._mtime:\n self._config = self._load()\n self._mtime = self.path.stat().st_mtime\n return self._config.get(key, default)"
))
entries.append(g(
"Missing Configuration Schema Validation",
"config = json.loads(os.environ['CONFIG_JSON'])\n# No validation - missing keys cause runtime errors",
"from pydantic import BaseModel, Field, model_validator\n\nclass AppConfig(BaseModel):\n host: str = Field(default='localhost', description='Server hostname')\n port: int = Field(default=8080, ge=1, le=65535)\n db_url: str = Field(..., description='Database connection string')\n workers: int = Field(default=4, ge=1, le=32)\n \n @model_validator(mode='after')\n def validate_environment(self):\n if self.host == '0.0.0.0' and self.port == 8080:\n if os.environ.get('ENV') == 'production':\n raise ValueError('Production must use secure host/port')\n return self\n\n# Raises clear errors on invalid config\ntry:\n config = AppConfig.model_validate(json.loads(os.environ['CONFIG_JSON']))\nexcept Exception as e:\n logger.error(\"Invalid config: %s\", e)\n sys.exit(1)"
))
# =============================================================================
# SAVE
# =============================================================================
random.seed(42) # Reproducible shuffle
random.shuffle(entries)
os.makedirs(BASE, exist_ok=True)
out = BASE / "dev_finetuning_python_expansion.json"
with open(out, 'w') as f:
json.dump(entries, f, indent=2)
print(f"Expansion entries: {len(entries)}")
print(f"Saved: {out}")
# Merge with existing
existing_file = BASE / "dev_finetuning_python.json"
if existing_file.exists():
with open(existing_file) as f:
existing = json.load(f)
combined = existing + entries
random.shuffle(combined)
with open(existing_file, 'w') as f:
json.dump(combined, f, indent=2)
print(f"\\nMerged total: {len(combined)} entries")
print(f"Updated: {existing_file}")
"