#!/usr/bin/env python3
"""
Generate comprehensive Python training data for Dev-AI.
Covers: security, performance, modern patterns, anti-patterns, backend, async, testing.
Target: 100-150 high-quality entries.
"""
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",
"tags": ["python", issue.lower().replace(" ", "_")]
}
entries = []
# =============================================================================
# SECURITY (15 entries)
# =============================================================================
entries.append(g(
"SQL Injection",
"cursor.execute(f\"SELECT * FROM users WHERE name = '{name}'\")",
'cursor.execute("SELECT * FROM users WHERE name = ?", (name,))'
))
entries.append(g(
"Hardcoded Secret",
"API_KEY = 'sk-1234567890abcdef'\nclient = Client(api_key=API_KEY)",
"import os\nfrom dotenv import load_dotenv\nload_dotenv()\nAPI_KEY = os.environ['API_KEY']"
))
entries.append(g(
"eval() on User Input",
"def parse_expr(expr):\n return eval(expr)",
"import ast\nimport operator\nALLOWED = {ast.Add: operator.add, ast.Sub: operator.sub}\ndef eval_node(node):\n if isinstance(node, ast.Constant):\n return node.value\n if isinstance(node, ast.BinOp):\n left = eval_node(node.left)\n right = eval_node(node.right)\n return ALLOWED[type(node.op)](left, right)\n raise ValueError('Not allowed')"
))
entries.append(g(
"Path Traversal",
"def get_file(filename):\n path = os.path.join('/data', filename)\n return open(path).read()",
"from pathlib import Path\ndef get_file(filename):\n base = Path('/data').resolve()\n target = (base / filename).resolve()\n if not str(target).startswith(str(base)):\n raise ValueError('Path traversal detected')\n return target.read_text()"
))
entries.append(g(
"XSS in Template",
"from flask import Flask, request, render_template_string\napp = Flask(__name__)\n@app.route('/search')\ndef search():\n q = request.args.get('q', '')\n return render_template_string(f'<h1>Results for {q}</h1>')",
"# Jinja2 auto-escapes by default - use proper templates\nfrom flask import Flask, request, render_template\napp = Flask(__name__)\n@app.route('/search')\ndef search():\n q = request.args.get('q', '')\n return render_template('search.html', query=q)"
))
entries.append(g(
"Insecure Deserialization",
"import pickle\ndata = pickle.loads(user_input)",
"import json\ndata = json.loads(user_input)\n# Use JSON or protobuf instead of pickle"
))
entries.append(g(
"Weak Password Hash",
"import hashlib\nhash = hashlib.md5(password.encode()).hexdigest()",
"import hashlib\nhash = hashlib.sha256(password.encode()).hexdigest()\n# Better: use bcrypt or argon2\nimport argon2\nph = argon2.PasswordHasher()\nhash = ph.hash(password)"
))
entries.append(g(
"Command Injection",
"import subprocess\nsubprocess.run(f\"ping {host}\", shell=True)",
"import shlex\nsubprocess.run(['ping', shlex.split(host)[0]])\n# Never use shell=True with user input"
))
entries.append(g(
"Log Injection",
"logger.info(f\"User input: {user_input}\")",
"import logging\nlogger.info(\"User input: %s\", user_input)\n# Use % formatting to prevent log injection via \\n"
))
entries.append(g(
"Missing HTTPS Redirect",
"from flask import Flask\napp = Flask(__name__)",
"from flask import Flask\nimport flask_talisman\napp = Flask(__name__)\ntalisman = flask_talisman.Talisman(app)\n# Enforces HTTPS, HSTS, and other security headers"
))
entries.append(g(
"CSRF Missing",
"@app.route('/transfer', methods=['POST'])\ndef transfer():\n amount = request.form['amount']\n process_transfer(amount)",
"from flask_wtf.csrf import CSRFProtect\ncsrf = CSRFProtect(app)\n\n@app.route('/transfer', methods=['POST'])\ndef transfer():\n amount = request.form['amount']\n # CSRF token validated automatically\n process_transfer(amount)"
))
entries.append(g(
"Insecure Random",
"import random\ntoken = random.randint(100000, 999999)",
"import secrets\ntoken = secrets.randbelow(900000) + 100000\n# secrets uses OS-level CSPRNG"
))
entries.append(g(
"Sensitive Data in URL",
"requests.get(f\"https://api.com/users?id={user_id}&token={api_token}\")",
"requests.get(\n \"https://api.com/users\",\n params={\"id\": user_id},\n headers={\"Authorization\": f\"Bearer {api_token}\"}\n)"
))
entries.append(g(
"Missing Input Validation",
"@app.route('/api/user/<int:user_id>')\ndef get_user(user_id):\n return db.get_user(user_id)",
"@app.route('/api/user/<int:user_id>')\ndef get_user(user_id):\n if user_id <= 0:\n return {'error': 'Invalid ID'}, 400\n return db.get_user(user_id)"
))
entries.append(g(
"Race Condition in Token",
"def generate_token():\n token = str(uuid.uuid4())\n if token in existing_tokens:\n return generate_token()\n return token",
"def generate_token():\n token = secrets.token_urlsafe(32)\n # UUIDs + secrets are already unique enough\n # No need to check existence\n return token"
))
# =============================================================================
# PERFORMANCE (15 entries)
# =============================================================================
entries.append(g(
"String Concatenation in Loop",
"result = ''\nfor item in items:\n result += item",
"result = ''.join(items)\n# join() pre-allocates, += creates new string each time"
))
entries.append(g(
"Unnecessary List Copy",
"def process(items):\n data = list(items)\n return [x * 2 for x in data]",
"def process(items):\n return [x * 2 for x in items]\n# List comprehension creates a new list anyway"
))
entries.append(g(
"Repeated len() Call",
"for i in range(len(items)):\n print(items[i])",
"for i, item in enumerate(items):\n print(item)\n# enumerate is Pythonic and faster than range(len())"
))
entries.append(g(
"Global Variable Lookup",
"THRESHOLD = 100\ndef check(value):\n return value > THRESHOLD",
"THRESHOLD = 100\ndef check(value, threshold=THRESHOLD):\n return value > threshold\n# Default arg captures value at definition time (local lookup)"
))
entries.append(g(
"Inefficient Dict Lookup",
"if key in d and d[key] > 0:\n process(d[key])",
"val = d.get(key)\nif val is not None and val > 0:\n process(val)\n# .get() avoids double hash lookup"
))
entries.append(g(
"Missing __slots__",
"class Point:\n def __init__(self, x, y):\n self.x = x\n self.y = y\n# Creating millions of instances",
"class Point:\n __slots__ = ('x', 'y')\n def __init__(self, x, y):\n self.x = x\n self.y = y\n# Saves ~40% memory per instance"
))
entries.append(g(
"Linear Search in Large List",
"found = item in large_list\n# large_list has 1M+ items",
"found = item in large_set\n# Use set() for O(1) lookups instead of O(n)"
))
entries.append(g(
"Unnecessary Generator Materialization",
"total = sum([x ** 2 for x in range(1000000)])",
"total = sum(x ** 2 for x in range(1000000))\n# Generator expression saves memory (no intermediate list)"
))
entries.append(g(
"Inefficient String Formatting",
"name = 'world'\ngreeting = 'Hello ' + name + '!'",
"greeting = f'Hello {name}!'\n# f-strings are faster than concatenation"
))
entries.append(g(
"Missing Cache for Expensive Call",
"def get_config():\n with open('config.yaml') as f:\n return yaml.safe_load(f)\n# Called thousands of times",
"from functools import lru_cache\n@lru_cache(maxsize=1)\ndef get_config():\n with open('config.yaml') as f:\n return yaml.safe_load(f)"
))
entries.append(g(
"Blocking I/O in Main Thread",
"def handle_request():\n data = requests.get('https://api.com/data')\n # Blocks the entire server",
"async def handle_request():\n async with aiohttp.ClientSession() as session:\n async with session.get('https://api.com/data') as resp:\n return await resp.json()"
))
entries.append(g(
"Unnecessary Import in Function",
"def process(data):\n import json\n return json.loads(data)\n# Called in a hot loop",
"import json\ndef process(data):\n return json.loads(data)\n# Module import is cached but function-local still has overhead"
))
entries.append(g(
"Inefficient Regex",
"import re\nfor line in lines:\n match = re.match(r'\\d+', line)\n # Recompile regex each iteration",
"import re\nDIGIT_RE = re.compile(r'\\d+')\nfor line in lines:\n match = DIGIT_RE.match(line)\n# Compile once, reuse"
))
entries.append(g(
"Unnecessary deepcopy",
"import copy\nnew_config = copy.deepcopy(config)",
"import json\nnew_config = json.loads(json.dumps(config))\n# Or better: use copy.deepcopy only when needed\n# For simple dicts: {**config} is faster"
))
entries.append(g(
"GIL-Bound Parallelism",
"from concurrent.futures import ThreadPoolExecutor\nwith ThreadPoolExecutor() as pool:\n results = pool.map(process_data, items)\n# CPU-bound work",
"from concurrent.futures import ProcessPoolExecutor\nwith ProcessPoolExecutor() as pool:\n results = pool.map(process_data, items)\n# Use ProcessPoolExecutor for CPU-bound tasks"
))
# =============================================================================
# MODERN PATTERNS & BEST PRACTICES (15 entries)
# =============================================================================
entries.append(g(
"Type Hints Missing",
"def process_data(data, key):\n return data[key]",
"def process_data(data: dict[str, Any], key: str) -> Any:\n return data[key]\n# Type hints enable static analysis and IDE support"
))
entries.append(g(
"Using old-style String formatting",
"greeting = \"Hello %s, you are %d years old\" % (name, age)",
"greeting = f'Hello {name}, you are {age} years old'\n# f-strings (3.6+) are cleaner and faster"
))
entries.append(g(
"Missing Context Manager",
"f = open('data.txt')\ndata = f.read()\nf.close()\n# File left open if read() raises",
"with open('data.txt') as f:\n data = f.read()\n# Context manager guarantees cleanup"
))
entries.append(g(
"Bare except",
"try:\n process()\nexcept:\n pass",
"try:\n process()\nexcept (ValueError, TypeError) as e:\n logger.error('Processing failed: %s', e)\n# Catch specific exceptions"
))
entries.append(g(
"Mutable Default Arg",
"def add_item(item, items=[]):\n items.append(item)\n return items",
"def add_item(item, items=None):\n if items is None:\n items = []\n items.append(item)\n return items\n# Mutable defaults are shared across calls"
))
entries.append(g(
"Using list for stack operations",
"stack = []\nstack.append(1)\nstack.insert(0, 2) # O(n)",
"from collections import deque\nstack = deque()\nstack.append(1)\nstack.appendleft(2) # O(1)"
))
entries.append(g(
"Manual isinstance checks",
"def process(val):\n if isinstance(val, str):\n return val.upper()\n elif isinstance(val, int):\n return val * 2\n elif isinstance(val, list):\n return [process(x) for x in val]",
"from typing import Union\ndef process(val: Union[str, int, list]) -> Any:\n if isinstance(val, str):\n return val.upper()\n if isinstance(val, int):\n return val * 2\n if isinstance(val, list):\n return [process(x) for x in val]\n raise TypeError(f'Unsupported type: {type(val).__name__}')\n# Use structural pattern matching (3.10+)\nmatch val:\n case str():\n return val.upper()\n case int():\n return val * 2\n case list():\n return [process(x) for x in val]\n case _:\n raise TypeError(f'Unsupported: {type(val).__name__}')"
))
entries.append(g(
"Missing dataclass",
"class Config:\n def __init__(self, host, port, debug):\n self.host = host\n self.port = port\n self.debug = debug\n def __repr__(self):\n return f'Config({self.host}, {self.port})'",
"@dataclass\nclass Config:\n host: str\n port: int = 8080\n debug: bool = False"
))
entries.append(g(
"Using dict.get() with default factory",
"d = {}\nif key not in d:\n d[key] = []\nd[key].append(item)",
"from collections import defaultdict\nd = defaultdict(list)\nd[key].append(item)\n# defaultdict handles missing keys automatically"
))
entries.append(g(
"Manual Enum with constants",
"COLOR_RED = 1\nCOLOR_GREEN = 2\nCOLOR_BLUE = 3",
"from enum import Enum\nclass Color(Enum):\n RED = 1\n GREEN = 2\n BLUE = 3\n# Type-safe, IDE autocomplete, prevents typos"
))
entries.append(g(
"Not using pathlib",
"import os\npath = os.path.join('/data', 'files', 'data.txt')\nif os.path.exists(path):\n with open(path) as f:\n data = f.read()",
"from pathlib import Path\npath = Path('/data/files/data.txt')\nif path.exists():\n data = path.read_text()"
))
entries.append(g(
"Manual retry logic",
"for _ in range(3):\n try:\n return requests.get(url)\n except requests.RequestException:\n pass",
"from tenacity import retry, stop_after_attempt, wait_exponential\n@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))\ndef fetch(url):\n return requests.get(url).raise_for_status()"
))
entries.append(g(
"Not using typing for return values",
"def find_user(user_id):\n user = db.query(User).filter_by(id=user_id).first()\n if user:\n return user.to_dict()\n return None",
"from typing import Optional\ndef find_user(user_id: int) -> Optional[dict]:\n user = db.query(User).filter_by(id=user_id).first()\n if user:\n return user.to_dict()\n return None"
))
entries.append(g(
"Using __future__ annotations (recommended)",
"from typing import List, Dict\n# Old style annotations\nfrom __future__ import annotations\n# Forward refs work without quotes",
"# Already correct - from __future__ annotations enables:\n# 1. Forward references without quotes\n# 2. No need to import List, Dict from typing\n# 3. Annotations are strings by default (PEP 563)"
))
entries.append(g(
"Not using match/case (3.10+)",
"if status == 200:\n return handle_ok(data)\nelif status == 404:\n return handle_not_found()\nelif status == 500:\n return handle_error()\nelse:\n return handle_unknown()",
"match status:\n case 200:\n return handle_ok(data)\n case 404:\n return handle_not_found()\n case 500:\n return handle_error()\n case _:\n return handle_unknown()"
))
# =============================================================================
# BACKEND / WEB FRAMEWORKS (15 entries)
# =============================================================================
entries.append(g(
"Flask - Missing error handler",
"@app.route('/api/users')\ndef get_users():\n users = db.query(User).all()\n return jsonify([u.to_dict() for u in users])",
"@app.route('/api/users')\ndef get_users():\n try:\n users = db.query(User).all()\n return jsonify([u.to_dict() for u in users])\n except Exception as e:\n logger.error('Failed to fetch users: %s', e)\n return jsonify({'error': 'Internal error'}), 500\n\n@app.errorhandler(404)\ndef not_found(e):\n return jsonify({'error': 'Not found'}), 404"
))
entries.append(g(
"FastAPI - Synchronous endpoint",
"@app.get('/items')\ndef get_items():\n # Blocking DB call\n items = db.query(Item).all()\n return items",
"@app.get('/items')\nasync def get_items(session: AsyncSession = Depends(get_session)):\n result = await session.execute(select(Item))\n return result.scalars().all()"
))
entries.append(g(
"Django - N+1 Query",
"# views.py\ndef post_list(request):\n posts = Post.objects.all()\n # Template accesses post.author.name -> N+1\n return render(request, 'posts.html', {'posts': posts})",
"from django.db.models import Prefetch\ndef post_list(request):\n posts = Post.objects.select_related('author').all()\n # Single query with JOIN\n return render(request, 'posts.html', {'posts': posts})"
))
entries.append(g(
"Django - Raw SQL in QuerySet",
"results = User.objects.extra(\n where=[\"last_login > %s\"],\n params=[threshold]\n)",
"from django.db.models import Q\nresults = User.objects.filter(\n last_login__gt=threshold\n)"
))
entries.append(g(
"Django - Missing select_for_update",
"def transfer(from_acc, to_acc, amount):\n a1 = Account.objects.get(id=from_acc)\n a2 = Account.objects.get(id=to_acc)\n a1.balance -= amount\n a2.balance += amount\n a1.save()\n a2.save()",
"from django.db import transaction\n@transaction.atomic\ndef transfer(from_acc, to_acc, amount):\n a1 = Account.objects.select_for_update().get(id=from_acc)\n a2 = Account.objects.select_for_update().get(id=to_acc)\n a1.balance -= amount\n a2.balance += amount\n a1.save()\n a2.save()"
))
entries.append(g(
"FastAPI - Missing Pydantic validation",
"@app.post('/users')\ndef create_user(name: str, email: str):\n # No validation\n return {'name': name, 'email': email}",
"from pydantic import BaseModel, EmailStr\nclass UserCreate(BaseModel):\n name: str = Field(..., min_length=1, max_length=100)\n email: EmailStr\n\n@app.post('/users')\ndef create_user(user: UserCreate):\n # Automatically validated\n return {'name': user.name, 'email': user.email}"
))
entries.append(g(
"Flask - Not using Blueprint",
"# app.py - growing monolith\n@app.route('/users')\ndef users():\n ...\n@app.route('/orders')\ndef orders():\n ...",
"from flask import Blueprint\nuser_bp = Blueprint('users', __name__)\norder_bp = Blueprint('orders', __name__)\n\n# user_routes.py\n@user_bp.route('/users')\ndef users():\n ...\n\n# app.py\napp.register_blueprint(user_bp)\napp.register_blueprint(order_bp)"
))
entries.append(g(
"Missing request validation middleware",
"@app.route('/api/process', methods=['POST'])\ndef process():\n data = request.json\n name = data['name'] # KeyError if missing",
"@app.route('/api/process', methods=['POST'])\ndef process():\n data = request.get_json()\n if not data or 'name' not in data:\n return {'error': 'Missing name field'}, 400\n name = data['name']"
))
entries.append(g(
"Django REST - Serializer field exposure",
"class UserSerializer(serializers.ModelSerializer):\n class Meta:\n model = User\n fields = '__all__'",
"class UserSerializer(serializers.ModelSerializer):\n class Meta:\n model = User\n fields = ['id', 'username', 'email']\n # Never expose password, is_active, etc."
))
entries.append(g(
"FastAPI - Missing dependency injection",
"@app.get('/users/{user_id}')\ndef get_user(user_id: int):\n db = get_database_connection() # Called every request\n return db.get_user(user_id)",
"from fastapi import Depends\n@asynccontextmanager\nasync def get_db():\n db = SessionLocal()\n try:\n yield db\n finally:\n db.close()\n\n@app.get('/users/{user_id}')\ndef get_user(user_id: int, db: Session = Depends(get_db)):\n return db.get_user(user_id)"
))
entries.append(g(
"Flask - Missing app factory",
"from flask import Flask\napp = Flask(__name__)\napp.config['SECRET_KEY'] = os.environ['SECRET_KEY']\n\n@app.route('/')\ndef index():\n return 'Hello'",
"from flask import Flask\ndef create_app():\n app = Flask(__name__)\n app.config.from_object('config.Config')\n register_blueprints(app)\n return app"
))
entries.append(g(
"Missing pagination",
"@app.route('/api/items')\ndef list_items():\n items = Item.query.all()\n return jsonify([i.to_dict() for i in items])",
"@app.route('/api/items')\ndef list_items():\n page = request.args.get('page', 1, type=int)\n per_page = request.args.get('per_page', 20, type=int)\n items = Item.query.paginate(page=page, per_page=per_page)\n return jsonify({\n 'items': [i.to_dict() for i in items.items],\n 'total': items.total,\n 'pages': items.pages\n })"
))
entries.append(g(
"Django - Missing CSRF for AJAX",
"$.ajax({\n url: '/api/update',\n method: 'POST',\n data: {value: 42}\n})",
"$.ajax({\n url: '/api/update',\n method: 'POST',\n data: {value: 42, csrfmiddlewaretoken: '{{ csrf_token }}'},\n headers: {'X-CSRFToken': getCookie('csrftoken')}\n})"
))
entries.append(g(
"Flask - Not using flask-testing",
"# Manual test setup\ndef test_index():\n app = create_app()\n with app.test_client() as client:\n resp = client.get('/')\n assert resp.status_code == 200",
"import pytest\n@pytest.fixture\ndef client():\n app = create_app()\n app.config['TESTING'] = True\n with app.test_client() as client:\n yield client\n\ndef test_index(client):\n resp = client.get('/')\n assert resp.status_code == 200"
))
entries.append(g(
"FastAPI - Missing response model",
"@app.get('/users/{user_id}', response_model=dict)\ndef get_user(user_id: int):\n return db.get_user(user_id)",
"from pydantic import BaseModel\nclass UserOut(BaseModel):\n id: int\n username: str\n email: str\n class Config:\n from_attributes = True\n\n@app.get('/users/{user_id}', response_model=UserOut)\ndef get_user(user_id: int):\n return db.get_user(user_id)"
))
# =============================================================================
# ASYNC / CONCURRENCY (10 entries)
# =============================================================================
entries.append(g(
"Mixing sync and async",
"async def fetch_data():\n result = requests.get(url) # Blocking!\n return result.json()",
"async def fetch_data():\n async with aiohttp.ClientSession() as session:\n async with session.get(url) as resp:\n return await resp.json()"
))
entries.append(g(
"Unawaited coroutine",
"async def handle():\n send_notification(user_id) # Forgetting await\n return {'status': 'ok'}",
"async def handle():\n await send_notification(user_id) # Properly awaited\n return {'status': 'ok'}"
))
entries.append(g(
"asyncio.gather without error handling",
"results = await asyncio.gather(\n fetch_a(),\n fetch_b(),\n fetch_c()\n)",
"results = await asyncio.gather(\n fetch_a(),\n fetch_b(),\n fetch_c(),\n return_exceptions=True\n)\n# Prevents one failure from cancelling all tasks"
))
entries.append(g(
"Missing asyncio semaphore",
"async def process_all(urls):\n tasks = [fetch(url) for url in urls]\n return await asyncio.gather(*tasks)\n# Thousands of concurrent requests",
"async def process_all(urls):\n semaphore = asyncio.Semaphore(50)\n async def limited_fetch(url):\n async with semaphore:\n return await fetch(url)\n tasks = [limited_fetch(url) for url in urls]\n return await asyncio.gather(*tasks)"
))
entries.append(g(
"Blocking sleep in async",
"import time\nasync def delay():\n time.sleep(1) # Blocks event loop",
"import asyncio\nasync def delay():\n await asyncio.sleep(1) # Non-blocking"
))
entries.append(g(
"Race condition in asyncio",
"counter = 0\nasync def increment():\n global counter\n counter += 1",
"import asyncio\nlock = asyncio.Lock()\ncounter = 0\nasync def increment():\n global counter\n async with lock:\n counter += 1"
))
entries.append(g(
"asyncio Task cancellation",
"task = asyncio.create_task(long_running())\n# If cancelled, resources leaked",
"task = asyncio.create_task(long_running())\ntry:\n result = await task\nexcept asyncio.CancelledError:\n await cleanup_resources()\n raise"
))
entries.append(g(
"Missing contextvars propagation",
"async def handler(request):\n # Context not shared with child tasks\n asyncio.create_task(process())",
"import contextvars\nrequest_id = contextvars.ContextVar('request_id')\nasync def handler(request):\n request_id.set(request.id)\n await process() # Context propagates to child tasks"
))
entries.append(g(
"asyncio Queue without timeout",
"while True:\n item = await queue.get()\n process(item)\n# Blocks forever if queue is empty",
"while True:\n try:\n item = await asyncio.wait_for(queue.get(), timeout=5.0)\n process(item)\n except asyncio.TimeoutError:\n break # Graceful shutdown"
))
entries.append(g(
"Structural concurrency pattern",
"async def server():\n while True:\n reader, writer = await asyncio.get_running_loop().accept()\n asyncio.create_task(handle(reader, writer))",
"async def server():\n server = await asyncio.start_unix_server(handle)\n async with server:\n await server.serve_forever()\n\nasync def handle(reader, writer):\n try:\n while True:\n data = await reader.read(1024)\n if not data:\n break\n writer.write(data)\n finally:\n writer.close()"
))
# =============================================================================
# TESTING (10 entries)
# =============================================================================
entries.append(g(
"Testing implementation details",
"def test_process():\n obj = Processor()\n obj.process()\n assert obj._internal_counter == 5 # Testing private state",
"def test_process():\n obj = Processor()\n result = obj.process()\n assert result == expected_output # Test behavior, not internals"
))
entries.append(g(
"Missing test fixtures",
"def test_user_create():\n db.setup()\n user = User(name='Alice')\n db.add(user)\n ...\n db.teardown()\n\ndef test_user_update():\n db.setup() # Repeated setup\n user = User(name='Alice')\n db.add(user)\n ...",
"@pytest.fixture\ndef db_session():\n session = Session()\n yield session\n session.close()\n\n@pytest.fixture\ndef sample_user(db_session):\n user = User(name='Alice')\n db_session.add(user)\n db_session.commit()\n return user\n\ndef test_user_create(db_session, sample_user):\n ...\n\ndef test_user_update(db_session, sample_user):\n ..."
))
entries.append(g(
"Testing with real dependencies",
"def test_send_email():\n send_email('test@test.com', 'Hello')\n assert email_was_sent() # Hits real SMTP server",
"from unittest.mock import patch, MagicMock\n@patch('app.send_smtp')\ndef test_send_email(mock_smtp):\n send_email('test@test.com', 'Hello')\n mock_smtp.assert_called_once_with(\n to='test@test.com',\n body='Hello'\n )"
))
entries.append(g(
"Missing parameterized tests",
"def test_add_positive():\n assert add(2, 3) == 5\n\ndef test_add_negative():\n assert add(-1, -1) == -2\n\ndef test_add_zero():\n assert add(0, 0) == 0",
"@pytest.mark.parametrize('a,b,expected', [\n (2, 3, 5),\n (-1, -1, -2),\n (0, 0, 0),\n (100, -50, 50),\n])\ndef test_add(a, b, expected):\n assert add(a, b) == expected"
))
entries.append(g(
"Flaky time-dependent test",
"def test_timeout():\n start = time.time()\n slow_operation()\n elapsed = time.time() - start\n assert elapsed > 0.1 # Flaky on slow machines",
"def test_timeout(mocker):\n clock = mocker.patch('time.time')\n clock.return_value = 1000.0\n slow_operation()\n clock.assert_called() # Test logic, not timing"
))
entries.append(g(
"Missing test isolation",
"class TestUser:\n def setup_method(self):\n self.user = User('Alice')\n # Shared state between tests",
"class TestUser:\n @pytest.fixture(autouse=True)\n def clean_state(self):\n # Each test gets fresh state\n User.query.delete()\n yield\n db.session.rollback()"
))
entries.append(g(
"Testing exception handling",
"def test_invalid_input():\n result = divide(1, 0)\n assert result is None",
"import pytest\ndef test_invalid_input():\n with pytest.raises(ZeroDivisionError):\n divide(1, 0)"
))
entries.append(g(
"Missing async test",
"def test_async_fetch():\n # Can't test async function synchronously\n result = fetch_data() # Returns coroutine, not result",
"import pytest\n@pytest.mark.asyncio\nasync def test_async_fetch():\n result = await fetch_data()\n assert result['status'] == 'ok'"
))
entries.append(g(
"Over-mocking",
"@patch('db.connect')\n@patch('db.query')\n@patch('db.fetch')\n@patch('db.close')\n@patch('utils.format')\n@patch('utils.validate')\ndef test_create_user(mock_validate, mock_format, mock_close, ...):\n # Testing nothing real",
"def test_create_user(db_session):\n # Use test database, mock only external APIs\n user = User(name='Alice')\n db_session.add(user)\n db_session.commit()\n assert User.query.filter_by(name='Alice').first() is not None"
))
entries.append(g(
"Missing property-based tests",
"def test_sort():\n assert sort([3, 1, 2]) == [1, 2, 3]\n # One example, doesn't cover edge cases",
"from hypothesis import given, strategies as st\n@given(st.lists(st.integers(), min_size=0, max_size=100))\ndef test_sort_always_sorted(lst):\n result = sort(lst)\n assert result == sorted(lst)\n assert len(result) == len(lst)"
))
# =============================================================================
# DATA PROCESSING & ALGORITHMS (10 entries)
# =============================================================================
entries.append(g(
"Inefficient data processing",
"def count_words(text):\n counts = {}\n for word in text.split():\n if word in counts:\n counts[word] += 1\n else:\n counts[word] = 0",
"from collections import Counter\ndef count_words(text):\n return Counter(text.split())"
))
entries.append(g(
"List comprehension abuse",
"# Overly complex list comprehension\nresult = [x for x in data if x > 0 for y in other if y in x for z in third if z == y]",
"result = []\nfor x in data:\n if x <= 0:\n continue\n for y in other:\n if y in x:\n for z in third:\n if z == y:\n result.append(x)\n# Readability > cleverness"
))
entries.append(g(
"Using list for FIFO queue",
"queue = []\nqueue.append(item)\nitem = queue.pop(0) # O(n)",
"from collections import deque\nqueue = deque()\nqueue.append(item)\nitem = queue.popleft() # O(1)"
))
entries.append(g(
"Manual binary search",
"def find(arr, target):\n lo, hi = 0, len(arr) - 1\n while lo <= hi:\n mid = (lo + hi) // 2\n if arr[mid] == target:\n return mid\n elif arr[mid] < target:\n lo = mid + 1\n else:\n hi = mid - 1\n return -1",
"import bisect\ndef find(arr, target):\n idx = bisect.bisect_left(arr, target)\n if idx < len(arr) and arr[idx] == target:\n return idx\n return -1"
))
entries.append(g(
"Inefficient groupby",
"def group_by_key(items, key_fn):\n groups = {}\n for item in items:\n k = key_fn(item)\n if k not in groups:\n groups[k] = []\n groups[k].append(item)\n return groups",
"from itertools import groupby\ndef group_by_key(items, key_fn):\n return {\n k: list(v) for k, v in groupby(sorted(items, key=key_fn), key_fn)\n }"
))
entries.append(g(
"Missing itertools for combinations",
"def all_pairs(items):\n result = []\n for i in range(len(items)):\n for j in range(i+1, len(items)):\n result.append((items[i], items[j]))\n return result",
"from itertools import combinations\ndef all_pairs(items):\n return list(combinations(items, 2))"
))
entries.append(g(
"Inefficient deduplication",
"def deduplicate(items):\n seen = []\n result = []\n for item in items:\n if item not in seen: # O(n) lookup\n seen.append(item)\n result.append(item)\n return result",
"def deduplicate(items):\n seen = set()\n result = []\n for item in items:\n if item not in seen: # O(1) lookup\n seen.add(item)\n result.append(item)\n return result"
))
entries.append(g(
"Manual chunking",
"def chunks(lst, n):\n result = []\n for i in range(0, len(lst), n):\n result.append(lst[i:i+n])\n return result",
"from itertools import islice\ndef chunks(iterable, n):\n it = iter(iterable)\n while batch := list(islice(it, n)):\n yield batch\n# Generator - memory efficient for large data"
))
entries.append(g(
"Using pandas for simple CSV",
"import pandas as pd\ndf = pd.read_csv('data.csv')\nresult = df.groupby('category')['value'].sum()\n# Overkill for simple aggregation",
"import csv\nfrom collections import defaultdict\nresult = defaultdict(float)\nwith open('data.csv') as f:\n reader = csv.DictReader(f)\n for row in reader:\n result[row['category']] += float(row['value'])\n# No pandas dependency needed"
))
entries.append(g(
"Inefficient file processing",
"with open('large_file.txt') as f:\n lines = f.readlines() # Loads entire file\n for line in lines:\n process(line)",
"with open('large_file.txt') as f:\n for line in f: # Iterates line by line\n process(line)\n# Memory efficient for large files"
))
# =============================================================================
# DESIGN PATTERNS (10 entries)
# =============================================================================
entries.append(g(
"God class",
"class UserManager:\n def create_user(self): ...\n def send_email(self): ...\n def process_payment(self): ...\n def generate_report(self): ...\n def update_stats(self): ...",
"class UserService:\n def __init__(self, email_service, payment_service, stats_service):\n self.email = email_service\n self.payment = payment_service\n self.stats = stats_service\n\n def create_user(self):\n # Delegate concerns\n ...\n# Single Responsibility Principle"
))
entries.append(g(
"Missing Strategy Pattern",
"def calculate_price(order):\n if order.customer_type == 'regular':\n return order.total\n elif order.customer_type == 'vip':\n return order.total * 0.9\n elif order.customer_type == 'enterprise':\n return order.total * 0.8",
"from abc import ABC, abstractmethod\nclass PricingStrategy(ABC):\n @abstractmethod\n def apply(self, total: float) -> float: ...\n\nclass RegularPricing(PricingStrategy):\n def apply(self, total): return total\n\nclass VIPPricing(PricingStrategy):\n def apply(self, total): return total * 0.9\n\n# Open/Closed Principle - add new strategies without modifying existing code"
))
entries.append(g(
"Missing Observer Pattern",
"class EventEmitter:\n def __init__(self):\n self._callbacks = {}\n def on(self, event, callback):\n ...\n def emit(self, event, data):\n ...",
"from typing import Callable, Any\nclass EventEmitter:\n def __init__(self):\n self._listeners: dict[str, list[Callable]] = {}\n\n def on(self, event: str, callback: Callable) -> None:\n self._listeners.setdefault(event, []).append(callback)\n\n def emit(self, event: str, *args: Any, **kwargs: Any) -> None:\n for cb in self._listeners.get(event, []):\n cb(*args, **kwargs)"
))
entries.append(g(
"Hardcoded dependencies",
"class OrderProcessor:\n def __init__(self):\n self.db = DatabaseConnection()\n self.email = EmailService()\n self.logger = Logger()",
"class OrderProcessor:\n def __init__(\n self,\n db: DatabaseConnection,\n email: EmailService,\n logger: Logger\n ):\n self.db = db\n self.email = email\n self.logger = logger\n # Dependency Injection - testable, flexible"
))
entries.append(g(
"Missing Builder Pattern for complex objects",
"class HttpRequest:\n def __init__(self, method, url, headers, body, auth, timeout, retries):\n # 10+ parameters\n ...",
"class HttpRequestBuilder:\n def __init__(self):\n self._method = 'GET'\n self._url = ''\n self._headers = {}\n self._body = None\n self._timeout = 30\n\n def method(self, m): self._method = m; return self\n def url(self, u): self._url = u; return self\n def header(self, k, v): self._headers[k] = v; return self\n def body(self, b): self._body = b; return self\n def timeout(self, t): self._timeout = t; return self\n def build(self): return HttpRequest(**self.__dict__)"
))
entries.append(g(
"Missing Singleton for shared resource",
"class Database:\n def __init__(self):\n self.conn = connect() # Multiple connections",
"class Database:\n _instance = None\n def __new__(cls):\n if cls._instance is None:\n cls._instance = super().__new__(cls)\n cls._instance.conn = connect()\n return cls._instance\n # Or better: use dependency injection container"
))
entries.append(g(
"Missing Factory for polymorphic creation",
"if shape_type == 'circle':\n shape = Circle()\nelif shape_type == 'square':\n shape = Square()\nelif shape_type == 'triangle':\n shape = Triangle()",
"from abc import ABC\nclass Shape(ABC): pass\n\nclass ShapeFactory:\n _creators = {\n 'circle': Circle,\n 'square': Square,\n 'triangle': Triangle,\n }\n @classmethod\n def create(cls, shape_type: str) -> Shape:\n creator = cls._creators.get(shape_type)\n if not creator:\n raise ValueError(f'Unknown shape: {shape_type}')\n return creator()"
))
entries.append(g(
"Missing Repository Pattern",
"def get_user(user_id):\n session = Session()\n return session.query(User).filter_by(id=user_id).first()",
"class UserRepository:\n def __init__(self, session: Session):\n self.session = session\n\n def get_by_id(self, user_id: int) -> User:\n return self.session.query(User).filter_by(id=user_id).first()\n\n def save(self, user: User):\n self.session.add(user)\n self.session.commit()"
))
entries.append(g(
"Missing Middleware pattern",
"def handle_request(request):\n # Auth check\n # Rate limiting\n # Logging\n # CORS\n # Request processing\n # Error handling\n # Response formatting\n ...",
"class Middleware:\n def __init__(self, handler=None):\n self.handler = handler\n\n def process(self, request):\n # Pre-processing\n if self.handler:\n return self.handler.process(request)\n return None\n\nclass AuthMiddleware(Middleware): ...\nclass RateLimitMiddleware(Middleware): ...\n\n# Chain: AuthMiddleware(RateLimitMiddleware(handler))"
))
entries.append(g(
"Missing Unit of Work",
"def process_order(order):\n db.save(order)\n db.save(order.items)\n db.save(order.payment)\n # If payment fails, order and items already saved",
"from contextlib import contextmanager\n@contextmanager\ndef unit_of_work(session):\n try:\n yield session\n session.commit()\n except Exception:\n session.rollback()\n raise\n\n# All changes atomic"
))
# =============================================================================
# STANDARD LIBRARY & IDIOMS (10 entries)
# =============================================================================
entries.append(g(
"Manual flag variable",
"def contains_even(items):\n found = False\n for item in items:\n if item % 2 == 0:\n found = True\n break\n return found",
"def contains_even(items):\n return any(item % 2 == 0 for item in items)"
))
entries.append(g(
"Using dict.setdefault incorrectly",
"d.setdefault(key, []).append(item)\n# Creates list even if key exists",
"from collections import defaultdict\nd = defaultdict(list)\nd[key].append(item)\n# Or: d.setdefault(key, []).append(item) is actually correct\n# The issue is performance - defaultdict is cleaner"
))
entries.append(g(
"Manual min/max with loop",
"def find_largest(items):\n largest = items[0]\n for item in items[1:]:\n if item > largest:\n largest = item\n return largest",
"def find_largest(items):\n return max(items)\n# Built-in max/min are implemented in C"
))
entries.append(g(
"Using zip incorrectly",
"for i in range(len(names)):\n print(names[i], ages[i])",
"for name, age in zip(names, ages):\n print(name, age)\n# zip() pairs elements cleanly"
))
entries.append(g(
"Manual dict merge",
"merged = {}\nfor d in [d1, d2, d3]:\n merged.update(d)",
"merged = {**d1, **d2, **d3}\n# Python 3.5+ dict unpacking"
))
entries.append(g(
"Missing itertools.chain",
"result = []\nfor lst in lists:\n result.extend(lst)\nreturn result",
"from itertools import chain\nreturn list(chain.from_iterable(lists))"
))
entries.append(g(
"Using sorted with key function incorrectly",
"items.sort(key=lambda x: (x[0], x[1], x[2]))\n# Unnecessary lambda",
"items.sort(key=lambda x: x) # Already tuples\n# Or: items.sort() for default tuple comparison"
))
entries.append(g(
"Manual enum with integers",
"STATUS_DRAFT = 0\nSTATUS_PENDING = 1\nSTATUS_PUBLISHED = 2",
"from enum import IntEnum\nclass Status(IntEnum):\n DRAFT = 0\n PENDING = 1\n PUBLISHED = 2\n# Type-safe, iterable, comparable"
))
entries.append(g(
"Using list as boolean check",
"if len(items) > 0:\n process(items)",
"if items:\n process(items)\n# Empty sequences are falsy"
))
entries.append(g(
"Missing walrus operator",
"result = search(data)\nif result:\n print(result)",
"if result := search(data):\n print(result)\n# Assignment expression (Python 3.8+)"
))
# =============================================================================
# DEPLOYMENT & OPS (10 entries)
# =============================================================================
entries.append(g(
"Missing requirements.txt pinning",
"requests==2.28\nflask\nnumpy",
"flask==2.3.2\nrequests==2.28.2\nnumpy==1.24.3\n# Pin all dependencies for reproducible builds"
))
entries.append(g(
"Running as root in Docker",
"FROM python:3.11\nRUN pip install -r requirements.txt\nCMD [\"python\", \"app.py\"]",
"FROM python:3.11-slim\nWORKDIR /app\nCOPY requirements.txt .\nRUN pip install --no-cache-dir -r requirements.txt\nCOPY . .\nRUN useradd -m appuser\nUSER appuser\nCMD [\"python\", \"app.py\"]"
))
entries.append(g(
"Missing health check",
"app = Flask(__name__)",
"app = Flask(__name__)\n\n@app.route('/health')\ndef health():\n return {'status': 'healthy'}\n\n# Docker HEALTHCHECK\n# docker build --health-cmd='curl -f http://localhost:5000/health'"
))
entries.append(g(
"Not using gunicorn for production",
"if __name__ == '__main__':\n app.run(host='0.0.0.0', port=5000)",
"# Use gunicorn:\n# gunicorn --workers 4 --bind 0.0.0.0:5000 app:app\n# Or uvicorn for async:\n# uvicorn app:app --host 0.0.0.0 --port 5000 --workers 4"
))
entries.append(g(
"Missing environment variable handling",
"DB_HOST = 'localhost'\nDB_PORT = 5432\nDB_NAME = 'mydb'",
"import os\nfrom dotenv import load_dotenv\nload_dotenv()\n\nDB_HOST = os.getenv('DB_HOST', 'localhost')\nDB_PORT = int(os.getenv('DB_PORT', '5432'))\nDB_NAME = os.environ['DB_NAME'] # Required - fails early if missing"
))
entries.append(g(
"Not using virtual environment",
"pip install flask requests\npython app.py",
"python -m venv .venv\nsource .venv/bin/activate # Linux/Mac\n# .venv\\Scripts\\activate # Windows\npip install -r requirements.txt\npython app.py"
))
entries.append(g(
"Missing logging configuration",
"import logging\nlogger = logging.getLogger(__name__)\nlogger.info('Processing...')\n# No handler configured",
"import logging\nlogging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s %(levelname)s %(name)s: %(message)s'\n)\nlogger = logging.getLogger(__name__)"
))
entries.append(g(
"Hardcoded database URL",
"SQLALCHEMY_DATABASE_URI = 'postgresql://user:pass@localhost/db'",
"import os\nSQLALCHEMY_DATABASE_URI = os.environ.get(\n 'DATABASE_URL',\n 'postgresql://user:pass@localhost/db'\n)"
))
entries.append(g(
"Missing graceful shutdown",
"import signal\nimport sys\ndef shutdown(signum, frame):\n sys.exit(0)",
"import signal\nimport sys\n\ndef graceful_shutdown(signum, frame):\n logger.info('Shutting down gracefully...')\n # Close connections\n # Finish current requests\n # Cleanup resources\n sys.exit(0)\n\nsignal.signal(signal.SIGTERM, graceful_shutdown)\nsignal.signal(signal.SIGINT, graceful_shutdown)"
))
entries.append(g(
"Not using pre-commit hooks",
"# No code quality checks",
"# .pre-commit-config.yaml\nrepos:\n - repo: https://github.com/psf/black\n rev: 23.7.0\n hooks:\n - id: black\n - repo: https://github.com/pycqa/flake8\n rev: 6.1.0\n hooks:\n - id: flake8\n - repo: https://github.com/pycqa/isort\n rev: 5.12.0\n hooks:\n - id: isort\n\n# pip install pre-commit\n# pre-commit install"
))
# =============================================================================
# SAVE
# =============================================================================
random.seed(42) # Reproducible shuffle
random.shuffle(entries)
os.makedirs(BASE, exist_ok=True)
out = BASE / "dev_finetuning_python.json"
with open(out, 'w') as f:
json.dump(entries, f, indent=2)
# Count by tag category
categories = {}
for e in entries:
for tag in e['tags']:
if tag != 'python':
categories[tag] = categories.get(tag, 0) + 1
print(f"Total entries: {len(entries)}")
print(f"Categories:")
for cat, count in sorted(categories.items(), key=lambda x: -x[1]):
print(f" {cat}: {count}")
print(f"Saved: {out}")