python
exercises
exercises.py🐍python
"""
API Development - Exercises
Build API components to practice REST API development.
Run with: pytest 27_api_development/exercises.py -v
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Callable
import json
# =============================================================================
# EXERCISE 1: Request Validator
# =============================================================================
@dataclass
class FieldError:
"""Validation error for a field."""
field: str
message: str
code: str
class RequestValidator:
"""
Implement a request validator for API endpoints.
Requirements:
- required(data, field) - field must exist and not be None/empty
- string(data, field, min_len, max_len) - validate string with length
- integer(data, field, min_val, max_val) - validate integer with range
- email(data, field) - validate email format (must contain @)
- one_of(data, field, choices) - value must be in choices list
- nested(data, field, validator) - validate nested object
- validate() - return (is_valid, errors) tuple
Example:
v = RequestValidator()
v.required(data, "name")
v.string(data, "name", min_len=2, max_len=50)
v.email(data, "email")
v.one_of(data, "role", ["admin", "user", "guest"])
is_valid, errors = v.validate()
"""
def __init__(self):
# YOUR CODE HERE
pass
def required(self, data: dict, field: str, message: str = None) -> 'RequestValidator':
# YOUR CODE HERE
pass
def string(self, data: dict, field: str, min_len: int = 0, max_len: int = None) -> 'RequestValidator':
# YOUR CODE HERE
pass
def integer(self, data: dict, field: str, min_val: int = None, max_val: int = None) -> 'RequestValidator':
# YOUR CODE HERE
pass
def email(self, data: dict, field: str) -> 'RequestValidator':
# YOUR CODE HERE
pass
def one_of(self, data: dict, field: str, choices: list) -> 'RequestValidator':
# YOUR CODE HERE
pass
def validate(self) -> tuple[bool, list[FieldError]]:
# YOUR CODE HERE
pass
def test_request_validator():
"""Test the RequestValidator."""
# Valid data
data = {"name": "Alice", "email": "alice@example.com", "age": 25, "role": "admin"}
v = RequestValidator()
v.required(data, "name")
v.string(data, "name", min_len=2, max_len=50)
v.email(data, "email")
v.integer(data, "age", min_val=18, max_val=120)
v.one_of(data, "role", ["admin", "user"])
is_valid, errors = v.validate()
assert is_valid == True
assert len(errors) == 0
# Invalid data
bad_data = {"name": "A", "email": "invalid", "age": 15, "role": "superuser"}
v = RequestValidator()
v.required(bad_data, "name")
v.string(bad_data, "name", min_len=2)
v.email(bad_data, "email")
v.integer(bad_data, "age", min_val=18)
v.one_of(bad_data, "role", ["admin", "user"])
is_valid, errors = v.validate()
assert is_valid == False
assert len(errors) >= 3
# =============================================================================
# EXERCISE 2: Response Builder
# =============================================================================
class ResponseBuilder:
"""
Implement a fluent API response builder.
Requirements:
- success() / error() - set success status
- data(value) - set response data
- message(text) - set message
- meta(key, value) - add metadata
- paginate(items, total, page, per_page) - add pagination
- status(code) - set HTTP status code
- header(key, value) - add response header
- build() - return final response dict
Example:
response = (ResponseBuilder()
.success()
.data({"users": [...]})
.message("Users retrieved successfully")
.paginate(users, total=100, page=1, per_page=10)
.header("X-Request-Id", "abc123")
.build())
"""
def __init__(self):
# YOUR CODE HERE
pass
def success(self) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def error(self, error_message: str = None) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def data(self, value: Any) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def message(self, text: str) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def meta(self, key: str, value: Any) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def paginate(self, items: list, total: int, page: int, per_page: int) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def status(self, code: int) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def header(self, key: str, value: str) -> 'ResponseBuilder':
# YOUR CODE HERE
pass
def build(self) -> dict:
# YOUR CODE HERE
pass
def test_response_builder():
"""Test the ResponseBuilder."""
# Success response
response = (ResponseBuilder()
.success()
.data({"id": 1, "name": "Alice"})
.message("User created")
.meta("request_id", "abc123")
.status(201)
.header("Location", "/users/1")
.build())
assert response["body"]["success"] == True
assert response["body"]["data"]["id"] == 1
assert response["body"]["message"] == "User created"
assert response["body"]["meta"]["request_id"] == "abc123"
assert response["status"] == 201
assert response["headers"]["Location"] == "/users/1"
# Paginated response
items = [{"id": i} for i in range(10)]
response = (ResponseBuilder()
.success()
.paginate(items, total=100, page=2, per_page=10)
.build())
assert len(response["body"]["data"]) == 10
assert response["body"]["pagination"]["total"] == 100
assert response["body"]["pagination"]["page"] == 2
assert response["body"]["pagination"]["total_pages"] == 10
# =============================================================================
# EXERCISE 3: Simple Token Authentication
# =============================================================================
class TokenAuth:
"""
Implement a simple token-based authentication system.
Requirements:
- generate_token(user_id, expires_in_minutes) - create a token
- validate_token(token) - return user_id if valid, None otherwise
- revoke_token(token) - invalidate a token
- cleanup_expired() - remove expired tokens
Token format can be simple: "{user_id}:{random_string}:{expiry_timestamp}"
Example:
auth = TokenAuth()
token = auth.generate_token("user123", expires_in_minutes=60)
user_id = auth.validate_token(token) # Returns "user123"
auth.revoke_token(token)
user_id = auth.validate_token(token) # Returns None
"""
def __init__(self):
# YOUR CODE HERE
pass
def generate_token(self, user_id: str, expires_in_minutes: int = 60) -> str:
# YOUR CODE HERE
pass
def validate_token(self, token: str) -> str | None:
# YOUR CODE HERE
pass
def revoke_token(self, token: str) -> bool:
# YOUR CODE HERE
pass
def cleanup_expired(self) -> int:
"""Remove expired tokens and return count of removed."""
# YOUR CODE HERE
pass
def get_active_tokens(self, user_id: str) -> list[str]:
"""Get all active tokens for a user."""
# YOUR CODE HERE
pass
def test_token_auth():
"""Test the TokenAuth system."""
auth = TokenAuth()
# Generate token
token = auth.generate_token("user123", expires_in_minutes=60)
assert token is not None
assert len(token) > 10
# Validate token
user_id = auth.validate_token(token)
assert user_id == "user123"
# Invalid token
assert auth.validate_token("invalid-token") is None
# Revoke token
assert auth.revoke_token(token) == True
assert auth.validate_token(token) is None
# Multiple tokens for same user
token1 = auth.generate_token("user456")
token2 = auth.generate_token("user456")
active = auth.get_active_tokens("user456")
assert len(active) == 2
# =============================================================================
# EXERCISE 4: Rate Limiter
# =============================================================================
@dataclass
class RateLimitResult:
"""Result of rate limit check."""
allowed: bool
limit: int
remaining: int
reset_at: datetime
class SlidingWindowRateLimiter:
"""
Implement a sliding window rate limiter.
Requirements:
- check(key) - check if request is allowed, return RateLimitResult
- get_usage(key) - return current usage count
- reset(key) - reset rate limit for a key
Use sliding window algorithm:
- Track timestamps of requests
- Only count requests within the window
Example:
limiter = SlidingWindowRateLimiter(max_requests=10, window_seconds=60)
result = limiter.check("user123")
print(result.allowed) # True
print(result.remaining) # 9
"""
def __init__(self, max_requests: int = 100, window_seconds: int = 60):
# YOUR CODE HERE
pass
def check(self, key: str) -> RateLimitResult:
# YOUR CODE HERE
pass
def get_usage(self, key: str) -> int:
# YOUR CODE HERE
pass
def reset(self, key: str) -> None:
# YOUR CODE HERE
pass
def test_sliding_window_rate_limiter():
"""Test the SlidingWindowRateLimiter."""
limiter = SlidingWindowRateLimiter(max_requests=5, window_seconds=60)
# First request should be allowed
result = limiter.check("user1")
assert result.allowed == True
assert result.remaining == 4
assert result.limit == 5
# Make more requests
for _ in range(4):
result = limiter.check("user1")
# 5th request was last allowed
assert result.allowed == True
assert result.remaining == 0
# 6th request should be denied
result = limiter.check("user1")
assert result.allowed == False
# Different user should have own limit
result = limiter.check("user2")
assert result.allowed == True
assert result.remaining == 4
# Reset should work
limiter.reset("user1")
result = limiter.check("user1")
assert result.allowed == True
assert result.remaining == 4
# =============================================================================
# EXERCISE 5: API Router
# =============================================================================
@dataclass
class Route:
"""A route definition."""
path: str
method: str
handler: Callable
middleware: list = field(default_factory=list)
class Router:
"""
Implement an API router with path parameters.
Requirements:
- add_route(method, path, handler) - register a route
- get/post/put/delete decorators for convenience
- match(method, path) - find matching route and extract params
- use(middleware) - add global middleware
- group(prefix) - create a sub-router with prefix
Path parameter format: /users/{id}/posts/{post_id}
Example:
router = Router()
@router.get("/users/{user_id}")
def get_user(user_id: str):
return {"id": user_id}
route, params = router.match("GET", "/users/123")
# route.handler, params = {"user_id": "123"}
"""
def __init__(self, prefix: str = ""):
# YOUR CODE HERE
pass
def add_route(self, method: str, path: str, handler: Callable, middleware: list = None):
# YOUR CODE HERE
pass
def get(self, path: str):
"""Decorator for GET routes."""
# YOUR CODE HERE
pass
def post(self, path: str):
"""Decorator for POST routes."""
# YOUR CODE HERE
pass
def put(self, path: str):
"""Decorator for PUT routes."""
# YOUR CODE HERE
pass
def delete(self, path: str):
"""Decorator for DELETE routes."""
# YOUR CODE HERE
pass
def match(self, method: str, path: str) -> tuple[Route | None, dict]:
"""Match a route and extract path parameters."""
# YOUR CODE HERE
pass
def use(self, middleware: Callable):
"""Add global middleware."""
# YOUR CODE HERE
pass
def group(self, prefix: str) -> 'Router':
"""Create a sub-router with prefix."""
# YOUR CODE HERE
pass
def test_router():
"""Test the Router."""
router = Router()
@router.get("/users")
def list_users():
return "list"
@router.get("/users/{user_id}")
def get_user(user_id):
return f"user {user_id}"
@router.post("/users")
def create_user():
return "created"
@router.get("/users/{user_id}/posts/{post_id}")
def get_user_post(user_id, post_id):
return f"user {user_id} post {post_id}"
# Test matching
route, params = router.match("GET", "/users")
assert route is not None
assert params == {}
route, params = router.match("GET", "/users/123")
assert route is not None
assert params == {"user_id": "123"}
route, params = router.match("GET", "/users/123/posts/456")
assert route is not None
assert params == {"user_id": "123", "post_id": "456"}
# Test no match
route, params = router.match("DELETE", "/users")
assert route is None
# =============================================================================
# EXERCISE 6: CRUD Resource
# =============================================================================
class CRUDResource:
"""
Implement a generic CRUD resource handler.
Requirements:
- list(filters, page, per_page) - list items with pagination
- get(id) - get single item
- create(data) - create new item
- update(id, data) - update item
- delete(id) - delete item
- find(query) - search items
Uses in-memory storage for demonstration.
Example:
users = CRUDResource("users")
user = users.create({"name": "Alice", "email": "alice@example.com"})
print(user["id"]) # Auto-generated ID
all_users = users.list(page=1, per_page=10)
alice = users.get(user["id"])
users.update(user["id"], {"name": "Alice Smith"})
users.delete(user["id"])
"""
def __init__(self, name: str):
# YOUR CODE HERE
pass
def list(self, filters: dict = None, page: int = 1, per_page: int = 10) -> dict:
"""
List items with filtering and pagination.
Returns: {"items": [...], "total": N, "page": N, "per_page": N}
"""
# YOUR CODE HERE
pass
def get(self, id: int) -> dict | None:
"""Get a single item by ID."""
# YOUR CODE HERE
pass
def create(self, data: dict) -> dict:
"""Create a new item. Auto-generates ID and timestamps."""
# YOUR CODE HERE
pass
def update(self, id: int, data: dict) -> dict | None:
"""Update an item. Returns None if not found."""
# YOUR CODE HERE
pass
def delete(self, id: int) -> bool:
"""Delete an item. Returns True if deleted."""
# YOUR CODE HERE
pass
def find(self, query: dict) -> list[dict]:
"""Find items matching query (simple field matching)."""
# YOUR CODE HERE
pass
def test_crud_resource():
"""Test the CRUDResource."""
users = CRUDResource("users")
# Create
user1 = users.create({"name": "Alice", "email": "alice@example.com"})
assert "id" in user1
assert user1["name"] == "Alice"
assert "created_at" in user1
user2 = users.create({"name": "Bob", "email": "bob@example.com"})
assert user2["id"] != user1["id"]
# Get
fetched = users.get(user1["id"])
assert fetched["name"] == "Alice"
assert users.get(9999) is None
# List
result = users.list()
assert result["total"] == 2
assert len(result["items"]) == 2
# Update
updated = users.update(user1["id"], {"name": "Alice Smith"})
assert updated["name"] == "Alice Smith"
assert "updated_at" in updated
# Find
found = users.find({"name": "Bob"})
assert len(found) == 1
assert found[0]["email"] == "bob@example.com"
# Delete
assert users.delete(user1["id"]) == True
assert users.get(user1["id"]) is None
assert users.delete(user1["id"]) == False
# =============================================================================
# EXERCISE 7: API Error Handler
# =============================================================================
class APIError(Exception):
"""Base API error."""
def __init__(self, message: str, code: str, status: int = 400, details: Any = None):
super().__init__(message)
self.message = message
self.code = code
self.status = status
self.details = details
class NotFoundError(APIError):
"""Resource not found error."""
def __init__(self, resource: str, id: Any):
super().__init__(
message=f"{resource} with id '{id}' not found",
code="not_found",
status=404
)
class ValidationError(APIError):
"""Validation error."""
def __init__(self, errors: list[dict]):
super().__init__(
message="Validation failed",
code="validation_error",
status=400,
details=errors
)
class ErrorHandler:
"""
Implement an API error handler.
Requirements:
- register(exception_class, handler) - register error handler
- handle(exception) - convert exception to API response
- default handlers for common errors
Example:
handler = ErrorHandler()
@handler.register(NotFoundError)
def handle_not_found(error):
return {"status": 404, "error": error.message}
try:
raise NotFoundError("User", 123)
except Exception as e:
response = handler.handle(e)
"""
def __init__(self):
# YOUR CODE HERE
pass
def register(self, exception_class: type):
"""Decorator to register an error handler."""
# YOUR CODE HERE
pass
def handle(self, exception: Exception) -> dict:
"""Convert exception to API response dict."""
# YOUR CODE HERE
pass
def to_response(self, error: APIError) -> dict:
"""Convert APIError to response dict."""
# YOUR CODE HERE
pass
def test_error_handler():
"""Test the ErrorHandler."""
handler = ErrorHandler()
# Test NotFoundError
error = NotFoundError("User", 123)
response = handler.handle(error)
assert response["status"] == 404
assert "not found" in response["body"]["error"].lower()
# Test ValidationError
error = ValidationError([
{"field": "email", "message": "Invalid email"}
])
response = handler.handle(error)
assert response["status"] == 400
assert response["body"]["code"] == "validation_error"
assert len(response["body"]["details"]) == 1
# Test generic exception
error = Exception("Something went wrong")
response = handler.handle(error)
assert response["status"] == 500
# =============================================================================
# RUN TESTS
# =============================================================================
if __name__ == "__main__":
import pytest
pytest.main([__file__, "-v"])