python
exercises
exercises.py🐍python
"""
Real World Projects - Exercises
Build complete, production-ready applications.
"""
import pytest
from typing import Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from abc import ABC, abstractmethod
from enum import Enum, auto
# =============================================================================
# EXERCISE 1: User Authentication System
# =============================================================================
class User:
"""
Create a user authentication system that:
- Registers users with email and password
- Hashes passwords securely
- Validates login credentials
- Tracks login attempts and locks accounts
- Manages user sessions
Example:
auth = AuthSystem()
auth.register("user@example.com", "SecurePass123!")
session = auth.login("user@example.com", "SecurePass123!")
assert session is not None
user = auth.get_user_from_session(session.token)
assert user.email == "user@example.com"
"""
def __init__(self, email: str, password_hash: str):
self.email = email
self.password_hash = password_hash
self.created_at = datetime.now()
self.login_attempts = 0
self.locked_until: Optional[datetime] = None
self.is_active = True
@dataclass
class Session:
"""User session."""
token: str
user_email: str
created_at: datetime = field(default_factory=datetime.now)
expires_at: datetime = field(default_factory=lambda: datetime.now() + timedelta(hours=24))
class AuthSystem:
"""User authentication system."""
MAX_LOGIN_ATTEMPTS = 5
LOCK_DURATION_MINUTES = 15
def __init__(self):
self.users: dict[str, User] = {}
self.sessions: dict[str, Session] = {}
def _hash_password(self, password: str) -> str:
"""Hash password securely."""
# TODO: Implement secure password hashing
# Use hashlib with salt
pass
def _verify_password(self, password: str, password_hash: str) -> bool:
"""Verify password against hash."""
# TODO: Verify password
pass
def _generate_token(self) -> str:
"""Generate a session token."""
# TODO: Generate secure random token
pass
def register(self, email: str, password: str) -> Optional[User]:
"""Register a new user."""
# TODO: Implement user registration
# - Validate email format
# - Check password strength
# - Hash password
# - Store user
pass
def login(self, email: str, password: str) -> Optional[Session]:
"""Login a user."""
# TODO: Implement login
# - Check if user exists
# - Check if account is locked
# - Verify password
# - Track login attempts
# - Create session on success
pass
def logout(self, token: str) -> bool:
"""Logout a user by invalidating session."""
# TODO: Remove session
pass
def get_user_from_session(self, token: str) -> Optional[User]:
"""Get user from session token."""
# TODO: Validate session and return user
pass
def change_password(self, email: str, old_password: str,
new_password: str) -> bool:
"""Change user password."""
# TODO: Verify old password and update to new
pass
class TestAuthSystem:
"""Tests for AuthSystem."""
def test_register_user(self):
"""Test user registration."""
auth = AuthSystem()
user = auth.register("test@example.com", "SecurePass123!")
assert user is not None
assert user.email == "test@example.com"
assert user.password_hash != "SecurePass123!" # Hashed
def test_login_success(self):
"""Test successful login."""
auth = AuthSystem()
auth.register("test@example.com", "SecurePass123!")
session = auth.login("test@example.com", "SecurePass123!")
assert session is not None
assert session.user_email == "test@example.com"
def test_login_wrong_password(self):
"""Test login with wrong password."""
auth = AuthSystem()
auth.register("test@example.com", "SecurePass123!")
session = auth.login("test@example.com", "WrongPassword!")
assert session is None
def test_account_lockout(self):
"""Test account lockout after failed attempts."""
auth = AuthSystem()
auth.register("test@example.com", "SecurePass123!")
# Fail multiple times
for _ in range(auth.MAX_LOGIN_ATTEMPTS):
auth.login("test@example.com", "WrongPassword!")
# Should be locked even with correct password
session = auth.login("test@example.com", "SecurePass123!")
assert session is None
def test_session_validation(self):
"""Test session token validation."""
auth = AuthSystem()
auth.register("test@example.com", "SecurePass123!")
session = auth.login("test@example.com", "SecurePass123!")
user = auth.get_user_from_session(session.token)
assert user is not None
assert user.email == "test@example.com"
# =============================================================================
# EXERCISE 2: Rate Limiter
# =============================================================================
class RateLimiter:
"""
Create a rate limiter that:
- Limits requests per time window
- Supports different limits per key/user
- Uses sliding window algorithm
- Provides remaining requests info
Example:
limiter = RateLimiter(requests=100, window_seconds=60)
for i in range(100):
assert limiter.allow("user_123") == True
assert limiter.allow("user_123") == False # Rate limited
print(limiter.get_remaining("user_123")) # 0
"""
def __init__(self, requests: int = 100, window_seconds: int = 60):
self.max_requests = requests
self.window_seconds = window_seconds
self.requests: dict[str, list[datetime]] = {}
def _cleanup_old_requests(self, key: str) -> None:
"""Remove requests outside the window."""
# TODO: Remove old timestamps
pass
def allow(self, key: str) -> bool:
"""Check if request is allowed."""
# TODO: Implement sliding window rate limiting
pass
def get_remaining(self, key: str) -> int:
"""Get remaining requests in window."""
# TODO: Return remaining quota
pass
def reset(self, key: str) -> None:
"""Reset rate limit for a key."""
# TODO: Clear request history
pass
def get_reset_time(self, key: str) -> Optional[datetime]:
"""Get when rate limit resets."""
# TODO: Return time of oldest request expiry
pass
class TestRateLimiter:
"""Tests for RateLimiter."""
def test_allows_within_limit(self):
"""Test requests within limit are allowed."""
limiter = RateLimiter(requests=10, window_seconds=60)
for _ in range(10):
assert limiter.allow("user1") == True
def test_blocks_over_limit(self):
"""Test requests over limit are blocked."""
limiter = RateLimiter(requests=5, window_seconds=60)
for _ in range(5):
limiter.allow("user1")
assert limiter.allow("user1") == False
def test_remaining_requests(self):
"""Test remaining requests count."""
limiter = RateLimiter(requests=10, window_seconds=60)
assert limiter.get_remaining("user1") == 10
for _ in range(3):
limiter.allow("user1")
assert limiter.get_remaining("user1") == 7
def test_independent_keys(self):
"""Test different keys have independent limits."""
limiter = RateLimiter(requests=5, window_seconds=60)
for _ in range(5):
limiter.allow("user1")
# user2 should still be allowed
assert limiter.allow("user2") == True
# =============================================================================
# EXERCISE 3: Job Queue
# =============================================================================
class JobStatus(Enum):
"""Job status enumeration."""
PENDING = auto()
RUNNING = auto()
COMPLETED = auto()
FAILED = auto()
CANCELLED = auto()
@dataclass
class Job:
"""Represents a job in the queue."""
id: str
name: str
payload: dict
status: JobStatus = JobStatus.PENDING
priority: int = 0
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
result: Optional[Any] = None
error: Optional[str] = None
retries: int = 0
max_retries: int = 3
class JobQueue:
"""
Create a job queue that:
- Adds jobs with priority
- Processes jobs in priority order
- Supports job retries on failure
- Tracks job status and history
- Provides job statistics
Example:
queue = JobQueue()
job1 = queue.add("send_email", {"to": "user@example.com"}, priority=5)
job2 = queue.add("process_data", {"file": "data.csv"}, priority=10)
# Higher priority first
next_job = queue.get_next()
assert next_job.name == "process_data"
"""
def __init__(self):
self.jobs: dict[str, Job] = {}
self.completed: list[Job] = []
def add(self, name: str, payload: dict, priority: int = 0) -> Job:
"""Add a job to the queue."""
# TODO: Create and store job
pass
def get_next(self) -> Optional[Job]:
"""Get the next job to process."""
# TODO: Return highest priority pending job
pass
def start_job(self, job_id: str) -> bool:
"""Mark a job as running."""
# TODO: Update job status and started_at
pass
def complete_job(self, job_id: str, result: Any = None) -> bool:
"""Mark a job as completed."""
# TODO: Update status, result, completed_at
pass
def fail_job(self, job_id: str, error: str) -> bool:
"""Mark a job as failed."""
# TODO: Update status, handle retries
pass
def cancel_job(self, job_id: str) -> bool:
"""Cancel a pending job."""
# TODO: Cancel if still pending
pass
def get_stats(self) -> dict:
"""Get queue statistics."""
# TODO: Return counts by status, avg processing time, etc.
pass
class TestJobQueue:
"""Tests for JobQueue."""
def test_add_job(self):
"""Test adding a job."""
queue = JobQueue()
job = queue.add("test_job", {"key": "value"})
assert job.status == JobStatus.PENDING
assert job.name == "test_job"
def test_priority_ordering(self):
"""Test jobs are returned by priority."""
queue = JobQueue()
queue.add("low", {}, priority=1)
queue.add("high", {}, priority=10)
queue.add("medium", {}, priority=5)
next_job = queue.get_next()
assert next_job.name == "high"
def test_job_lifecycle(self):
"""Test complete job lifecycle."""
queue = JobQueue()
job = queue.add("lifecycle_test", {})
queue.start_job(job.id)
assert queue.jobs[job.id].status == JobStatus.RUNNING
queue.complete_job(job.id, result="success")
assert queue.jobs[job.id].status == JobStatus.COMPLETED
def test_retry_on_failure(self):
"""Test job retry mechanism."""
queue = JobQueue()
job = queue.add("retry_test", {})
job.max_retries = 2
queue.start_job(job.id)
queue.fail_job(job.id, "error")
# Should be retried (back to pending)
assert queue.jobs[job.id].status == JobStatus.PENDING
assert queue.jobs[job.id].retries == 1
# =============================================================================
# EXERCISE 4: Notification System
# =============================================================================
class NotificationChannel(ABC):
"""Abstract notification channel."""
@abstractmethod
def send(self, recipient: str, message: str) -> bool:
pass
class EmailChannel(NotificationChannel):
"""Email notification channel (mock)."""
def __init__(self):
self.sent: list[dict] = []
def send(self, recipient: str, message: str) -> bool:
self.sent.append({"to": recipient, "message": message})
return True
class SMSChannel(NotificationChannel):
"""SMS notification channel (mock)."""
def __init__(self):
self.sent: list[dict] = []
def send(self, recipient: str, message: str) -> bool:
self.sent.append({"to": recipient, "message": message})
return True
class NotificationSystem:
"""
Create a notification system that:
- Supports multiple channels (email, SMS, push)
- Allows user preferences for channels
- Templates messages with variables
- Tracks delivery status
- Queues notifications for batch sending
Example:
system = NotificationSystem()
system.register_channel("email", EmailChannel())
system.register_channel("sms", SMSChannel())
system.set_preferences("user@example.com", ["email", "sms"])
system.send_notification(
"user@example.com",
"Welcome, {name}!",
{"name": "John"}
)
"""
def __init__(self):
self.channels: dict[str, NotificationChannel] = {}
self.preferences: dict[str, list[str]] = {}
self.templates: dict[str, str] = {}
self.history: list[dict] = []
def register_channel(self, name: str, channel: NotificationChannel) -> None:
"""Register a notification channel."""
# TODO: Store channel
pass
def set_preferences(self, user: str, channels: list[str]) -> None:
"""Set user notification preferences."""
# TODO: Store user channel preferences
pass
def register_template(self, name: str, template: str) -> None:
"""Register a message template."""
# TODO: Store template
pass
def send_notification(self, user: str, message: str,
variables: dict = None) -> list[bool]:
"""Send notification through user's preferred channels."""
# TODO: Implement notification sending
# - Get user preferences
# - Substitute variables in message
# - Send through each channel
# - Track delivery status
pass
def send_template(self, user: str, template_name: str,
variables: dict = None) -> list[bool]:
"""Send notification using a template."""
# TODO: Load template and send
pass
def get_history(self, user: str = None) -> list[dict]:
"""Get notification history."""
# TODO: Return history, optionally filtered by user
pass
class TestNotificationSystem:
"""Tests for NotificationSystem."""
def test_register_channel(self):
"""Test registering a channel."""
system = NotificationSystem()
system.register_channel("email", EmailChannel())
assert "email" in system.channels
def test_send_to_preferred_channels(self):
"""Test sending to user's preferred channels."""
system = NotificationSystem()
email = EmailChannel()
sms = SMSChannel()
system.register_channel("email", email)
system.register_channel("sms", sms)
system.set_preferences("user1", ["email", "sms"])
system.send_notification("user1", "Hello!")
assert len(email.sent) == 1
assert len(sms.sent) == 1
def test_variable_substitution(self):
"""Test message variable substitution."""
system = NotificationSystem()
email = EmailChannel()
system.register_channel("email", email)
system.set_preferences("user1", ["email"])
system.send_notification(
"user1",
"Hello, {name}! Your order #{order_id} is ready.",
{"name": "John", "order_id": "12345"}
)
assert "John" in email.sent[0]["message"]
assert "12345" in email.sent[0]["message"]
# =============================================================================
# EXERCISE 5: Inventory Management
# =============================================================================
@dataclass
class Product:
"""Represents a product."""
sku: str
name: str
price: float
quantity: int = 0
class InventorySystem:
"""
Create an inventory management system that:
- Tracks product quantities
- Handles stock adjustments (add, remove, transfer)
- Tracks inventory movements
- Generates low stock alerts
- Provides inventory valuation
Example:
inventory = InventorySystem()
inventory.add_product(Product("SKU001", "Widget", 9.99, 100))
inventory.adjust_stock("SKU001", -10, reason="sale")
assert inventory.get_quantity("SKU001") == 90
alerts = inventory.get_low_stock_alerts(threshold=20)
"""
def __init__(self):
self.products: dict[str, Product] = {}
self.movements: list[dict] = []
def add_product(self, product: Product) -> None:
"""Add a product to inventory."""
# TODO: Store product
pass
def get_product(self, sku: str) -> Optional[Product]:
"""Get product by SKU."""
# TODO: Return product
pass
def get_quantity(self, sku: str) -> int:
"""Get current quantity for a product."""
# TODO: Return quantity
pass
def adjust_stock(self, sku: str, quantity: int, reason: str = "") -> bool:
"""Adjust stock level (positive or negative)."""
# TODO: Implement stock adjustment
# - Validate product exists
# - Check for negative inventory
# - Record movement
pass
def transfer_stock(self, sku: str, quantity: int,
from_location: str, to_location: str) -> bool:
"""Transfer stock between locations."""
# TODO: Record transfer movement
pass
def get_low_stock_alerts(self, threshold: int = 10) -> list[Product]:
"""Get products below threshold."""
# TODO: Return low stock products
pass
def get_total_valuation(self) -> float:
"""Calculate total inventory value."""
# TODO: Sum of quantity * price for all products
pass
def get_movements(self, sku: str = None) -> list[dict]:
"""Get inventory movements history."""
# TODO: Return movements, optionally filtered
pass
class TestInventorySystem:
"""Tests for InventorySystem."""
def test_add_product(self):
"""Test adding a product."""
inventory = InventorySystem()
inventory.add_product(Product("SKU001", "Widget", 9.99, 50))
product = inventory.get_product("SKU001")
assert product is not None
assert product.quantity == 50
def test_stock_adjustment(self):
"""Test stock adjustments."""
inventory = InventorySystem()
inventory.add_product(Product("SKU001", "Widget", 9.99, 100))
inventory.adjust_stock("SKU001", -30, "sale")
assert inventory.get_quantity("SKU001") == 70
inventory.adjust_stock("SKU001", 20, "restock")
assert inventory.get_quantity("SKU001") == 90
def test_prevents_negative_stock(self):
"""Test that negative stock is prevented."""
inventory = InventorySystem()
inventory.add_product(Product("SKU001", "Widget", 9.99, 10))
result = inventory.adjust_stock("SKU001", -20, "sale")
assert result == False
assert inventory.get_quantity("SKU001") == 10
def test_low_stock_alerts(self):
"""Test low stock alerts."""
inventory = InventorySystem()
inventory.add_product(Product("SKU001", "High Stock", 9.99, 100))
inventory.add_product(Product("SKU002", "Low Stock", 9.99, 5))
alerts = inventory.get_low_stock_alerts(threshold=10)
assert len(alerts) == 1
assert alerts[0].sku == "SKU002"
# =============================================================================
# EXERCISE 6: Simple Web Framework
# =============================================================================
class Request:
"""Represents an HTTP request."""
def __init__(self, method: str, path: str, headers: dict = None,
body: str = ""):
self.method = method.upper()
self.path = path
self.headers = headers or {}
self.body = body
self.params: dict = {}
class Response:
"""Represents an HTTP response."""
def __init__(self, body: str = "", status: int = 200,
headers: dict = None):
self.body = body
self.status = status
self.headers = headers or {"Content-Type": "text/plain"}
class SimpleWebFramework:
"""
Create a simple web framework that:
- Registers routes with HTTP methods
- Matches requests to handlers
- Supports middleware
- Handles route parameters
- Returns proper responses
Example:
app = SimpleWebFramework()
@app.route("/hello/{name}", methods=["GET"])
def hello(request):
name = request.params["name"]
return Response(f"Hello, {name}!")
request = Request("GET", "/hello/World")
response = app.handle(request)
assert response.body == "Hello, World!"
"""
def __init__(self):
self.routes: list[dict] = []
self.middleware: list = []
def route(self, path: str, methods: list[str] = None):
"""Decorator to register a route handler."""
# TODO: Return decorator that registers route
pass
def use(self, middleware) -> None:
"""Add middleware."""
# TODO: Store middleware
pass
def _match_route(self, method: str, path: str) -> tuple:
"""Match request to a route."""
# TODO: Find matching route and extract params
pass
def handle(self, request: Request) -> Response:
"""Handle an incoming request."""
# TODO: Implement request handling
# - Run middleware
# - Match route
# - Extract parameters
# - Call handler
# - Return response
pass
class TestSimpleWebFramework:
"""Tests for SimpleWebFramework."""
def test_simple_route(self):
"""Test simple route matching."""
app = SimpleWebFramework()
@app.route("/hello", methods=["GET"])
def hello(request):
return Response("Hello!")
request = Request("GET", "/hello")
response = app.handle(request)
assert response.status == 200
assert response.body == "Hello!"
def test_route_parameters(self):
"""Test route with parameters."""
app = SimpleWebFramework()
@app.route("/users/{user_id}", methods=["GET"])
def get_user(request):
return Response(f"User: {request.params['user_id']}")
request = Request("GET", "/users/123")
response = app.handle(request)
assert response.body == "User: 123"
def test_method_matching(self):
"""Test HTTP method matching."""
app = SimpleWebFramework()
@app.route("/resource", methods=["GET"])
def get_resource(request):
return Response("GET")
@app.route("/resource", methods=["POST"])
def create_resource(request):
return Response("POST")
get_response = app.handle(Request("GET", "/resource"))
post_response = app.handle(Request("POST", "/resource"))
assert get_response.body == "GET"
assert post_response.body == "POST"
def test_404_response(self):
"""Test 404 for unmatched routes."""
app = SimpleWebFramework()
request = Request("GET", "/nonexistent")
response = app.handle(request)
assert response.status == 404
# =============================================================================
# EXERCISE 7: Build a Complete Application
# =============================================================================
class BookLibrary:
"""
Build a complete book library application that:
1. Book Management:
- Add, update, delete books
- Track ISBN, title, author, genre, copies available
2. Member Management:
- Register members
- Track borrowing history
- Set borrowing limits
3. Borrowing System:
- Check out books
- Return books
- Track due dates
- Handle late returns
4. Search & Discovery:
- Search by title, author, genre
- Get recommendations
- View popular books
5. Reporting:
- Overdue books report
- Member activity report
- Inventory report
Example:
library = BookLibrary()
# Add books
library.add_book(isbn="978-0-123456-78-9",
title="Python Programming",
author="John Smith",
genre="Technology",
copies=5)
# Register member
library.register_member(id="M001", name="Jane Doe",
email="jane@example.com")
# Borrow book
library.checkout("M001", "978-0-123456-78-9")
# Return book
library.return_book("M001", "978-0-123456-78-9")
# Get reports
overdue = library.get_overdue_report()
"""
def __init__(self):
# TODO: Initialize data structures
pass
# Book Management
def add_book(self, isbn: str, title: str, author: str,
genre: str, copies: int) -> bool:
pass
def update_book(self, isbn: str, **kwargs) -> bool:
pass
def remove_book(self, isbn: str) -> bool:
pass
def get_book(self, isbn: str) -> Optional[dict]:
pass
# Member Management
def register_member(self, id: str, name: str, email: str) -> bool:
pass
def get_member(self, id: str) -> Optional[dict]:
pass
def get_member_history(self, id: str) -> list[dict]:
pass
# Borrowing
def checkout(self, member_id: str, isbn: str,
days: int = 14) -> Optional[dict]:
pass
def return_book(self, member_id: str, isbn: str) -> dict:
pass
def renew(self, member_id: str, isbn: str, days: int = 7) -> bool:
pass
# Search
def search(self, query: str, field: str = "all") -> list[dict]:
pass
def get_recommendations(self, member_id: str) -> list[dict]:
pass
def get_popular(self, limit: int = 10) -> list[dict]:
pass
# Reports
def get_overdue_report(self) -> list[dict]:
pass
def get_inventory_report(self) -> dict:
pass
def get_member_activity(self, member_id: str) -> dict:
pass
class TestBookLibrary:
"""Tests for BookLibrary."""
def test_add_and_get_book(self):
"""Test adding and retrieving a book."""
library = BookLibrary()
library.add_book(
isbn="978-0-123456-78-9",
title="Test Book",
author="Author",
genre="Fiction",
copies=3
)
book = library.get_book("978-0-123456-78-9")
assert book is not None
assert book["title"] == "Test Book"
def test_member_registration(self):
"""Test member registration."""
library = BookLibrary()
library.register_member("M001", "John Doe", "john@example.com")
member = library.get_member("M001")
assert member is not None
assert member["name"] == "John Doe"
def test_checkout_and_return(self):
"""Test book checkout and return."""
library = BookLibrary()
library.add_book("ISBN001", "Book", "Author", "Genre", 2)
library.register_member("M001", "Member", "m@example.com")
# Checkout
result = library.checkout("M001", "ISBN001")
assert result is not None
book = library.get_book("ISBN001")
assert book["available_copies"] == 1
# Return
library.return_book("M001", "ISBN001")
book = library.get_book("ISBN001")
assert book["available_copies"] == 2
def test_cannot_checkout_unavailable(self):
"""Test cannot checkout when no copies available."""
library = BookLibrary()
library.add_book("ISBN001", "Book", "Author", "Genre", 1)
library.register_member("M001", "Member1", "m1@example.com")
library.register_member("M002", "Member2", "m2@example.com")
library.checkout("M001", "ISBN001") # Takes last copy
result = library.checkout("M002", "ISBN001") # Should fail
assert result is None
# =============================================================================
# MAIN
# =============================================================================
if __name__ == "__main__":
pytest.main([__file__, "-v"])