Python's dynamic nature and import system create unique proximity opportunities, especially with type hints and docstrings.
Traditional (Anti-Pattern):
project/
├── models/
│ └── user.py
├── views/
│ └── user_view.py
├── controllers/
│ └── user_controller.py
├── validators/
│ └── user_validator.py
└── tests/
└── test_user.py
Proximity Pattern:
project/
├── user/
│ ├── __init__.py # Public API exports
│ ├── model.py # User model and business logic
│ ├── model_test.py # Model tests
│ ├── api.py # API endpoints
│ ├── api_test.py # API tests
│ ├── validation.py # User-specific validation
│ └── fixtures.py # Test fixtures
# 🧠 DECISION: 10-minute cache TTL for user sessions
# Why: Balance between memory usage and database load
# Measured: 10min reduces DB queries by 85% with 50MB memory overhead
# Alternative: 60min TTL - rejected due to 300MB memory requirement
USER_SESSION_TTL_SECONDS = 600
# 🛡️ SECURITY: Bcrypt cost factor of 12
# Threat: Password brute force attacks
# Benchmark: 12 rounds = 250ms per hash on our servers
# Standard: OWASP recommends 10-12 for 2024
BCRYPT_COST_FACTOR = 12
# ⚡ PERFORMANCE: Batch size of 1000 for bulk operations
# Benchmarked: 100=slow, 1000=optimal, 10000=memory issues
# Memory: 1000 records ≈ 10MB in memory
BATCH_SIZE = 1000from typing import Optional, Union, TypedDict
# 🧠 DECISION: Using TypedDict for API responses
# Why: Provides IDE support while maintaining dict compatibility
# Alternative: Dataclass - would break existing JSON serialization
class UserResponse(TypedDict):
id: int
email: str
created_at: str # ISO 8601 string, not datetime
def get_user(user_id: int) -> Optional[UserResponse]:
"""
🧠 DECISION: Return None instead of raising for missing users
Why: Allows callers to handle missing users gracefully
Pattern: Follows Python's 'easier to ask forgiveness' principle
"""
user = db.query(User).filter_by(id=user_id).first()
return user.to_dict() if user else None# Strike 1: In user_service.py
def normalize_email(email: str) -> str:
return email.lower().strip()
# Strike 2: In admin_service.py
def normalize_email(email: str) -> str:
return email.lower().strip()
# Strike 3: Extract to utils/email.py
# 🔧 EXTRACTION: Email normalization
# Used in: user_service.py, admin_service.py, api_service.py
def normalize_email(email: str) -> str:
"""Normalize email to lowercase and strip whitespace."""
return email.lower().strip()# user/model.py
class User:
def __init__(self, email: str):
self.email = email
def is_active(self) -> bool:
# 🧠 DECISION: Users active by default
# Why: Reduces onboarding friction
# Risk: Spam accounts remain active
# Mitigation: Separate spam detection system
return not self.deleted_at
# user/model_test.py
import pytest
from .model import User
class TestUser:
def test_is_active_for_new_user(self):
"""New users should be active by default."""
user = User(email="test@example.com")
assert user.is_active()
def test_is_active_for_deleted_user(self):
"""Deleted users should not be active."""
user = User(email="test@example.com")
user.deleted_at = datetime.now()
assert not user.is_active()# user/fixtures.py
import pytest
from .model import User
@pytest.fixture
def active_user():
"""
🧠 DECISION: Use factory fixtures over static data
Why: Prevents test interdependence from shared mutable state
"""
return User(email="active@example.com")
@pytest.fixture
def deleted_user():
user = User(email="deleted@example.com")
user.deleted_at = datetime.now()
return user# In payment/errors.py - where payment errors are understood
class PaymentError(Exception):
"""Base payment error with rich context."""
def __init__(self, message: str, **context):
# 🧠 DECISION: Include context dict in all errors
# Why: Critical for debugging production issues
# Example: PaymentError("Failed", amount=100, currency="USD")
super().__init__(message)
self.context = context
class InsufficientFundsError(PaymentError):
"""
🛡️ SECURITY: Don't expose actual balance in error
Why: Prevents information leakage about account balances
"""
def __init__(self, requested_amount: float):
super().__init__(
"Insufficient funds for transaction",
requested_amount=requested_amount,
# Note: Deliberately not including actual_balance
)
class PaymentGatewayError(PaymentError):
"""
⚡ PERFORMANCE: Include retry information
Why: Helps circuit breaker decide on retries
"""
def __init__(self, gateway: str, status_code: int, can_retry: bool):
super().__init__(
f"Gateway {gateway} returned {status_code}",
gateway=gateway,
status_code=status_code,
can_retry=can_retry,
retry_after=self._calculate_retry_delay(status_code)
)class DataProcessor:
# 🧠 DECISION: Group methods by workflow, not alphabetically
# Why: Related operations should be visually proximate
# === Data Loading Methods ===
def load_from_file(self, path: str) -> pd.DataFrame:
"""Load data from file."""
pass
def load_from_database(self, query: str) -> pd.DataFrame:
"""Load data from database."""
pass
# === Data Validation Methods ===
def validate_schema(self, df: pd.DataFrame) -> bool:
"""Validate dataframe schema."""
pass
def validate_values(self, df: pd.DataFrame) -> List[str]:
"""Validate data values, return errors."""
pass
# === Data Transformation Methods ===
def normalize(self, df: pd.DataFrame) -> pd.DataFrame:
"""Normalize numeric columns."""
pass
def encode_categoricals(self, df: pd.DataFrame) -> pd.DataFrame:
"""Encode categorical columns."""
pass# Don't put all decorators in decorators.py
# Put them near where they're used
# In api/auth.py
def require_auth(permission: str = None):
"""
🛡️ SECURITY: Authentication decorator
Why: Centralized auth check prevents forgetting auth
Pattern: Fail-secure - default to denying access
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if not current_user.is_authenticated:
raise Unauthorized("Authentication required")
if permission and not current_user.has_permission(permission):
raise Forbidden(f"Permission '{permission}' required")
return func(*args, **kwargs)
return wrapper
return decorator
# Use immediately below definition
@require_auth(permission="admin")
def delete_user(user_id: int):
"""Delete a user (admin only)."""
pass# ⚡ PERFORMANCE: LRU cache for expensive computations
# Benchmark: Reduces response time from 500ms to 5ms for repeat calls
# Memory: Cache size of 128 = ~10MB memory overhead
# TTL: No TTL, relies on LRU eviction
from functools import lru_cache
@lru_cache(maxsize=128)
def calculate_user_score(user_id: int) -> float:
"""
Calculate user score (expensive operation).
🧠 DECISION: Cache user scores instead of real-time calculation
Why: Score changes infrequently (daily recalc)
Impact: 100x speedup for API responses
"""
# Expensive calculation here
pass
# ⚡ PERFORMANCE: Generator for memory efficiency
# Memory: Processes 1GB file with 10MB memory usage
def process_large_file(filepath: str):
"""
🧠 DECISION: Use generator to avoid loading entire file
Why: Files can be multiple GB
Alternative: pandas.read_csv() - would OOM on large files
"""
with open(filepath) as f:
for line in f:
yield process_line(line)# 🛡️ SECURITY: SQL injection prevention
# Threat: User input in SQL queries
# Mitigation: Parameterized queries only
def get_user_by_email(email: str) -> Optional[User]:
# NEVER: f"SELECT * FROM users WHERE email = '{email}'"
# ALWAYS: Parameterized query
query = "SELECT * FROM users WHERE email = %s"
return db.execute(query, (email,)).fetchone()
# 🛡️ SECURITY: Path traversal prevention
# Threat: User-supplied paths accessing system files
# Mitigation: Resolve and check path is within allowed directory
def read_user_file(username: str, filename: str) -> str:
base_dir = Path("/data/users")
user_path = (base_dir / username / filename).resolve()
# Critical: Ensure resolved path is within base_dir
if not str(user_path).startswith(str(base_dir)):
raise SecurityError(f"Path traversal attempted: {filename}")
return user_path.read_text()# requirements.txt with decision context
# 🧠 DECISION: FastAPI for web framework
# Why: Async support + automatic OpenAPI docs
# Alternative: Flask - no built-in async support
fastapi==0.104.1
# 🛡️ SECURITY: Latest cryptography for security fixes
# Why: CVE-2023-49083 fixed in 41.0.7
# Update policy: Always use latest for security libs
cryptography==41.0.7
# ⚡ PERFORMANCE: uvloop for async performance
# Benchmark: 2x faster than default asyncio loop
# Compatibility: Linux/Mac only, falls back on Windows
uvloop==0.19.0; sys_platform != 'win32'
# 📊 DATA: Pandas for data manipulation
# Why: Industry standard, team expertise
# Version pin: 2.x has breaking changes, staying on 1.x
pandas>=1.5,<2.0# In email/sender.py - not in global config.py
# 🧠 DECISION: Email configuration as class attributes
# Why: Keeps email config near email logic
# Alternative: Global settings.py - too distant from usage
class EmailSender:
# Email service configuration
SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com")
SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
# 🧠 DECISION: 30-second timeout for SMTP operations
# Why: Prevents hanging on network issues
# Observed: 99% of sends complete in <5 seconds
SMTP_TIMEOUT = 30
# 🛡️ SECURITY: Always use TLS for email
# Why: Prevents credential interception
USE_TLS = True
# ⚡ PERFORMANCE: Batch size for bulk emails
# Benchmark: 50 emails/batch optimal for our SMTP server
# Higher: Rate limiting kicks in
# Lower: Too many connections
BATCH_SIZE = 50# 🧠 DECISION: Async by default for I/O operations
# Why: 10x concurrency improvement for API endpoints
# Measurement: Sync = 100 req/s, Async = 1000 req/s
async def fetch_user_data(user_id: int) -> dict:
"""
⚡ PERFORMANCE: Concurrent fetching of user data
Why: Parallelize independent I/O operations
Impact: 3x speedup (300ms -> 100ms)
"""
# These can run concurrently
user_task = fetch_user(user_id)
posts_task = fetch_posts(user_id)
stats_task = fetch_stats(user_id)
# Wait for all to complete
user, posts, stats = await asyncio.gather(
user_task, posts_task, stats_task
)
return {
"user": user,
"posts": posts,
"stats": stats
}- Are magic numbers replaced with documented constants?
- Do type hints include rationale for complex types?
- Are test files colocated with implementation?
- Do exceptions provide rich context?
- Are decorators defined near their usage?
- Is async/await usage documented with performance impact?
- Are security validations marked and explained?
- Do requirements.txt entries explain choices?
- Are configuration values near their usage?
- Is the 3-strikes rule followed for utilities?
See python-examples/ for complete examples including:
- FastAPI application with proximity patterns
- Data pipeline with decision archaeology
- Async service with performance documentation