#!/usr/bin/env python3
"""
Collect high-quality code dataset for fine-tuning a dev-focused model.
Sources:
1. Stack Overflow (generated Q&A pairs from common patterns)
2. GitHub repos (READMEs + code samples)
3. Code patterns (best practices, anti-patterns, bug fixes)
4. API documentation
5. Code review examples
6. Architecture patterns
7. Testing patterns
8. Database patterns
9. DevOps/CI patterns
Target: 2000+ training pairs
"""
import json
import os
import sys
import time
from pathlib import Path
from datetime import datetime
import requests
from bs4 import BeautifulSoup
BASE_DIR = Path(__file__).parent.parent
RAW_DIR = BASE_DIR / "data" / "raw"
PROCESSED_DIR = BASE_DIR / "data" / "processed"
REPO_SOURCES = [
"psf/requests", "pallets/flask", "django/django", "getmoto/moto",
"vercel/next.js", "facebook/react", "microsoft/TypeScript",
"rust-lang/rust", "serde-rs/serde", "golang/go",
"torvalds/linux", "microsoft/vscode", "kubernetes/kubernetes",
"docker/compose", "nginx/nginx", "redis/redis",
"pallets/werkzeug", "pydantic/pydantic", "tiangolo/fastapi",
"encode/starlette", "sqlalchemy/sqlalchemy", "psf/black",
]
def generate_python_patterns():
"""Generate Python best practice patterns."""
patterns = []
# Error handling patterns
patterns.append({
"instruction": "What's the proper way to handle errors in Python with context?",
"output": '''Python error handling best practices:
```python
# BAD - bare except
try:
do_something()
except:
pass
# BAD - too broad
try:
do_something()
except Exception as e:
print(e)
# GOOD - specific with context
try:
result = do_something()
except ValueError as e:
logger.error(f"Invalid input: {e}")
raise
except ConnectionError as e:
logger.error(f"Connection failed: {e}", exc_info=True)
# Retry or fallback
raise ServiceUnavailable() from e
```
Key principles:
- Never use bare `except` - catches KeyboardInterrupt, SystemExit
- Log with context: what operation failed, input values
- Use `raise ... from e` to preserve the original traceback
- Catch specific exceptions, not broad ones
- Fail fast - don't swallow exceptions silently''',
"source": "python_pattern",
"tags": ["python", "error_handling", "best_practice"]
})
# Type hints patterns
patterns.append({
"instruction": "How do you write proper type hints for complex Python functions?",
"output": '''Complex type hints in Python 3.10+:
```python
from typing import (
Any, Callable, Dict, Generic, Iterable, List, Optional,
Protocol, Sequence, TypeVar, Union, Literal, overload
)
# Generic function
T = TypeVar('T')
def first(items: Sequence[T]) -> T:
"""Get first item from sequence."""
return items[0]
# Union types (use | in Python 3.10+)
def parse_value(raw: str | int | float) -> float:
return float(raw)
# Protocol for duck typing
class Stringifiable(Protocol):
def __str__(self) -> str: ...
def format_output(obj: Stringifiable) -> str:
return f"Result: {obj}"
# Overloads for different return types
@overload
def get_config(key: str, as_json: Literal[True]) -> dict: ...
@overload
def get_config(key: str, as_json: Literal[False] = ...) -> str: ...
def get_config(key: str, as_json: bool = False) -> str | dict:
data = load(key)
return json.loads(data) if as_json else data
# Return type annotations
def process(
items: List[dict],
transform: Callable[[dict], dict] | None = None,
*, # keyword-only args after this
batch_size: int = 100,
retries: int = 3,
) -> Iterator[dict]:
...
```
Tips:
- Use `|` for unions in Python 3.10+ instead of `Union`
- `Protocol` for structural subtyping (duck typing with types)
- `@overload` when return type depends on input
- `*` to enforce keyword-only arguments after positional ones''',
"source": "python_pattern",
"tags": ["python", "type_hints", "typing"]
})
# Async patterns
patterns.append({
"instruction": "What are the key patterns for writing robust async Python code?",
"output": '''Async Python patterns:
```python
import asyncio
import aiohttp
from typing import List
# Pattern 1: Concurrent requests with semaphore
class APIClient:
def __init__(self, base_url: str, max_concurrent: int = 10):
self.base_url = base_url
self.semaphore = asyncio.Semaphore(max_concurrent)
async def fetch(self, session: aiohttp.ClientSession, path: str):
async with self.semaphore: # Limit concurrency
async with session.get(f"{self.base_url}{path}") as resp:
return await resp.json()
async def fetch_many(self, paths: List[str]) -> List[dict]:
async with aiohttp.ClientSession() as session:
tasks = [self.fetch(session, p) for p in paths]
# Return first failure, or all results
return await asyncio.gather(*tasks, return_exceptions=True)
# Pattern 2: Retry with exponential backoff
async def fetch_with_retry(url: str, max_retries: int = 3):
for attempt in range(max_retries):
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_retries - 1:
raise
wait = 2 ** attempt # 1s, 2s, 4s
await asyncio.sleep(wait)
# Pattern 3: Async context manager
class AsyncResource:
async def __aenter__(self):
self.conn = await connect()
return self
async def __aexit__(self, *exc):
await self.conn.close()
async def query(self, sql: str):
return await self.conn.execute(sql)
# Usage: async with AsyncResource() as db: result = await db.query(...)
# Pattern 4: Task groups (Python 3.11+)
async def process_all(items):
async with asyncio.TaskGroup() as tg:
futures = {tg.create_task(process(item)) for item in items}
# If any task fails, TaskGroup raises first exception
# All other tasks are cancelled automatically
```
Critical rules:
- Never use `time.sleep()` in async code - use `asyncio.sleep()`
- Use `asyncio.Semaphore` to limit concurrent operations
- Always use context managers (`async with`) for resources
- `asyncio.gather` for fan-out, `TaskGroup` for error propagation''',
"source": "python_pattern",
"tags": ["python", "async", "concurrency"]
})
# Testing patterns
patterns.append({
"instruction": "How do you write effective unit tests with pytest?",
"output": '''Pytest testing patterns:
```python
import pytest
from unittest.mock import patch, MagicMock
from myapp import User, create_user, validate_email
# Pattern 1: Parametrize for multiple test cases
@pytest.mark.parametrize("email,expected", [
("user@example.com", True),
("invalid", False),
("@nope.com", False),
("user@", False),
("user@.com", False),
])
def test_validate_email(email, expected):
assert validate_email(email) == expected
# Pattern 2: Fixtures for setup
@pytest.fixture
def db_session():
session = create_session()
yield session
session.close() # Teardown
@pytest.fixture
def sample_user(db_session):
user = User(name="Test", email="test@example.com")
db_session.add(user)
db_session.commit()
return user
def test_user_creation(sample_user):
assert sample_user.name == "Test"
assert sample_user.id is not None
# Pattern 3: Mock external dependencies
@patch("myapp.requests.post")
def test_create_user_external(mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {"id": 1}
user = create_user("test@example.com")
assert user.id == 1
mock_post.assert_called_once()
# Pattern 4: Test exceptions
def test_invalid_input():
with pytest.raises(ValueError, match="must be positive"):
calculate(-1)
# Pattern 5: Async tests
@pytest.mark.asyncio
async def test_async_fetch():
result = await fetch_data("/api/users")
assert len(result) > 0
# Pattern 6: Property-based testing (hypothesis)
from hypothesis import given, strategies as st
@given(st.integers(min_value=0, max_value=1000))
def test_sort_preserves_length(n):
data = list(range(n))
assert len(sorted(data)) == n
```
Best practices:
- One assertion per test when possible
- Use fixtures over setUp/tearDown
- Mock external services, don't hit real APIs in unit tests
- Parametrize for multiple inputs
- Test edge cases: empty, None, negative, max values''',
"source": "python_pattern",
"tags": ["python", "testing", "pytest"]
})
# Design patterns
patterns.append({
"instruction": "What are the essential Python design patterns every developer should know?",
"output": '''Essential Python design patterns:
**1. Singleton (Module-level)**
```python
# Python modules are singletons by default
# config.py
DATABASE_URL = os.environ["DATABASE_URL"]
# Import anywhere - same instance
from config import DATABASE_URL
```
**2. Factory Pattern**
```python
from abc import ABC, abstractmethod
class Notifier(ABC):
@abstractmethod
def send(self, message: str) -> None: ...
class EmailNotifier(Notifier):
def send(self, message): ...
class SMSNotifier(Notifier):
def send(self, message): ...
class NotifierFactory:
_registry = {"email": EmailNotifier, "sms": SMSNotifier}
@classmethod
def create(cls, channel: str) -> Notifier:
notifier_cls = cls._registry.get(channel)
if not notifier_cls:
raise ValueError(f"Unknown channel: {channel}")
return notifier_cls()
```
**3. Strategy Pattern**
```python
from typing import Protocol
class SortingStrategy(Protocol):
def sort(self, data: list) -> list: ...
class QuickSort:
def sort(self, data): ...
class MergeSort:
def sort(self, data): ...
class DataProcessor:
def __init__(self, strategy: SortingStrategy):
self.strategy = strategy
def process(self, data):
return self.strategy.sort(data)
# Usage: processor = DataProcessor(QuickSort())
```
**4. Observer Pattern (Pythonic)**
```python
import weakref
class EventEmitter:
def __init__(self):
self._listeners = weakref.WeakSet()
def subscribe(self, listener):
self._listeners.add(listener)
def emit(self, event, **data):
for listener in list(self._listeners):
listener(event, **data)
```
**5. Builder Pattern**
```python
class QueryBuilder:
def __init__(self, table: str):
self.table = table
self._conditions = []
self._limit_val = None
def where(self, column: str, value):
self._conditions.append(f"{column} = ?")
return self # Method chaining
def limit(self, n: int):
self._limit_val = n
return self
def build(self) -> str:
query = f"SELECT * FROM {self.table}"
if self._conditions:
query += " WHERE " + " AND ".join(self._conditions)
if self._limit_val:
query += f" LIMIT {self._limit_val}"
return query
# Usage: QueryBuilder("users").where("active", 1).limit(10).build()
```''',
"source": "python_pattern",
"tags": ["python", "design_patterns", "architecture"]
})
# Data validation
patterns.append({
"instruction": "How do you validate and parse data in Python using Pydantic?",
"output": '''Pydantic data validation patterns:
```python
from pydantic import BaseModel, Field, validator, field_validator
from pydantic import EmailStr, HttpUrl, confloat, conint
from typing import List, Optional
from enum import Enum
class Role(str, Enum):
ADMIN = "admin"
USER = "user"
MODERATOR = "moderator"
class UserInput(BaseModel):
email: EmailStr
age: conint(ge=0, le=150) # 0-150
score: confloat(ge=0.0, le=1.0) # 0.0-1.0
role: Role = Role.USER
tags: List[str] = Field(default_factory=list, max_length=10)
website: Optional[HttpUrl] = None
@field_validator("tags")
@classmethod
def tags_must_be_lowercase(cls, v):
return [tag.lower() for tag in v]
# Parsing and validation
data = {"email": "user@example.com", "age": 25, "score": 0.95}
user = UserInput(**data) # Validates automatically
# Model with computed fields
from pydantic import computed_field
class Product(BaseModel):
price: float
tax_rate: float = 0.1
@computed_field
@property
def price_with_tax(self) -> float:
return round(self.price * (1 + self.tax_rate), 2)
# Nested models
class OrderItem(BaseModel):
product_id: int
quantity: conint(ge=1)
class Order(BaseModel):
items: List[OrderItem]
@computed_field
@property
def total_items(self) -> int:
return sum(item.quantity for item in self.items)
# Settings management
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
debug: bool = False
api_key: str
model_config = {"env_file": ".env", "case_sensitive": False}
settings = Settings() # Loads from .env automatically
```
Key concepts:
- Type annotations ARE the validation rules
- `conint`/`confloat` for constrained numbers
- Custom validators with `@field_validator`
- `computed_field` for derived properties
- `BaseSettings` for config from env vars/files''',
"source": "python_pattern",
"tags": ["python", "validation", "pydantic"]
})
return patterns
def generate_javascript_patterns():
"""Generate JavaScript/TypeScript patterns."""
patterns = []
patterns.append({
"instruction": "What are common JavaScript closures mistakes and how to fix them?",
"output": '''JavaScript closure pitfalls:
**1. Loop variable capture**
```javascript
// BAD - all callbacks log 5
for (var i = 0; i < 5; i++) {
setTimeout(() => console.log(i), 100);
}
// FIX 1 - Use let (block-scoped)
for (let i = 0; i < 5; i++) {
setTimeout(() => console.log(i), 100); // 0,1,2,3,4
}
// FIX 2 - IIFE (for legacy)
for (var i = 0; i < 5; i++) {
((j) => setTimeout(() => console.log(j), 100))(i);
}
```
**2. Memory leaks from closures**
```javascript
// BAD - largeData held in memory by closure
function processData(largeData) {
const result = largeData.filter(x => x > 0);
return () => result.length; // Closes over largeData!
}
// FIX - only close over what you need
function processData(largeData) {
const count = largeData.filter(x => x > 0).length;
return () => count; // Just a number, no leak
}
```
**3. This binding in callbacks**
```javascript
// BAD - 'this' is undefined in arrow-free callback
class Timer {
constructor() { this.count = 0; }
start() {
setInterval(function() { this.count++; }, 1000); // this = window/undefined
}
}
// FIX 1 - Arrow function (inherits this)
start() {
setInterval(() => { this.count++; }, 1000);
}
// FIX 2 - Bind
start() {
setInterval(function() { this.count++; }.bind(this), 1000);
}
```
**4. Event listener accumulation**
```javascript
// BAD - adds listener every call, memory leak
function setupButton(btn) {
btn.addEventListener('click', () => handleClick());
}
// FIX - Remove before adding, or use one-time
function setupButton(btn) {
btn.replaceWith(btn.cloneNode(true));
btn.addEventListener('click', handleClick, { once: true });
}
```''',
"source": "javascript_pattern",
"tags": ["javascript", "closures", "common_mistakes"]
})
patterns.append({
"instruction": "How do you properly handle async/await error cases in JavaScript?",
"output": '''Async/await error handling patterns:
```javascript
// Pattern 1: Try/catch with specific errors
async function fetchUser(id) {
try {
const resp = await fetch(`/api/users/${id}`);
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
return await resp.json();
} catch (error) {
if (error.name === 'TypeError') {
console.error('Network error:', error);
} else {
console.error('API error:', error);
}
throw error; // Re-throw after logging
}
}
// Pattern 2: Race condition handling
async function fetchWithTimeout(url, timeout = 5000) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeout);
try {
const resp = await fetch(url, { signal: controller.signal });
return await resp.json();
} catch (error) {
if (error.name === 'AbortError') {
throw new Error(`Request timed out after ${timeout}ms`);
}
throw error;
} finally {
clearTimeout(timeoutId);
}
}
// Pattern 3: Concurrent with error isolation
async function fetchAll(urls) {
const results = await Promise.allSettled(
urls.map(url => fetch(url).then(r => r.json()))
);
return results.map((result, i) => ({
url: urls[i],
success: result.status === 'fulfilled',
data: result.status === 'fulfilled' ? result.value : null,
error: result.status === 'rejected' ? result.reason : null,
}));
}
// Pattern 4: Retry with backoff
async function fetchWithRetry(url, retries = 3) {
for (let i = 0; i < retries; i++) {
try {
const resp = await fetch(url);
if (resp.ok) return await resp.json();
throw new Error(`HTTP ${resp.status}`);
} catch (error) {
if (i === retries - 1) throw error;
await new Promise(r => setTimeout(r, 1000 * Math.pow(2, i)));
}
}
}
```
Key rules:
- Always wrap await in try/catch
- Use `Promise.allSettled` when you want all results despite failures
- Use `AbortController` for cancellable requests
- Retry transient errors, not logic errors''',
"source": "javascript_pattern",
"tags": ["javascript", "async", "error_handling"]
})
patterns.append({
"instruction": "What are essential TypeScript patterns for type-safe code?",
"output": '''Essential TypeScript patterns:
```typescript
// 1. Generic utility types
type DeepPartial<T> = {
[P in keyof T]?: T[P] extends object ? DeepPartial<T[P]> : T[P];
};
type ReadonlyDeep<T> = {
readonly [P in keyof T]: T[P] extends object ? ReadonlyDeep<T[P]> : T[P];
};
// 2. Discriminated unions
type Action =
| { type: 'ADD'; payload: number }
| { type: 'REMOVE'; payload: number }
| { type: 'CLEAR' };
function handle(action: Action) {
switch (action.type) {
case 'ADD':
return action.payload; // TS knows payload exists
case 'REMOVE':
return -action.payload;
case 'CLEAR':
return 0; // TS knows no payload
}
}
// 3. Template literal types
type HttpMethod = 'GET' | 'POST' | 'PUT' | 'DELETE';
type Endpoint = '/users' | '/posts' | '/comments';
type APIRoute = `/${HttpMethod.toLowerCase()}${Endpoint}`;
// '/get/users' | '/post/users' | ...
// 4. Type-safe event emitter
type Events = {
'user:created': { userId: string; name: string };
'user:deleted': { userId: string };
'error': { code: number; message: string };
};
class TypedEmitter<K extends keyof Events> {
private listeners: Partial<Record<K, Array<(data: Events[K]) => void>>> = {};
on<E extends K>(event: E, listener: (data: Events[E]) => void) {
this.listeners[event] = [...(this.listeners[event] || []), listener];
}
emit<E extends K>(event: E, data: Events[E]) {
this.listeners[event]?.forEach(fn => fn(data));
}
}
// 5. Result type (no exceptions)
type Result<T, E = Error> =
| { ok: true; value: T }
| { ok: false; error: E };
async function fetchUser(id: string): Promise<Result<User>> {
try {
const user = await api.get(`/users/${id}`);
return { ok: true, value: user };
} catch (error) {
return { ok: false, error: error as Error };
}
}
// Usage
const result = await fetchUser('123');
if (result.ok) {
console.log(result.value.name); // Type-safe
}
```
Best practices:
- Use `satisfies` operator for type checking without widening
- Discriminated unions for state machines
- Generic constraints with `extends`
- `Result<T>` pattern instead of exceptions for control flow''',
"source": "typescript_pattern",
"tags": ["typescript", "generics", "type_safety"]
})
return patterns
def generate_rust_patterns():
"""Generate Rust patterns."""
patterns = []
patterns.append({
"instruction": "What are the core Rust ownership and borrowing concepts?",
"output": '''Rust ownership rules:
**The Three Rules:**
1. Each value has exactly one owner
2. When the owner goes out of scope, the value is dropped
3. You can have EITHER one mutable reference OR any number of immutable references
```rust
// Ownership transfer
fn main() {
let s1 = String::from("hello");
let s2 = s1; // s1 is MOVED, no longer valid
// println!("{}", s1); // ERROR: value borrowed here after move
}
// Borrowing (no transfer)
fn calculate_length(s: &String) -> usize { // &String = borrowed reference
s.len()
} // s goes out of scope, but nothing is dropped (we don't own it)
// Mutable borrowing
fn change(s: &mut String) { // &mut String = mutable reference
s.push_str(", world");
}
// Rule: can't have mutable + immutable refs at same time
let mut s = String::from("hello");
let r1 = &s; // immutable borrow
let r2 = &s; // another immutable borrow - OK
// let r3 = &mut s; // ERROR: can't borrow as mutable while immutable borrows exist
println!("{} {}", r1, r2);
// r1, r2 no longer used after here
let r3 = &mut s; // OK now - immutable borrows are done
println!("{}", r3);
```
**Lifetimes** - telling the compiler references are valid:
```rust
fn longest<'a>(x: &'a str, y: &'a str) -> &'a str {
if x.len() > y.len() { x } else { y }
}
// The returned reference lives as long as the shorter of x and y
```
**Key patterns:**
- Use `&T` for read-only access, `&mut T` for mutation
- `Clone` when you need actual copies: `let s2 = s1.clone();`
- `Copy` trait for simple types (int, bool, f32) - they copy, don't move
- Lifetimes ensure references don't outlive the data they point to''',
"source": "rust_pattern",
"tags": ["rust", "ownership", "borrowing"]
})
patterns.append({
"instruction": "How do you handle errors idiomatically in Rust?",
"output": '''Rust error handling:
```rust
// Pattern 1: Result for recoverable errors
fn divide(a: f64, b: f64) -> Result<f64, String> {
if b == 0.0 {
Err(String::from("Division by zero"))
} else {
Ok(a / b)
}
}
// Pattern 2: ? operator for propagation
use std::fs::File;
use std::io::{self, Read};
fn read_file(path: &str) -> Result<String, io::Error> {
let mut file = File::open(path)?; // Propagate error
let mut contents = String::new();
file.read_to_string(&mut contents)?; // Propagate error
Ok(contents)
}
// Pattern 3: Custom error types
use thiserror::Error;
#[derive(Debug, Error)]
enum AppError {
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
#[error("Parse error: {0}")]
Parse(#[from] std::num::ParseIntError),
#[error("Not found: {resource}")]
NotFound { resource: String },
}
fn process(id: &str) -> Result<i32, AppError> {
let content = std::fs::read_to_string(id)?; // ? auto-converts via From
let number = content.trim().parse()?; // ? auto-converts via From
Ok(number)
}
// Pattern 4: anyhow for application code
use anyhow::{Context, Result};
fn run() -> Result<()> {
let content = std::fs::read_to_string("config.json")
.context("Failed to read config file")?; // Adds context to error
let config: Config = serde_json::from_str(&content)
.context("Failed to parse config JSON")?;
Ok(())
}
// Pattern 5: panic! for unrecoverable
fn main() {
let values = vec![1, 2, 3];
let fourth = values[3]; // PANIC: index out of bounds
// Better: use get() and handle
match values.get(3) {
Some(val) => println!("{}", val),
None => eprintln!("Index 3 out of bounds"),
}
}
```
When to use what:
- `Result<T, E>` - recoverable errors (file not found, parse error)
- `panic!` - programming bugs (index out of bounds, logic error)
- `?` operator - propagate errors up the call chain
- `thiserror` - library code (define error types)
- `anyhow` - application code (flexible error handling)''',
"source": "rust_pattern",
"tags": ["rust", "error_handling", "result"]
})
return patterns
def generate_golang_patterns():
"""Generate Go patterns."""
patterns = []
patterns.append({
"instruction": "What are essential Go error handling patterns?",
"output": '''Go error handling:
```go
// Pattern 1: Basic error wrapping (Go 1.20+)
func readFile(path string) ([]byte, error) {
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("reading file %s: %w", path, err)
}
return data, nil
}
// Pattern 2: Sentinel errors
var (
ErrNotFound = errors.New("not found")
ErrUnauthorized = errors.New("unauthorized")
ErrInvalidInput = errors.New("invalid input")
)
func getUser(id string) (*User, error) {
if id == "" {
return nil, ErrInvalidInput
}
user, ok := users[id]
if !ok {
return nil, ErrNotFound
}
return user, nil
}
// Check with errors.Is
func handle(id string) {
_, err := getUser(id)
if errors.Is(err, ErrNotFound) {
log.Println("User doesn't exist")
}
}
// Pattern 3: Custom error types
type ValidationError struct {
Field string
Message string
}
func (e *ValidationError) Error() string {
return fmt.Sprintf("validation failed for %s: %s", e.Field, e.Message)
}
// Pattern 4: Context cancellation
func fetchData(ctx context.Context, url string) ([]byte, error) {
req, _ := http.NewRequest("GET", url, nil)
req = req.WithContext(ctx)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, fmt.Errorf("request failed: %w", err)
}
defer resp.Body.Close()
return io.ReadAll(resp.Body)
}
// Pattern 5: Concurrent with errgroup
import "golang.org/x/sync/errgroup"
func fetchAll(ctx context.Context, urls []string) error {
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(5) // Max 5 concurrent
for _, url := range urls {
url := url // Capture for goroutine
g.Go(func() error {
_, err := fetchData(ctx, url)
return err // First error stops all
})
}
return g.Wait()
}
```
Rules:
- Don't ignore errors: `result, err := fn(); if err != nil { ... }`
- Use `%w` for wrapping, `%v` for formatting
- Sentinel errors for known error cases
- `errors.Is` / `errors.As` for checking wrapped errors
- `context.Context` for cancellation and timeouts''',
"source": "go_pattern",
"tags": ["go", "error_handling", "concurrency"]
})
return patterns
def generate_system_design_patterns():
"""Generate system design and architecture patterns."""
patterns = []
patterns.append({
"instruction": "How do you design a REST API following best practices?",
"output": '''REST API design best practices:
```
# Resource naming
GET /api/v1/users # List users
GET /api/v1/users/:id # Get user
POST /api/v1/users # Create user
PUT /api/v1/users/:id # Replace user
PATCH /api/v1/users/:id # Partial update
DELETE /api/v1/users/:id # Delete user
# Nested resources
GET /api/v1/users/:id/posts # User's posts
GET /api/v1/posts/:id/comments # Post comments
POST /api/v1/posts/:id/comments # Add comment
# Filtering, sorting, pagination
GET /api/v1/users?status=active&sort=-created_at&page=2&per_page=20
# Sub-resources via query
GET /api/v1/users?expand=posts,profile # Include related data
```
**Status codes:**
- 200 OK - successful GET, PUT, PATCH
- 201 Created - successful POST
- 204 No Content - successful DELETE
- 400 Bad Request - validation error
- 401 Unauthorized - not authenticated
- 403 Forbidden - authenticated but no permission
- 404 Not Found - resource doesn't exist
- 409 Conflict - duplicate or state conflict
- 422 Unprocessable Entity - semantic error
- 429 Too Many Requests - rate limited
- 500 Internal Server Error - unexpected error
**Response format:**
```json
// Success
{
"data": { "id": 1, "name": "John" },
"meta": { "page": 1, "total": 100 }
}
// Error
{
"error": {
"code": "VALIDATION_ERROR",
"message": "Request validation failed",
"details": [
{ "field": "email", "message": "Invalid email format" }
]
}
}
// List
{
"data": [...],
"meta": { "page": 1, "per_page": 20, "total_pages": 5 },
"links": {
"self": "/api/v1/users?page=1",
"next": "/api/v1/users?page=2",
"last": "/api/v1/users?page=5"
}
}
```
**Headers:**
- `Content-Type: application/json`
- `Accept: application/json`
- `Authorization: Bearer <token>`
- `X-Request-Id: <uuid>` (tracing)
- `RateLimit-Remaining: 49`
- `ETag: "<hash>"` (caching)''',
"source": "system_design",
"tags": ["rest", "api", "architecture"]
})
patterns.append({
"instruction": "How do you design database schemas for a multi-tenant application?",
"output": '''Multi-tenant database patterns:
**Pattern 1: Shared database, shared schema (row-level tenant)**
```sql
CREATE TABLE users (
id SERIAL PRIMARY KEY,
tenant_id UUID NOT NULL,
email VARCHAR(255) NOT NULL,
UNIQUE(tenant_id, email), -- Unique within tenant
INDEX idx_tenant_id (tenant_id)
);
-- Always filter by tenant_id
SELECT * FROM users WHERE tenant_id = 'abc-123';
```
Pros: Simple, cheap. Cons: Need to remember tenant_id everywhere, noisy neighbors.
**Pattern 2: Shared database, separate schemas**
```sql
-- Each tenant gets a schema
CREATE SCHEMA tenant_abc123;
CREATE SCHEMA tenant_def456;
-- Use search_path or explicit schema prefix
SET search_path TO tenant_abc123;
SELECT * FROM users; -- Uses tenant_abc123.users
```
Pros: Isolation, easy to backup per tenant. Cons: More connections, migration complexity.
**Pattern 3: Separate databases**
```python
# Router selects database per tenant
class TenantRouter:
def db_for_read(self, model, **hints):
tenant = hints.get('tenant')
return f'db_{tenant.slug}'
```
Pros: Maximum isolation, easy migration/deletion. Cons: Most expensive, connection overhead.
**Pattern 4: Row-Level Security (PostgreSQL)**
```sql
ALTER TABLE users ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON users
USING (tenant_id = current_setting('app.current_tenant')::uuid);
-- Now all queries automatically filtered
SET app.current_tenant = 'abc-123';
SELECT * FROM users; -- Only tenant abc-123 rows
```
Pros: Database enforces isolation, hard to bypass. Cons: PostgreSQL-specific.
**Recommendation:** Start with Pattern 1 + RLS (Pattern 4) for best balance of cost and safety.''',
"source": "system_design",
"tags": ["database", "multi_tenant", "architecture"]
})
patterns.append({
"instruction": "What are common CI/CD pipeline patterns for Python projects?",
"output": '''CI/CD pipeline for Python (GitHub Actions):
```yaml
name: CI/CD Pipeline
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
# 1. Lint and type check
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- run: pip install ruff mypy
- run: ruff check . # Linting + formatting
- run: ruff format --check . # Format check
- run: mypy src/ # Type checking
# 2. Test with coverage
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16
env:
POSTGRES_DB: testdb
POSTGRES_PASSWORD: test
ports: ['5432:5432']
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- run: pip install -r requirements-dev.txt
- run: |
pytest tests/ \
--cov=src/ \
--cov-report=xml \
--junitxml=test-results.xml
- uses: codecov/codecov-action@v3
# 3. Build and push Docker image
build:
needs: [lint, test]
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- uses: docker/build-push-action@v5
with:
push: true
tags: ghcr.io/${{ github.repository }}:${{ github.sha }}
cache-from: type=gha
cache-to: type=gha,mode=max
# 4. Deploy to staging
deploy-staging:
needs: build
runs-on: ubuntu-latest
environment: staging
steps:
- name: Deploy
run: |
kubectl set image deployment/app \
app=ghcr.io/${{ github.repository }}:${{ github.sha }} \
-n staging
```
Best practices:
- Run linters before tests (fail fast)
- Cache pip dependencies
- Use matrix builds for multiple Python versions
- Separate jobs for lint/test/build/deploy
- Use environments for approval gates''',
"source": "devops_pattern",
"tags": ["ci_cd", "github_actions", "devops"]
})
return patterns
def collect_github_readmes():
"""Collect README files from popular repos."""
print("[*] Collecting GitHub READMEs...")
data = []
headers = {"Accept": "application/vnd.github.v3+json"}
for repo in REPO_SOURCES:
url = f"https://raw.githubusercontent.com/{repo}/main/README.md"
try:
resp = requests.get(url, timeout=30)
if resp.status_code == 200:
content = resp.text[:10000]
entry = {
"instruction": f"Project: {repo}\n\nDescribe this project, its purpose, and how to get started.",
"output": content,
"source": "github_readme",
"tags": ["documentation", "project_overview"]
}
data.append(entry)
except Exception as e:
print(f" [!] Error on {repo}: {e}")
time.sleep(0.3) # Rate limit
print(f" [*] Collected {len(data)} README entries")
return data
def generate_code_review_examples():
"""Generate code review Q&A pairs."""
print("[*] Generating code review examples...")
reviews = []
reviews.append({
"instruction": "Review this Python code for issues:\n\n```python\nclass Database:\n def __init__(self, conn_string):\n self.conn = psycopg2.connect(conn_string)\n \n def query(self, sql, params=None):\n cursor = self.conn.cursor()\n cursor.execute(sql, params)\n return cursor.fetchall()\n \n def close(self):\n self.conn.close()\n```",
"output": '''**Issues Found: 5**
**1. No error handling on connection**
If `psycopg2.connect()` fails, the whole constructor crashes.
```python
# Fix
def __init__(self, conn_string):
try:
self.conn = psycopg2.connect(conn_string, timeout=5)
except psycopg2.OperationalError as e:
raise ConnectionError(f"Cannot connect to database: {e}") from e
```
**2. Cursor never closed**
Every query creates a cursor that leaks.
```python
# Fix - use context manager
from contextlib import contextmanager
@contextmanager
def cursor(self):
cur = self.conn.cursor()
try:
yield cur
self.conn.commit()
except Exception:
self.conn.rollback()
raise
finally:
cur.close()
def query(self, sql, params=None):
with self.cursor() as cur:
cur.execute(sql, params)
return cur.fetchall()
```
**3. No connection pooling**
Each instance opens a new connection. At scale, you'll hit connection limits.
```python
# Fix - use connection pooling
from psycopg2 import pool
class Database:
_pool = None
@classmethod
def init_pool(cls, conn_string, min_conn=1, max_conn=10):
cls._pool = pool.ThreadedConnectionPool(min_conn, max_conn, conn_string)
```
**4. SQL injection risk**
No parameter validation. While params are used, there's no guard against malformed queries.
**5. No health check**
No way to verify connection is alive before querying.
```python
def health_check(self):
try:
with self.cursor() as cur:
cur.execute("SELECT 1")
return True
except Exception:
return False
```''',
"source": "code_review",
"tags": ["python", "database", "review"]
})
reviews.append({
"instruction": "Review this JavaScript API endpoint:\n\n```javascript\napp.get('/api/users/:id', async (req, res) => {\n const user = await User.findById(req.params.id);\n res.json(user);\n});\n```",
"output": '''**Issues Found: 4**
**1. No error handling**
If `findById` throws (DB down, invalid ID), the whole server crashes.
```javascript
app.get('/api/users/:id', async (req, res, next) => {
try {
const user = await User.findById(req.params.id);
res.json(user);
} catch (error) {
next(error); // Pass to error handler
}
});
```
**2. No validation on ID**
`req.params.id` could be any string. MongoDB will throw on invalid ObjectId.
```javascript
const { ObjectId } = require('mongodb');
if (!ObjectId.isValid(req.params.id)) {
return res.status(400).json({ error: 'Invalid user ID' });
}
```
**3. Returns null on not found**
When user doesn't exist, returns `null` with 200. Should be 404.
```javascript
if (!user) {
return res.status(404).json({ error: 'User not found' });
}
res.json({ data: user });
```
**4. No authentication/authorization**
Anyone can fetch any user's data. Add auth middleware.
```javascript
app.get('/api/users/:id', authenticate, authorize('read:user'), async (req, res) => {
// ...
});
```''',
"source": "code_review",
"tags": ["javascript", "api", "review"]
})
print(f" [*] Generated {len(reviews)} code review entries")
return reviews
def generate_bug_fix_examples():
"""Generate common bug and fix examples."""
print("[*] Generating bug fix examples...")
bugs = []
bugs.append({
"instruction": "This Python code has a bug. Find and fix it:\n\n```python\ndef merge_dicts(a, b):\n return {**a, **b}\n\nx = {'a': 1, 'b': [1, 2]}\ny = {'b': [3, 4], 'c': 5}\nresult = merge_dicts(x, y)\n# Expected: {'a': 1, 'b': [1, 2, 3, 4], 'c': 5}\n# Actual: {'a': 1, 'b': [3, 4], 'c': 5}\n```",
"output": '''**Bug:** Dictionary unpacking `**` replaces values entirely. List `[3, 4]` replaces `[1, 2]` instead of merging.
**Fix:**
```python
from copy import deepcopy
def merge_dicts(a, b, deep=True):
"""Deep merge two dictionaries, concatenating lists."""
result = deepcopy(a)
for key, value in b.items():
if key in result:
if isinstance(result[key], dict) and isinstance(value, dict):
result[key] = merge_dicts(result[key], value)
elif isinstance(result[key], list) and isinstance(value, list):
result[key] = result[key] + value
else:
result[key] = deepcopy(value)
else:
result[key] = deepcopy(value)
return result
```
**Lesson:** `{**a, **b}` is a shallow merge. For nested data, you need explicit merge logic per type.''',
"source": "bug_fix",
"tags": ["python", "dictionary", "mutation"]
})
bugs.append({
"instruction": "This Python code has a subtle bug with mutable default arguments. Find and fix it:\n\n```python\ndef add_item(item, items=[]):\n items.append(item)\n return items\n\na = add_item(1)\nb = add_item(2)\n# Expected: a = [1], b = [2]\n# Actual: a = [1, 2], b = [1, 2]\n```",
"output": '''**Bug:** The default `[]` is evaluated ONCE at function definition time. All calls without the second argument share the SAME list object.
**Fix:**
```python
def add_item(item, items=None):
if items is None:
items = [] # New list each call
items.append(item)
return items
```
**Lesson:** Never use mutable defaults (`[]`, `{}`, `set()`) in Python function signatures. Use `None` and create the mutable inside the function body.
**Other common mutable default traps:**
```python
# BAD
def func(data={}):
data['key'] = 'value'
# GOOD
def func(data=None):
if data is None:
data = {}
data['key'] = 'value'
```''',
"source": "bug_fix",
"tags": ["python", "mutable_default", "classic_bug"]
})
bugs.append({
"instruction": "This Go code has a goroutine bug. Find and fix it:\n\n```go\nfunc main() {\n names := []string{\"alice\", \"bob\", \"charlie\"}\n var wg sync.WaitGroup\n \n for _, name := range names {\n wg.Add(1)\n go func() {\n fmt.Println(\"Processing:\", name)\n wg.Done()\n }()\n }\n wg.Wait()\n}\n// Output is unpredictable - may print the same name 3 times\n```",
"output": '''**Bug:** The loop variable `name` is shared across all iterations. By the time goroutines start running, the loop may have finished and `name` points to the last value.
**Fix 1 - Pass as parameter:**
```go
for _, name := range names {
wg.Add(1)
go func(n string) { // Parameter captures current value
fmt.Println("Processing:", n)
wg.Done()
}(name) // Pass current iteration value
}
```
**Fix 2 - Shadow variable (Go 1.22+ handles this correctly):**
```go
for _, name := range names {
name := name // Shadow - new variable per iteration (pre-1.22)
wg.Add(1)
go func() {
fmt.Println("Processing:", name)
wg.Done()
}()
}
```
**Lesson:** In Go < 1.22, range variables are reused across iterations. Always pass loop variables as goroutine parameters or shadow them.''',
"source": "bug_fix",
"tags": ["go", "goroutine", "closure"]
})
print(f" [*] Generated {len(bugs)} bug fix entries")
return bugs
def collect_api_docs():
"""Collect API documentation."""
print("[*] Collecting API documentation...")
data = []
apis = [
("requests", "https://requests.readthedocs.io/en/latest/user/quickstart/"),
("flask", "https://flask.palletsprojects.com/en/latest/quickstart/"),
("sqlalchemy", "https://docs.sqlalchemy.org/en/20/quickstart.html"),
("fastapi", "https://fastapi.tiangolo.com/tutorial/first-steps/"),
]
for name, url in apis:
try:
resp = requests.get(url, timeout=30, headers={"User-Agent": "Mozilla/5.0"})
if resp.status_code == 200:
soup = BeautifulSoup(resp.text, "html.parser")
content = soup.get_text(separator="\n", strip=True)[:8000]
entry = {
"instruction": f"What are the key features and usage patterns of {name}?",
"output": content,
"source": "api_docs",
"tags": ["documentation", name]
}
data.append(entry)
except Exception as e:
print(f" [!] Error fetching {name}: {e}")
print(f" [*] Collected {len(data)} API doc entries")
return data
def generate_testing_patterns():
"""Generate testing patterns and examples."""
patterns = []
patterns.append({
"instruction": "How do you write integration tests that test real database interactions?",
"output": '''Integration testing with real database:
```python
# conftest.py
import pytest
import sqlalchemy as sa
from sqlalchemy.orm import Session
from myapp.database import Base, get_session
@pytest.fixture(scope="session")
def engine():
"""Create test database engine."""
return sa.create_engine("postgresql://localhost/test_db", echo=True)
@pytest.fixture(scope="session")
def tables(engine):
"""Create all tables, drop after tests."""
Base.metadata.create_all(engine)
yield
Base.metadata.drop_all(engine)
@pytest.fixture
def db_session(engine, tables) -> Session:
"""Provide a transactional scope around a test."""
connection = engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
session.close()
transaction.rollback() # Rollback - no test data persists
connection.close()
# test_users.py
def test_create_user(db_session: Session):
user = User(email="test@example.com", name="Test User")
db_session.add(user)
db_session.commit()
# Query back from DB
found = db_session.query(User).filter_by(email="test@example.com").one()
assert found.name == "Test User"
assert found.id is not None
def test_duplicate_email_rejected(db_session: Session):
db_session.add(User(email="dup@example.com"))
db_session.commit()
with pytest.raises(sa.exc.IntegrityError):
db_session.add(User(email="dup@example.com"))
db_session.commit()
```
Key principles:
- Use transaction rollback so test data doesn't persist
- `scope="session"` for expensive fixtures (engine, tables)
- Each test gets a fresh transaction
- Test against real database, not mocks''',
"source": "testing_pattern",
"tags": ["python", "integration_test", "database"]
})
patterns.append({
"instruction": "What are the key Docker patterns for production applications?",
"output": '''Production Docker patterns:
```dockerfile
# Multi-stage build (Python example)
# Stage 1: Build dependencies
FROM python:3.12-slim AS builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
# Stage 2: Runtime (minimal image)
FROM python:3.12-slim
WORKDIR /app
# Copy only installed packages
COPY --from=builder /install /usr/local
# Non-root user for security
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
# Copy application
COPY --chown=appuser:appuser . .
# Health check
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD python -c "import requests; requests.get('http://localhost:8000/health')"
# Run with gunicorn (not uvicorn directly)
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "app:app"]
```
**Docker Compose for development:**
```yaml
services:
app:
build: .
ports: ["8000:8000"]
environment:
- DATABASE_URL=postgresql://postgres:password@db:5432/app
depends_on:
db:
condition: service_healthy
volumes:
- ./src:/app/src # Hot reload in dev
db:
image: postgres:16
environment:
POSTGRES_DB: app
POSTGRES_PASSWORD: password
ports: ["5432:5432"]
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 5s
volumes:
pgdata:
```
Best practices:
- Multi-stage builds for small images
- Run as non-root user
- Use `.dockerignore` to exclude unnecessary files
- Health checks for orchestration
- Separate dev/prod configs''',
"source": "devops_pattern",
"tags": ["docker", "devops", "production"]
})
return patterns
def compile_dataset():
"""Compile all sources into a single training dataset."""
print("=" * 60)
print("Dev-AI Dataset Collection")
print("=" * 60)
all_data = []
# Generated patterns (reliable, no network needed)
all_data.extend(generate_python_patterns())
all_data.extend(generate_javascript_patterns())
all_data.extend(generate_rust_patterns())
all_data.extend(generate_golang_patterns())
all_data.extend(generate_system_design_patterns())
all_data.extend(generate_code_review_examples())
all_data.extend(generate_bug_fix_examples())
all_data.extend(generate_testing_patterns())
# Network sources (may fail, that's OK)
all_data.extend(collect_github_readmes())
all_data.extend(collect_api_docs())
# Format for training (instruction/output pairs)
training_data = []
for item in all_data:
training_data.append({
"messages": [
{"role": "user", "content": item["instruction"]},
{"role": "assistant", "content": item["output"]}
],
"source": item["source"],
"tags": item.get("tags", [])
})
# Save
os.makedirs(PROCESSED_DIR, exist_ok=True)
output_file = PROCESSED_DIR / "dev_finetuning.json"
with open(output_file, "w") as f:
json.dump(training_data, f, indent=2)
print(f"\n{'=' * 60}")
print(f"Total entries: {len(training_data)}")
print(f"Sources: {', '.join(sorted(set(d['source'] for d in all_data)))}")
print(f"Tags: {', '.join(sorted(set(t for d in all_data for t in d.get('tags', []))))}")
print(f"Saved to: {output_file}")
file_size = output_file.stat().st_size / 1024
print(f"File size: {file_size:.0f} KB")
print(f"{'=' * 60}")
return training_data
if __name__ == "__main__":
compile_dataset()