Key Code Examples
1. Design Pattern — Singleton (Thread-Safe)
import threading
from functools import wraps
def singleton(cls):
"""Thread-safe singleton decorator."""
instances = {}
lock = threading.Lock()
@wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
with lock:
# Double-checked locking
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class DatabaseConnection:
"""Singleton database connection manager."""
def __init__(self, connection_string="postgresql://localhost/app"):
self.connection_string = connection_string
self._connection = None
self._lock = threading.Lock()
print(f"Creating DB connection to {connection_string}")
def execute(self, query):
with self._lock:
print(f"Executing: {query}")
return [{"id": 1, "name": "result"}]
def close(self):
if self._connection:
self._connection.close()
self._connection = None
# Usage
db1 = DatabaseConnection() # Creates instance
db2 = DatabaseConnection() # Returns same instance
print(db1 is db2) # True
2. Design Pattern — Observer Pattern
from typing import List, Callable
from abc import ABC, abstractmethod
class Observer(ABC):
@abstractmethod
def update(self, event: str, data: dict):
pass
class Subject:
"""Observable subject that notifies observers of changes."""
def __init__(self):
self._observers: List[Observer] = []
self._state = {}
def attach(self, observer: Observer):
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer: Observer):
self._observers.remove(observer)
def notify(self, event: str, data: dict = None):
for observer in self._observers:
observer.update(event, data or {})
def set_state(self, key: str, value):
self._state[key] = value
self.notify("state_changed", {"key": key, "value": value})
# Concrete implementations
class EmailNotifier(Observer):
def update(self, event: str, data: dict):
print(f"[EMAIL] Event: {event}, Data: {data}")
class SMSNotifier(Observer):
def update(self, event: str, data: dict):
print(f"[SMS] Event: {event}, Data: {data}")
class Logger(Observer):
def update(self, event: str, data: dict):
print(f"[LOG] {event}: {data}")
# Usage
order_service = Subject()
order_service.attach(EmailNotifier())
order_service.attach(Logger())
order_service.set_state("order_status", "shipped")
# [EMAIL] Event: state_changed, Data: {'key': 'order_status', 'value': 'shipped'}
# [LOG] state_changed: {'key': 'order_status', 'value': 'shipped'}
3. RESTful Service with Clean Architecture
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional
import uuid
from datetime import datetime
# ===== Domain Layer =====
@dataclass
class User:
id: str
email: str
name: str
created_at: datetime
class UserRepository(ABC):
"""Abstract repository — domain layer defines interface."""
@abstractmethod
def find_by_id(self, user_id: str) -> Optional[User]:
pass
@abstractmethod
def find_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def delete(self, user_id: str) -> bool:
pass
@abstractmethod
def list_all(self, page: int = 1, limit: int = 20) -> List[User]:
pass
# ===== Application Layer (Use Cases) =====
class UserService:
"""Business logic lives here — independent of frameworks."""
def __init__(self, repository: UserRepository):
self.repository = repository
def register_user(self, email: str, name: str) -> User:
# Business validation
if not email or '@' not in email:
raise ValueError("Invalid email address")
# Check for existing user
existing = self.repository.find_by_email(email)
if existing:
raise ValueError(f"User with email {email} already exists")
# Create user
user = User(
id=str(uuid.uuid4()),
email=email,
name=name,
created_at=datetime.utcnow()
)
return self.repository.save(user)
def get_user_profile(self, user_id: str) -> dict:
user = self.repository.find_by_id(user_id)
if not user:
raise NotFoundError(f"User {user_id} not found")
return {
"id": user.id,
"email": user.email,
"name": user.name,
"member_since": user.created_at.strftime("%Y-%m-%d")
}
def deactivate_user(self, user_id: str) -> bool:
# Business rule: can't deactivate admin users
user = self.repository.find_by_id(user_id)
if not user:
raise NotFoundError(f"User {user_id} not found")
return self.repository.delete(user_id)
# ===== Infrastructure Layer =====
class InMemoryUserRepository(UserRepository):
"""In-memory implementation for testing."""
def __init__(self):
self._users = {}
def find_by_id(self, user_id: str) -> Optional[User]:
return self._users.get(user_id)
def find_by_email(self, email: str) -> Optional[User]:
for user in self._users.values():
if user.email == email:
return user
return None
def save(self, user: User) -> User:
self._users[user.id] = user
return user
def delete(self, user_id: str) -> bool:
if user_id in self._users:
del self._users[user_id]
return True
return False
def list_all(self, page=1, limit=20):
start = (page - 1) * limit
end = start + limit
return list(self._users.values())[start:end]
class NotFoundError(Exception):
pass
# ===== Dependency Injection =====
# Composition root — wire everything together
repository = InMemoryUserRepository()
user_service = UserService(repository)
4. Unit Testing with pytest
import pytest
from unittest.mock import Mock, patch
from datetime import datetime
class TestUserService:
"""Unit tests for UserService with mocked dependencies."""
def setup_method(self):
"""Set up test fixtures."""
self.mock_repo = Mock()
self.service = UserService(self.mock_repo)
def test_register_user_success(self):
"""Test successful user registration."""
# Arrange
self.mock_repo.find_by_email.return_value = None
self.mock_repo.save.side_effect = lambda u: u
# Act
user = self.service.register_user("alice@example.com", "Alice")
# Assert
assert user.email == "alice@example.com"
assert user.name == "Alice"
assert user.id is not None
self.mock_repo.save.assert_called_once()
def test_register_duplicate_email(self):
"""Test registration fails for duplicate email."""
# Arrange
existing = User(id="123", email="alice@example.com",
name="Existing", created_at=datetime.now())
self.mock_repo.find_by_email.return_value = existing
# Act & Assert
with pytest.raises(ValueError, match="already exists"):
self.service.register_user("alice@example.com", "Alice")
self.mock_repo.save.assert_not_called()
def test_register_invalid_email(self):
"""Test registration fails for invalid email."""
with pytest.raises(ValueError, match="Invalid email"):
self.service.register_user("not-an-email", "Alice")
def test_get_user_profile_success(self):
"""Test retrieving user profile."""
user = User(id="42", email="bob@example.com", name="Bob",
created_at=datetime(2026, 1, 15, 10, 30))
self.mock_repo.find_by_id.return_value = user
profile = self.service.get_user_profile("42")
assert profile["id"] == "42"
assert profile["email"] == "bob@example.com"
assert profile["name"] == "Bob"
assert profile["member_since"] == "2026-01-15"
def test_get_user_profile_not_found(self):
"""Test profile retrieval for non-existent user."""
self.mock_repo.find_by_id.return_value = None
with pytest.raises(NotFoundError):
self.service.get_user_profile("nonexistent")
@patch('builtins.print')
def test_service_prints_confirmation(self, mock_print):
"""Test that service logs confirmation (with mocked print)."""
# ... test logic with mocked side effects
pass
# Run: pytest test_user_service.py -v --tb=short
5. CI/CD Pipeline Configuration (GitHub Actions)
# .github/workflows/ci-cd.yml
name: CI/CD Pipeline
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
env:
NODE_VERSION: '20'
PYTHON_VERSION: '3.11'
jobs:
# ===== Stage 1: Code Quality =====
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-node@v4
with:
node-version: ${{ env.NODE_VERSION }}
cache: 'npm'
- run: npm ci
- run: npm run lint
- run: npm run type-check
# ===== Stage 2: Testing =====
test:
runs-on: ubuntu-latest
needs: lint
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12']
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: test_password
ports: ['5432:5432']
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-cov
- name: Run tests with coverage
env:
DATABASE_URL: postgresql://postgres:test_password@localhost:5432/test_db
run: |
pytest --cov=src --cov-report=xml --cov-report=html
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
- name: Enforce coverage threshold
run: |
coverage report --fail-under=80
# ===== Stage 3: Security Scan =====
security:
runs-on: ubuntu-latest
needs: test
steps:
- uses: actions/checkout@v4
- name: Run Trivy vulnerability scanner
uses: aquasecurity/trivy-action@master
with:
scan-type: 'fs'
scan-ref: '.'
format: 'sarif'
output: 'trivy-results.sarif'
- name: Run Bandit (Python security)
run: |
pip install bandit
bandit -r src/ -f json -o bandit-report.json || true
- name: Upload security results
uses: github/codeql-action/upload-sarif@v2
with:
sarif_file: 'trivy-results.sarif'
# ===== Stage 4: Build & Deploy =====
deploy:
runs-on: ubuntu-latest
needs: [test, security]
if: github.ref == 'refs/heads/main'
environment: production
steps:
- uses: actions/checkout@v4
- name: Configure AWS credentials
uses: aws-actions/configure-aws-credentials@v4
with:
aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
aws-region: us-east-1
- name: Login to Amazon ECR
id: login-ecr
uses: aws-actions/amazon-ecr-login@v2
- name: Build and push Docker image
env:
ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
IMAGE_TAG: ${{ github.sha }}
run: |
docker build -t $ECR_REGISTRY/eduverse-api:$IMAGE_TAG .
docker push $ECR_REGISTRY/eduverse-api:$IMAGE_TAG
- name: Deploy to ECS
run: |
aws ecs update-service \
--cluster eduverse-production \
--service api-service \
--force-new-deployment
- name: Smoke tests
run: |
sleep 30
curl -f https://api.eduverse.com/health || exit 1
6. Git Workflow & Branching Strategy
# GitFlow branching strategy implementation
# ===== Initialize GitFlow =====
git flow init -d # Uses default branch names
# ===== Feature Development =====
git flow feature start user-authentication
# Creates feature/user-authentication from develop
# Work on feature...
git add .
git commit -m "feat(auth): add JWT token generation"
git commit -m "feat(auth): implement refresh token rotation"
git commit -m "test(auth): add unit tests for token service"
git flow feature finish user-authentication
# Merges into develop, deletes feature branch
# ===== Release Process =====
git flow release start v2.1.0
# Creates release/v2.1.0 from develop
# Version bump, final fixes
echo "2.1.0" > VERSION
git add VERSION
git commit -m "chore(release): bump version to 2.1.0"
git flow release finish v2.1.0
# Merges into main and develop, creates tag v2.1.0
# Push everything
git push origin main develop --tags
# ===== Hotfix Process =====
git flow hotfix start v2.1.1
# Creates hotfix/v2.1.1 from main
# Fix the critical bug
git commit -m "fix(api): resolve null pointer in user endpoint"
git flow hotfix finish v2.1.1
# Merges into main AND develop, creates tag
# ===== Useful Aliases =====
git config --global alias.lg "log --graph --oneline --decorate --all"
git config --global alias.st "status -sb"
git config --global alias.co "checkout"
git config --global alias.br "branch -vv"
git config --global alias.ci "commit"
git config --global alias.unstage "reset HEAD --"
git config --global alias.last "log -1 HEAD"
7. Code Review Checklist Generator
from dataclasses import dataclass, field
from typing import List, Dict
from enum import Enum
import re
class Severity(Enum):
CRITICAL = "CRITICAL"
WARNING = "WARNING"
SUGGESTION = "SUGGESTION"
@dataclass
class ReviewFinding:
category: str
severity: Severity
message: str
line: int = 0
suggestion: str = ""
class CodeReviewer:
"""Automated code review assistant."""
CHECKS = {
"security": [
(r'print\(.*password', Severity.CRITICAL,
"Possible password in log output", "Use logging with redaction"),
(r'except\s*:', Severity.WARNING,
"Bare except clause catches all exceptions",
"Use 'except Exception:' or specific exceptions"),
(r'eval\(', Severity.CRITICAL,
"Use of eval() is dangerous",
"Use ast.literal_eval() or json.loads()"),
(r'subprocess\..*shell\s*=\s*True', Severity.WARNING,
"Shell=True can lead to command injection",
"Use shell=False with list arguments"),
],
"quality": [
(r'def\s+\w+\([^)]*\)\s*->\s*None.*\n(?:.*\n){50,}', Severity.SUGGESTION,
"Very long function (>50 lines)", "Consider breaking into smaller functions"),
(r'# TODO|# FIXME|# HACK', Severity.WARNING,
"Unresolved TODO/FIXME comment", "Address or create a ticket"),
(r'print\(', Severity.SUGGESTION,
"Using print() for logging", "Use the logging module instead"),
],
"performance": [
(r'for\s+\w+\s+in\s+.*:\s*\n\s+for\s+\w+\s+in', Severity.WARNING,
"Nested loops may have O(n²) complexity", "Consider optimization"),
(r'\.append\(.*\).*\n.*\.append\(', Severity.SUGGESTION,
"Repeated append in loop", "Consider list comprehension"),
]
}
def review_file(self, filepath: str, content: str) -> List[ReviewFinding]:
"""Review a single file for issues."""
findings = []
lines = content.split('\n')
for category, rules in self.CHECKS.items():
for pattern, severity, message, suggestion in rules:
for match in re.finditer(pattern, content, re.MULTILINE):
line_num = content[:match.start()].count('\n') + 1
findings.append(ReviewFinding(
category=category,
severity=severity,
message=message,
line=line_num,
suggestion=suggestion
))
return findings
def generate_report(self, findings: List[ReviewFinding]) -> str:
"""Generate a human-readable review report."""
report = ["# Code Review Report\n"]
for severity in [Severity.CRITICAL, Severity.WARNING, Severity.SUGGESTION]:
items = [f for f in findings if f.severity == severity]
if items:
report.append(f"## {severity.value} ({len(items)} issues)\n")
for item in items:
report.append(f" Line {item.line}: {item.message}")
report.append(f" Category: {item.category}")
if item.suggestion:
report.append(f" Suggestion: {item.suggestion}")
report.append("")
return "\n".join(report)
# Usage
reviewer = CodeReviewer()
with open('app.py', 'r') as f:
findings = reviewer.review_file('app.py', f.read())
print(reviewer.generate_report(findings))
8. Database Migration with Alembic
# alembic migration example
# alembic/versions/001_create_users_table.py
"""Create users table
Revision ID: 001
Revises:
Create Date: 2026-01-15 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers
revision = '001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
op.create_table(
'users',
sa.Column('id', sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column('email', sa.String(255), unique=True, nullable=False),
sa.Column('password_hash', sa.String(255), nullable=False),
sa.Column('first_name', sa.String(100), nullable=False),
sa.Column('last_name', sa.String(100), nullable=False),
sa.Column('is_active', sa.Boolean(), server_default='true'),
sa.Column('role', sa.String(20), server_default='customer'),
sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(),
onupdate=sa.func.now()),
)
op.create_index('idx_users_email', 'users', ['email'])
op.create_index('idx_users_role', 'users', ['role'])
def downgrade():
op.drop_index('idx_users_role', table_name='users')
op.drop_index('idx_users_email', table_name='users')
op.drop_table('users')
# Zero-downtime migration strategy for large tables:
# 1. Add new column (nullable)
# 2. Backfill in batches
# 3. Deploy code that writes to both old and new columns
# 4. Deploy code that reads from new column
# 5. Drop old column
#
# commands:
# alembic revision --autogenerate -m "Add user preferences"
# alembic upgrade head
# alembic downgrade -1
# alembic current
# alembic history
9. API Versioning & Documentation
from functools import wraps
from flask import Flask, request, jsonify
app = Flask(__name__)
class APIVersionManager:
"""Manage multiple API versions with deprecation support."""
SUPPORTED_VERSIONS = ['v1', 'v2']
DEFAULT_VERSION = 'v2'
DEPRECATED_VERSIONS = ['v1']
DEPRECATION_DATE = '2026-06-01'
SUNSET_DATE = '2026-12-01'
def versioned_route(self, rule, version=None, **options):
"""Decorator for versioned API routes."""
def decorator(f):
endpoint = options.pop('endpoint', None)
# Register for default version
if version is None or version == self.DEFAULT_VERSION:
app.add_url_rule(rule, endpoint, f, **options)
# Register version-specific
versioned_rule = f"/api/{version or self.DEFAULT_VERSION}{rule}"
app.add_url_rule(versioned_rule, f"{endpoint}_{version}",
f, **options)
# Add deprecation headers
@wraps(f)
def wrapped(*args, **kwargs):
response = f(*args, **kwargs)
if isinstance(response, tuple):
resp, status = response
if version in self.DEPRECATED_VERSIONS:
resp = jsonify(resp)
resp.headers['Deprecation'] = 'true'
resp.headers['Sunset'] = self.SUNSET_DATE
resp.headers['Link'] = f'; rel="successor-version"'
return resp, status
return response
return wrapped
return decorator
# Usage
api = APIVersionManager()
@app.route('/api/v1/users', methods=['GET'])
def get_users_v1():
"""Original v1 endpoint."""
return jsonify([{"id": 1, "name": "Alice"}])
@app.route('/api/v2/users', methods=['GET'])
def get_users_v2():
"""Enhanced v2 endpoint."""
return jsonify({
"data": [{"id": 1, "name": "Alice", "email": "alice@example.com"}],
"meta": {"page": 1, "total": 100, "version": "v2"}
})
# OpenAPI/Swagger documentation (using flasgger)
from flasgger import Swagger
swagger = Swagger(app, template={
"info": {
"title": "EduVerse API",
"version": "2.0.0",
"description": "API for EduVerse Platform"
},
"basePath": "/api/v2",
"schemes": ["https"]
})
@app.route('/api/v2/products', methods=['GET'])
def list_products():
"""
List all products
---
tags:
- Products
parameters:
- name: page
in: query
type: integer
default: 1
- name: category
in: query
type: string
responses:
200:
description: List of products
schema:
type: object
properties:
data:
type: array
items:
type: object
"""
return jsonify({"data": [], "meta": {}})
10. Performance Profiling & Monitoring
import time
import tracemalloc
from functools import wraps
from collections import defaultdict
import statistics
class PerformanceMonitor:
"""Application performance monitoring."""
def __init__(self):
self.metrics = defaultdict(list)
self.errors = defaultdict(int)
def track_time(self, operation_name):
"""Decorator to track function execution time."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
self.metrics[operation_name].append(elapsed)
return result
except Exception as e:
self.errors[f"{operation_name}_error"] += 1
raise
return wrapper
return decorator
def track_memory(self, func):
"""Decorator to track memory usage."""
@wraps(func)
def wrapper(*args, **kwargs):
tracemalloc.start()
result = func(*args, **kwargs)
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
self.metrics[f"{func.__name__}_memory_peak"].append(peak)
return result
return wrapper
def get_stats(self, operation_name: str = None) -> dict:
"""Get performance statistics."""
if operation_name:
times = self.metrics[operation_name]
if not times:
return {}
return {
"operation": operation_name,
"count": len(times),
"total_time": sum(times),
"mean": statistics.mean(times),
"median": statistics.median(times),
"p95": sorted(times)[int(len(times) * 0.95)],
"p99": sorted(times)[int(len(times) * 0.99)],
"min": min(times),
"max": max(times),
}
return {name: self.get_stats(name) for name in self.metrics}
def report(self) -> str:
"""Generate a performance report."""
lines = ["=== Performance Report ===\n"]
for name in sorted(self.metrics.keys()):
stats = self.get_stats(name)
lines.append(f"{name}:")
lines.append(f" Calls: {stats['count']}")
lines.append(f" Mean: {stats['mean']*1000:.2f}ms")
lines.append(f" P95: {stats['p95']*1000:.2f}ms")
lines.append(f" P99: {stats['p99']*1000:.2f}ms")
lines.append(f" Max: {stats['max']*1000:.2f}ms")
lines.append("")
if self.errors:
lines.append("Errors:")
for error, count in self.errors.items():
lines.append(f" {error}: {count}")
return "\n".join(lines)
# Usage
monitor = PerformanceMonitor()
@monitor.track_time("db_query")
def get_users_from_db():
time.sleep(0.05) # Simulate DB query
return [{"id": 1}]
@monitor.track_time("api_call")
def call_external_api():
time.sleep(0.1) # Simulate API call
return {"status": "ok"}
# Run and report
for _ in range(100):
get_users_from_db()
call_external_api()
print(monitor.report())
Core Concepts
Software Development Life Cycle (SDLC)
Definition: A structured process for planning, creating, testing, and deploying software.
- Requirements: Gather and document what the software needs to do
- Design: Create architecture, database schema, UI/UX designs
- Implementation: Write code following design specifications
- Testing: Verify correctness (unit, integration, system, acceptance)
- Deployment: Release to production environment
- Maintenance: Fix bugs, add features, optimize performance
Agile & Scrum Methodology
Agile: Iterative development with frequent delivery and feedback.
Scrum Framework:
- Sprint: 1-4 week time-boxed iteration
- Product Backlog: Prioritized list of features
- Sprint Backlog: Items selected for current sprint
- Daily Standup: 15-min daily sync meeting
- Sprint Review: Demo completed work to stakeholders
- Retrospective: Reflect on what went well/improve
- Key Roles: Product Owner, Scrum Master, Development Team
SOLID Principles
Single Responsibility: A class should have only one reason to change (one job).
Open/Closed: Open for extension, closed for modification (use interfaces).
Liskov Substitution: Subtypes must be substitutable for their base types.
Interface Segregation: Many small interfaces beat one large interface.
Dependency Inversion: Depend on abstractions, not concrete implementations.
Design Patterns (GoF)
Creational: How objects are created
- Singleton: One instance of a class (database connection)
- Factory: Create objects without specifying exact class
- Builder: Construct complex objects step by step
Structural: How classes/objects are composed
- Adapter: Make incompatible interfaces work together
- Decorator: Add behavior dynamically
- Facade: Simplified interface to complex subsystem
Behavioral: How objects interact
- Observer: Notify dependents of state changes
- Strategy: Interchangeable algorithms
- Command: Encapsulate request as object
Testing Pyramid
Unit Tests (70%): Test individual functions/components in isolation. Fast, reliable.
Integration Tests (20%): Test how modules work together. DB, API, services.
E2E Tests (10%): Test complete user workflows from UI to database. Slow but comprehensive.
Key Practices: TDD (Test-Driven Development), mocking, CI/CD integration, coverage targets
Version Control Best Practices
- Commit Often: Small, logical commits with descriptive messages
- Branching: GitFlow, GitHub Flow, or trunk-based development
- Code Review: All code reviewed by at least one peer
- Never Commit Secrets: Use .gitignore, environment variables
- Conventional Commits: feat:, fix:, chore:, docs:, test:
Microservices Architecture
Definition: Decomposing an application into small, independently deployable services.
Benefits: Independent scaling, technology diversity, fault isolation, faster deployment.
Challenges: Distributed system complexity, data consistency, service discovery, monitoring.
Key Patterns: API Gateway, Circuit Breaker, Event Sourcing, CQRS, Saga Pattern.
Clean Code Principles
- Meaningful Names: Variables, functions, classes should clearly describe their purpose
- Small Functions: Under 20 lines, single responsibility
- No Comments Needed: Code should be self-documenting
- DRY (Don't Repeat Yourself): Extract shared logic
- Error Handling: Use exceptions, not error codes
- Testing: Every function should have corresponding tests
Worked Examples
Example 1: Refactoring Legacy Code
Problem: Refactor this spaghetti code into clean, maintainable code.
Before:
def process_order(data):
# 200 lines of mixed concerns...
# validate, calculate, save, send email, update inventory
# all in one giant function
if data['type'] == 'standard':
price = data['price'] * 1.1 # tax
elif data['type'] == 'express':
price = data['price'] * 1.2
elif data['type'] == 'international':
price = data['price'] * 1.3
# ... 190 more lines
After (Clean Code):
class OrderProcessor:
def __init__(self, tax_calculator, notifier, inventory):
self.tax_calculator = tax_calculator
self.notifier = notifier
self.inventory = inventory
def process(self, order_data: dict) -> Order:
order = self._create_order(order_data)
order = self._apply_pricing(order)
self._save_order(order)
self._notify_customer(order)
self._update_inventory(order)
return order
def _create_order(self, data: dict) -> Order:
return Order(
id=str(uuid.uuid4()),
items=data['items'],
customer_email=data['email'],
shipping_address=data['address'],
created_at=datetime.utcnow()
)
def _apply_pricing(self, order: Order) -> Order:
order.subtotal = sum(item.price * item.qty for item in order.items)
order.tax = self.tax_calculator.calculate(order)
order.total = order.subtotal + order.tax
return order
def _save_order(self, order: Order):
self.repository.save(order)
def _notify_customer(self, order: Order):
self.notifier.send_order_confirmation(order)
def _update_inventory(self, order: Order):
for item in order.items:
self.inventory.decrease_stock(item.product_id, item.qty)
Example 2: Implement TDD Workflow
Problem: Implement a shopping cart using Test-Driven Development.
Step 1: Write failing test first
def test_add_item_to_cart():
cart = ShoppingCart()
cart.add_item("PROD-001", quantity=2, price=29.99)
assert cart.item_count == 1
assert cart.total == 59.98
Step 2: Write minimal implementation
class ShoppingCart:
def __init__(self):
self._items = []
self.item_count = 0
self.total = 0.0
def add_item(self, product_id, quantity, price):
self._items.append({"id": product_id, "qty": quantity,
"price": price})
self.item_count += 1
self.total += quantity * price
Step 3: Refactor and add more tests
def test_apply_discount():
cart = ShoppingCart()
cart.add_item("PROD-001", quantity=1, price=100.0)
cart.apply_discount(code="SAVE10", percent=10)
assert cart.total == 90.0
def test_remove_item():
cart = ShoppingCart()
cart.add_item("PROD-001", quantity=2, price=10.0)
cart.remove_item("PROD-001")
assert cart.item_count == 0
assert cart.total == 0.0
# Continue cycle: Red → Green → Refactor
Example 3: Database Migration Zero-Downtime
Problem: Add a NOT NULL column `phone_number` to a 10M-row users table without downtime.
Step-by-step migration:
-- Migration 1: Add nullable column (fast, no table rebuild)
ALTER TABLE users ADD COLUMN phone_number VARCHAR(20) NULL;
-- Completes in < 1 second (metadata only in MySQL 8+)
-- Migration 2: Backfill in batches (run over hours/days)
UPDATE users SET phone_number = '+10000000000'
WHERE id BETWEEN 1 AND 100000 AND phone_number IS NULL;
-- Repeat for each batch of 100K rows...
-- Migration 3: Deploy code that writes phone_number
-- Application now includes phone_number in INSERT/UPDATE
-- Migration 4: Verify backfill complete
SELECT COUNT(*) FROM users WHERE phone_number IS NULL; -- Should be 0
-- Migration 5: Add NOT NULL constraint
ALTER TABLE users MODIFY COLUMN phone_number VARCHAR(20) NOT NULL;
-- Fast because no NULL values exist
Example 4: Circuit Breaker Pattern
Problem: Implement a circuit breaker to prevent cascading failures when calling an unreliable external service.
import time
import threading
from enum import Enum
class CircuitState(Enum):
CLOSED = "CLOSED" # Normal operation
OPEN = "OPEN" # Failing, reject requests
HALF_OPEN = "HALF_OPEN" # Testing if service recovered
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=30,
half_open_max_calls=3):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = None
self.half_open_calls = 0
self.lock = threading.Lock()
def call(self, func, *args, **kwargs):
with self.lock:
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
else:
raise CircuitBreakerOpenError(
"Service unavailable — circuit breaker is OPEN"
)
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.half_open_max_calls:
raise CircuitBreakerOpenError(
"Half-open call limit reached"
)
self.half_open_calls += 1
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.failure_count = 0
def _on_failure(self):
with self.lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
def _should_attempt_reset(self):
return (time.time() - self.last_failure_time) >= self.recovery_timeout
class CircuitBreakerOpenError(Exception):
pass
# Usage
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
try:
result = breaker.call(unreliable_api_call, "param")
except CircuitBreakerOpenError:
# Return cached data or default response
result = get_cached_data()
Example 5: Implement Repository Pattern
Problem: Abstract data access to make the application testable and database-agnostic.
from abc import ABC, abstractmethod
from typing import List, Optional
# Domain model
@dataclass
class Product:
id: int
name: str
price: float
stock: int
# Abstract repository
class ProductRepository(ABC):
@abstractmethod
def get_by_id(self, product_id: int) -> Optional[Product]: pass
@abstractmethod
def search(self, query: str) -> List[Product]: pass
@abstractmethod
def save(self, product: Product) -> Product: pass
@abstractmethod
def update_stock(self, product_id: int, quantity: int) -> bool: pass
# SQL implementation
class SQLProductRepository(ProductRepository):
def __init__(self, db_connection):
self.db = db_connection
def get_by_id(self, product_id: int) -> Optional[Product]:
row = self.db.execute(
"SELECT id, name, price, stock FROM products WHERE id = %s",
(product_id,)
).fetchone()
return Product(**row) if row else None
def search(self, query: str) -> List[Product]:
rows = self.db.execute(
"SELECT id, name, price, stock FROM products WHERE name ILIKE %s",
(f'%{query}%',)
).fetchall()
return [Product(**row) for row in rows]
# Cache decorator
class CachedProductRepository(ProductRepository):
def __init__(self, repository: ProductRepository, ttl=300):
self.repository = repository
self.cache = {}
self.ttl = ttl
def get_by_id(self, product_id: int) -> Optional[Product]:
key = f"product:{product_id}"
if key in self.cache:
cached, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return cached
product = self.repository.get_by_id(product_id)
if product:
self.cache[key] = (product, time.time())
return product
Practice Problems
Problem 1: Implement Strategy Pattern
Create a payment processing system using the Strategy pattern for different payment methods.
from abc import ABC, abstractmethod
class PaymentStrategy(ABC):
@abstractmethod
def pay(self, amount: float) -> dict: pass
class CreditCardPayment(PaymentStrategy):
def __init__(self, card_number, cvv, expiry):
self.card_number = card_number[-4:] # Only store last 4 digits
def pay(self, amount: float) -> dict:
# Process credit card payment
return {"status": "success", "method": "credit_card",
"amount": amount, "last4": self.card_number}
class PayPalPayment(PaymentStrategy):
def __init__(self, email):
self.email = email
def pay(self, amount: float) -> dict:
return {"status": "success", "method": "paypal",
"amount": amount, "email": self.email}
class CryptoPayment(PaymentStrategy):
def __init__(self, wallet_address):
self.wallet_address = wallet_address
def pay(self, amount: float) -> dict:
return {"status": "success", "method": "crypto",
"amount": amount, "wallet": self.wallet_address[:8] + "..."}
class PaymentProcessor:
def __init__(self, strategy: PaymentStrategy):
self.strategy = strategy
def set_strategy(self, strategy: PaymentStrategy):
self.strategy = strategy
def process_payment(self, amount: float) -> dict:
return self.strategy.pay(amount)
# Usage
processor = PaymentProcessor(CreditCardPayment("4111111111111111", "123", "12/27"))
result = processor.process_payment(99.99)
Problem 2: Write Integration Tests
Write integration tests for a user registration API endpoint.
import pytest
import requests
BASE_URL = "http://localhost:8000/api"
class TestUserRegistration:
def test_register_success(self):
response = requests.post(f"{BASE_URL}/register", json={
"email": "test@example.com",
"password": "SecureP@ss123",
"name": "Test User"
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert "password" not in data # Never return password!
def test_register_duplicate_email(self):
# First registration
requests.post(f"{BASE_URL}/register", json={
"email": "dup@example.com",
"password": "SecureP@ss123",
"name": "User"
})
# Duplicate should fail
response = requests.post(f"{BASE_URL}/register", json={
"email": "dup@example.com",
"password": "SecureP@ss123",
"name": "User"
})
assert response.status_code == 409
def test_register_weak_password(self):
response = requests.post(f"{BASE_URL}/register", json={
"email": "weak@example.com",
"password": "123",
"name": "Weak User"
})
assert response.status_code == 400
assert "password" in response.json()["errors"]
Problem 3: Implement Decorator Pattern
Create decorators for logging, caching, and retry logic.
import time
import functools
import logging
def log_execution(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
logging.info(f"Calling {func.__name__}")
result = func(*args, **kwargs)
elapsed = time.time() - start
logging.info(f"{func.__name__} completed in {elapsed:.3f}s")
return result
return wrapper
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts:
raise
logging.warning(f"Attempt {attempt} failed: {e}. Retrying...")
time.sleep(delay * attempt) # Exponential backoff
return None
return wrapper
return decorator
def cache_result(ttl=300):
def decorator(func):
cache = {}
@functools.wraps(func)
def wrapper(*args):
if args in cache:
result, timestamp = cache[args]
if time.time() - timestamp < ttl:
return result
result = func(*args)
cache[args] = (result, time.time())
return result
return wrapper
return decorator
# Usage
@log_execution
@retry(max_attempts=3, delay=2)
def unreliable_api_call():
import random
if random.random() < 0.7:
raise ConnectionError("API unavailable")
return {"data": "success"}
@cache_result(ttl=60)
@log_execution
def expensive_computation(n):
time.sleep(2)
return n ** 2
Problem 4: Git Rebase vs Merge
Explain when to use rebase vs merge and demonstrate both workflows.
Answer:
- Merge: Preserves complete history, creates merge commit. Best for public branches.
- Rebase: Linear history, no merge commits. Best for feature branches before PR.
# Merge workflow
git checkout main
git pull origin main
git merge feature/user-auth
git push origin main
# Rebase workflow
git checkout feature/user-auth
git fetch origin
git rebase origin/main
# Resolve conflicts if any
git push origin feature/user-auth --force-with-lease
Problem 5: Implement a Thread-Safe Queue
Create a producer-consumer queue with proper synchronization.
import threading
import queue
import time
class Pipeline:
def __init__(self, max_size=10):
self.task_queue = queue.Queue(maxsize=max_size)
self.result_queue = queue.Queue()
self._shutdown = threading.Event()
def producer(self, tasks):
for task in tasks:
if self._shutdown.is_set():
break
self.task_queue.put(task)
print(f"Produced: {task}")
self.task_queue.put(None) # Sentinel
def consumer(self, worker_id):
while not self._shutdown.is_set():
try:
task = self.task_queue.get(timeout=1)
if task is None:
self.task_queue.put(None) # Pass sentinel
break
result = f"Worker {worker_id} processed: {task}"
self.result_queue.put(result)
self.task_queue.task_done()
except queue.Empty:
continue
def run(self, tasks, num_workers=3):
producer_thread = threading.Thread(
target=self.producer, args=(tasks,))
consumer_threads = [
threading.Thread(target=self.consumer, args=(i,))
for i in range(num_workers)
]
producer_thread.start()
for t in consumer_threads:
t.start()
producer_thread.join()
for t in consumer_threads:
t.join()
results = []
while not self.result_queue.empty():
results.append(self.result_queue.get())
return results
Problem 6: Calculate Cyclomatic Complexity
Write a function that calculates cyclomatic complexity of Python code.
import ast
def calculate_complexity(source_code: str) -> dict:
"""Calculate cyclomatic complexity per function."""
tree = ast.parse(source_code)
results = {}
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
complexity = 1 # Base complexity
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.For, ast.While,
ast.ExceptHandler, ast.With,
ast.Assert)):
complexity += 1
elif isinstance(child, ast.BoolOp):
complexity += len(child.values) - 1
results[node.name] = {
'complexity': complexity,
'lines': node.end_lineno - node.lineno
}
return results
# Usage
code = '''
def process(items):
for item in items:
if item > 0:
if item % 2 == 0:
print("even")
else:
print("odd")
try:
result = 10 / len(items)
except ZeroDivisionError:
result = 0
return result
'''
print(calculate_complexity(code))
# {'process': {'complexity': 5, 'lines': 11}}
Problem 7: Design System for Scale
Design a URL shortener service like bit.ly. Include API, database, and caching.
# URL Shortener Design
## API Endpoints
# POST /api/shorten → { "url": "https://..." } → { "short_code": "abc123" }
# GET /{short_code} → 302 redirect to original URL
# GET /api/stats/{short_code} → click count, referrers
## Database Schema
CREATE TABLE urls (
id BIGSERIAL PRIMARY KEY,
short_code VARCHAR(10) UNIQUE NOT NULL,
original_url TEXT NOT NULL,
created_at TIMESTAMP DEFAULT NOW(),
expires_at TIMESTAMP,
user_id INT REFERENCES users(id),
click_count BIGINT DEFAULT 0
);
CREATE INDEX idx_short_code ON urls(short_code);
## Architecture
# 1. User → POST /api/shorten with URL
# 2. Generate short code (base62 encoding of auto-increment ID)
# 3. Store in PostgreSQL + cache in Redis (TTL: 24h)
# 4. Return short URL: https://short.link/abc123
#
# Redirect Flow:
# 1. User visits https://short.link/abc123
# 2. Check Redis cache → if found, redirect
# 3. If cache miss → query PostgreSQL
# 4. Update click count asynchronously (via message queue)
# 5. 302 redirect to original URL
## Scaling Considerations
# - Use consistent hashing for cache sharding
# - Pre-generate short codes in batches
# - Read replicas for analytics queries
# - CDN for static assets
# - Rate limiting per user/IP
Problem 8: Implement Feature Flags
Create a feature flag system for gradual rollouts.
import hashlib
from enum import Enum
class RolloutStrategy(Enum):
BOOLEAN = "boolean"
PERCENTAGE = "percentage"
USER_LIST = "user_list"
class FeatureFlag:
def __init__(self, name, strategy, value, description=""):
self.name = name
self.strategy = strategy
self.value = value
self.description = description
self.enabled = True
def is_enabled_for(self, user_id=None):
if not self.enabled:
return False
if self.strategy == RolloutStrategy.BOOLEAN:
return self.value
elif self.strategy == RolloutStrategy.PERCENTAGE:
# Consistent hashing for stable assignment
hash_val = int(hashlib.md5(
f"{self.name}:{user_id}".encode()
).hexdigest(), 16)
return (hash_val % 100) < self.value
elif self.strategy == RolloutStrategy.USER_LIST:
return user_id in self.value
return False
class FeatureFlagManager:
def __init__(self):
self._flags = {}
def register(self, flag: FeatureFlag):
self._flags[flag.name] = flag
def is_enabled(self, name, user_id=None):
flag = self._flags.get(name)
return flag.is_enabled_for(user_id) if flag else False
# Usage
manager = FeatureFlagManager()
manager.register(FeatureFlag(
"new-dashboard", RolloutStrategy.PERCENTAGE, 25,
"25% of users see new dashboard"
))
manager.register(FeatureFlag(
"beta-feature", RolloutStrategy.USER_LIST,
["user-123", "user-456"]
))
if manager.is_enabled("new-dashboard", user_id="user-789"):
render_new_dashboard()
else:
render_legacy_dashboard()
Problem 9: Write a Makefile
Create a Makefile for a Python project with common development tasks.
PYTHON := python3
VENV := venv
PIP := $(VENV)/bin/pip
PYTEST := $(VENV)/bin/pytest
.PHONY: help install test lint format clean run migrate
help:
@echo "Available commands:"
@echo " install Install dependencies"
@echo " test Run tests with coverage"
@echo " lint Run linters"
@echo " format Format code"
@echo " clean Remove build artifacts"
@echo " run Start development server"
@echo " migrate Run database migrations"
install:
$(PYTHON) -m venv $(VENV)
$(PIP) install -U pip
$(PIP) install -r requirements.txt
test:
$(PYTEST) --cov=src --cov-report=html --cov-report=term
lint:
$(VENV)/bin/flake8 src/
$(VENV)/bin/mypy src/
format:
$(VENV)/bin/black src/ tests/
$(VENV)/bin/isort src/ tests/
clean:
rm -rf __pycache__ .pytest_cache htmlcov dist build *.egg-info
find . -name "*.pyc" -delete
run:
$(VENV)/bin/python -m uvicorn src.main:app --reload --port 8000
migrate:
$(VENV)/bin/alembic upgrade head
Problem 10: Implement Event Sourcing
Design an event-sourced system for a banking account.
from dataclasses import dataclass
from typing import List
from datetime import datetime
import uuid
@dataclass
class DomainEvent:
event_id: str
aggregate_id: str
event_type: str
timestamp: datetime
data: dict
class AccountAggregate:
def __init__(self, account_id: str):
self.account_id = account_id
self.balance = 0
self.version = 0
self._uncommitted_events = []
@classmethod
def load_from_history(cls, events: List[DomainEvent]):
"""Rebuild state from event history."""
if not events:
raise ValueError("Cannot load from empty history")
account = cls(events[0].aggregate_id)
for event in events:
account.apply(event)
account.version += 1
return account
def apply(self, event: DomainEvent):
"""Apply event to mutate state."""
if event.event_type == "AccountOpened":
self.balance = event.data["initial_balance"]
elif event.event_type == "MoneyDeposited":
self.balance += event.data["amount"]
elif event.event_type == "MoneyWithdrawn":
self.balance -= event.data["amount"]
def deposit(self, amount: float):
if amount <= 0:
raise ValueError("Amount must be positive")
self._uncommitted_events.append(DomainEvent(
event_id=str(uuid.uuid4()),
aggregate_id=self.account_id,
event_type="MoneyDeposited",
timestamp=datetime.utcnow(),
data={"amount": amount}
))
def withdraw(self, amount: float):
if amount > self.balance:
raise ValueError("Insufficient funds")
self._uncommitted_events.append(DomainEvent(
event_id=str(uuid.uuid4()),
aggregate_id=self.account_id,
event_type="MoneyWithdrawn",
timestamp=datetime.utcnow(),
data={"amount": amount}
))
# Event store
class EventStore:
def __init__(self):
self._events = {} # aggregate_id -> [events]
def save(self, aggregate_id: str, events: List[DomainEvent]):
if aggregate_id not in self._events:
self._events[aggregate_id] = []
self._events[aggregate_id].extend(events)
def get_events(self, aggregate_id: str) -> List[DomainEvent]:
return self._events.get(aggregate_id, [])