Witajcie! No to dotarliśmy do ostatniej części serii o wyjątkach. Skoro tu jesteście, to znaczy, że podstawy i średniozaawansowane techniki macie już w małym palcu. Teraz czas na mięso – rzeczy, które odróżniają seniorów od juniorów.
Dziś pogadamy o wyjątkach w asyncio (tak, to może boleć), obsłudze błędów we frameworkach webowych, testowaniu kodu z wyjątkami i najnowszych ficzerach z Pythona 3.11+. Będzie konkretnie, ale obiecuję – wytłumaczę wszystko od podstaw. No to jedziemy!
Wyjątki w programowaniu asynchronicznym (asyncio)
Programowanie asynchroniczne to taki sposób pisania kodu, że może robić kilka rzeczy naraz bez blokowania się. Brzmi super, prawda? No ale wyjątki w asyncio zachowują się… dziwnie.
Podstawy asyncio i wyjątków
Zacznijmy od prostego przykładu, który symuluje pobieranie danych z internetu i pokazuje, jak łapać wyjątki w kontekście async:
import asyncio
async def fetch_data(url):
"""Simulate fetching data from the internet"""
print(f"Fetching data from: {url}")
await asyncio.sleep(2) # Simulate network delay
# Simulate error for certain URLs
if "error" in url:
raise ValueError(f"Invalid URL: {url}")
return f"Data from {url}"
async def main():
try:
result = await fetch_data("https://example.com")
print(f"Received: {result}")
except ValueError as e:
print(f"Error: {e}")
# Run the async function
asyncio.run(main())try/except działa normalnie – wyjątek z fetch_data() wpada w blok except jak w każdym innym kodzie.
Problem z wieloma zadaniami – asyncio.gather()
Kiedy zaczniemy odpalać kilka zadań naraz, pojawia się kwestia, co zrobić z błędami. Poniższy listing demonstruje domyślne zachowanie asyncio.gather() – gdy jedno zadanie rzuci wyjątek, pierwsza napotkana wyjątkowa sytuacja jest natychmiast propagowana do miejsca wywołania, a pozostałe zadania wciąż działają w tle (nie są automatycznie anulowane). Jeśli w tym momencie kończysz pętlę zdarzeń (np. przez asyncio.run()), pozostałe zadania nie zostaną już obserwowane. Obsługa ich wyników lub debugowanie dziwnego stanu systemu, który powstał w ten sposób, to katorga.
import asyncio
async def task_success(name, duration):
"""Task that will succeed"""
await asyncio.sleep(duration)
return f"Success: {name}"
async def task_error(name, duration):
"""Task that will raise an error"""
await asyncio.sleep(duration)
raise ValueError(f"Error in task: {name}")
async def test_default_behavior():
"""Test default gather() behavior on error"""
print("=== Testing default behavior ===")
try:
results = await asyncio.gather(
task_success("A", 1),
task_error("B", 2), # This will raise error
task_success("C", 3),
)
print(f"Results: {results}")
except ValueError as e:
print(f"Caught error: {e}")
print("NOTE: other tasks may still be running in the background!")
asyncio.run(test_default_behavior())Uwaga: asyncio.gather() nie przerywa automatycznie działania pozostałych zadań. Żeby zebrać wyniki wszystkich zadań i otrzymać wyjątki jako elementy listy, użyj parametru return_exceptions=True.
Rozwiązanie – return_exceptions=True
Jeśli chcesz wykonać wszystkie zadania do końca i otrzymać także błędy, ustaw return_exceptions=True. Wówczas lista wyników będzie zawierała obiekty wyjątków zamiast przerywać działanie:
async def test_with_return_exceptions():
"""Test with return_exceptions=True - collect all results"""
print("\n=== Testing with return_exceptions=True ===")
results = await asyncio.gather(
task_success("A", 1),
task_error("B", 2),
task_success("C", 3),
return_exceptions=True # Key parameter!
)
print("Analyzing results:")
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Task {i}: ERROR - {result}")
else:
print(f"Task {i}: SUCCESS - {result}")
asyncio.run(test_with_return_exceptions())Wynik:
Task 0: SUCCESS - Success: A
Task 1: ERROR - Error in task: B
Task 2: SUCCESS - Success: CTeraz wszystkie zadania się wykonują, a błędy dostajemy jako wartości. Petarda!
Zaawansowane zarządzanie błędami w asyncio
Jeśli potrzebujesz większej kontroli, możesz owinąć każde zadanie w „bezpieczny” wrapper i zwracać obiekt z informacją o sukcesie lub błędzie:
import asyncio
import logging
from typing import List, Union, Any
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TaskResult:
"""Container for async task result"""
def __init__(self, name: str, result: Any = None, error: Exception = None):
self.name = name
self.result = result
self.error = error
self.success = error is None
def __str__(self):
if self.success:
return f"✓ {self.name}: {self.result}"
else:
return f"✗ {self.name}: {self.error}"
async def safe_task_wrapper(name: str, coroutine):
"""Wrapper that never raises exceptions"""
try:
result = await coroutine
logger.info(f"Task {name} completed successfully")
return TaskResult(name, result=result)
except Exception as e:
logger.error(f"Task {name} failed with error: {e}")
return TaskResult(name, error=e)
async def task_manager(tasks: List[tuple]):
"""Manages execution of multiple tasks with professional error handling"""
logger.info(f"Starting execution of {len(tasks)} tasks")
# Wrap each task in safe wrapper
safe_tasks = [
safe_task_wrapper(name, task)
for name, task in tasks
]
# All tasks will execute regardless of errors
results = await asyncio.gather(*safe_tasks)
# Analyze results
successes = [r for r in results if r.success]
errors = [r for r in results if not r.success]
logger.info(f"Completed: {len(successes)} successes, {len(errors)} errors")
return results
# Example usage
async def demo_advanced_management():
"""Demo of advanced task management"""
async def fetch_data(url):
"""Simulate API call"""
await asyncio.sleep(1)
if "error" in url:
raise ValueError(f"Invalid URL: {url}")
return f"Data from {url}"
tasks = [
("Fetch users", fetch_data("https://api.example.com/users")),
("Fetch products", fetch_data("https://error.example.com/products")), # Error!
("Fetch categories", fetch_data("https://api.example.com/categories")),
("Fetch orders", fetch_data("https://api.example.com/orders")),
]
results = await task_manager(tasks)
print("\n=== EXECUTION REPORT ===")
for result in results:
print(result)
# Check if any critical tasks failed
critical_tasks = ["Fetch users", "Fetch products"]
critical_errors = [
r for r in results
if not r.success and r.name in critical_tasks
]
if critical_errors:
print(f"\n⚠️ WARNING: {len(critical_errors)} critical errors!")
for error in critical_errors:
print(f" - {error}")
asyncio.run(demo_advanced_management())Timeouty i anulowanie zadań
W produkcji musicie ustawiać timeouty. Serio. Nie chcecie czekać w nieskończoność na odpowiedź z API, które padło. Poniższy kod pokazuje, jak wykorzystać asyncio.wait_for() do ograniczenia czasu wykonania pojedynczego zadania oraz całej grupy zadań. Pamiętaj tylko, że jeśli anulujesz zadanie, próba odczytu wyniku przez task.result() lub task.exception() może rzucić CancelledError, więc warto to obsłużyć w try/except.
async def task_with_timeout():
"""Demonstrate timeout handling"""
async def long_running_task():
"""Task that takes very long"""
await asyncio.sleep(10) # 10 seconds
return "Finally done!"
try:
# Wait maximum 3 seconds
result = await asyncio.wait_for(long_running_task(), timeout=3.0)
print(f"Result: {result}")
except asyncio.TimeoutError:
print("Task exceeded timeout (3 seconds)")
except Exception as e:
print(f"Other error: {e}")
# Test with multiple tasks and timeout
async def test_multiple_tasks_timeout():
"""Test timeout for multiple tasks"""
tasks = [
asyncio.create_task(task_success("Fast", 1)),
asyncio.create_task(task_success("Medium", 3)),
asyncio.create_task(task_success("Slow", 8)),
]
try:
# Timeout for entire group
results = await asyncio.wait_for(
asyncio.gather(*tasks),
timeout=5.0
)
print(f"All tasks completed: {results}")
except asyncio.TimeoutError:
print("Some tasks exceeded timeout")
# Check which tasks completed
for i, task in enumerate(tasks):
if task.done():
try:
if task.exception():
print(f"Task {i}: error - {task.exception()}")
else:
print(f"Task {i}: success - {task.result()}")
except asyncio.CancelledError:
print(f"Task {i}: was cancelled")
else:
print(f"Task {i}: not completed (cancelling...)")
task.cancel()
asyncio.run(test_multiple_tasks_timeout())Asynchroniczne context managers
Pamiętacie kontekst menedżerów z with? W asyncio też są, tylko asynchroniczne. Wymagają zdefiniowania metod __aenter__() i __aexit__(), które same w sobie są coroutine’ami:
import aiohttp
import asyncio
class AsyncDatabaseConnection:
"""Example of async context manager"""
def __init__(self, database_url):
self.database_url = database_url
self.connection = None
async def __aenter__(self):
print(f"Connecting asynchronously to database: {self.database_url}")
# Real database connection would go here
await asyncio.sleep(0.1) # Simulate connection delay
self.connection = f"connection to {self.database_url}"
return self.connection
async def __aexit__(self, exc_type, exc_value, traceback):
print("Closing database connection")
if exc_type:
print(f"Error detected, rolling back transaction: {exc_value}")
# Real rollback would go here
else:
print("Success, committing transaction")
# Real commit would go here
await asyncio.sleep(0.1) # Simulate closing delay
self.connection = None
return False # Don't suppress exceptions
async def demo_async_context_manager():
"""Demo async context manager"""
try:
async with AsyncDatabaseConnection("postgresql://localhost/mydb") as db:
print(f"Working with database: {db}")
# Simulate database operation
await asyncio.sleep(1)
# Raise error to see rollback
raise ValueError("Database operation error!")
except ValueError as e:
print(f"Handled error: {e}")
asyncio.run(demo_async_context_manager())Wyjątki w frameworkach webowych
Każdy framework ma swój sposób na błędy. Zobaczmy, jak to wygląda w najpopularniejszych.
Django – middleware i globalna obsługa błędów
W Django można napisać middleware, które przechwyci wyjątki w całej aplikacji. Funkcja process_exception() zostanie wywołana, gdy widok rzuci wyjątek i powinna zwrócić HttpResponse lub None:
# middleware.py
import logging
import json
from django.http import JsonResponse
from django.core.exceptions import ValidationError, PermissionDenied
from django.db import DatabaseError
from django.contrib.auth.models import User
from django.conf import settings # brakowało tego importu
logger = logging.getLogger(__name__)
class GlobalErrorHandler:
"""Middleware for global error handling in Django"""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
response = self.get_response(request)
return response
def process_exception(self, request, exception):
"""Method called when exception occurs in view"""
# Log all errors with context
logger.error(
f"Error in {request.method} {request.path}",
extra={
'user': getattr(request, 'user', None),
'POST_data': dict(request.POST) if request.POST else None,
'GET_data': dict(request.GET) if request.GET else None,
},
exc_info=True
)
# Handle different error types
if isinstance(exception, ValidationError):
return JsonResponse({
'error': 'Validation error',
'message': str(exception),
'code': 'VALIDATION_ERROR'
}, status=400)
elif isinstance(exception, PermissionDenied):
return JsonResponse({
'error': 'Permission denied',
'message': 'You do not have permission for this operation',
'code': 'PERMISSION_DENIED'
}, status=403)
elif isinstance(exception, DatabaseError):
return JsonResponse({
'error': 'Database error',
'message': 'Database problem occurred',
'code': 'DATABASE_ERROR'
}, status=500)
# For other errors in production don't reveal details
if settings.DEBUG:
return JsonResponse({
'error': 'Server error',
'message': str(exception),
'code': 'INTERNAL_ERROR'
}, status=500)
else:
return JsonResponse({
'error': 'Server error',
'message': 'An unexpected error occurred',
'code': 'INTERNAL_ERROR'
}, status=500)
# views.py – examples of error handling in views
from django.shortcuts import get_object_or_404
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
import json
class ApplicationError(Exception):
"""Custom application error"""
def __init__(self, message, code=None, status=400):
super().__init__(message)
self.code = code
self.status = status
@csrf_exempt
@require_http_methods(["POST"])
def create_user(request):
"""Endpoint to create user with error handling"""
try:
# Parse JSON
try:
data = json.loads(request.body)
except json.JSONDecodeError:
raise ApplicationError(
"Invalid JSON format",
code="INVALID_JSON",
status=400
)
# Validate required fields
required_fields = ['email', 'name', 'password']
missing_fields = [field for field in required_fields if not data.get(field)]
if missing_fields:
raise ApplicationError(
f"Missing fields: {', '.join(missing_fields)}",
code="MISSING_FIELDS",
status=400
)
# Validate email
email = data['email']
if '@' not in email:
raise ApplicationError(
"Invalid email format",
code="INVALID_EMAIL",
status=400
)
# Check if user already exists
if User.objects.filter(email=email).exists():
raise ApplicationError(
f"User with email {email} already exists",
code="USER_EXISTS",
status=409
)
# Create user
user = User.objects.create_user(
username=email,
email=email,
first_name=data['name'],
password=data['password']
)
return JsonResponse({
'success': True,
'user_id': user.id,
'message': 'User created successfully'
}, status=201)
except ApplicationError as e:
return JsonResponse({
'error': str(e),
'code': e.code
}, status=e.status)
except Exception as e:
# This error will be handled by middleware
logger.error(f"Unexpected error in create_user: {e}", exc_info=True)
raise
# settings.py – add middleware
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
# ... other middleware
'myapp.middleware.GlobalErrorHandler', # Our middleware
]
Flask – handlery błędów
Flask jest prostszy, ale super elastyczny. Możesz definiować własne wyjątki i globalne handlery:
from flask import Flask, request, jsonify
import logging
from werkzeug.exceptions import HTTPException
app = Flask(__name__)
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Custom exceptions for Flask
class APIError(Exception):
"""Base API error"""
def __init__(self, message, status_code=400, payload=None):
super().__init__(message)
self.message = message
self.status_code = status_code
self.payload = payload
class ValidationError(APIError):
"""Data validation error"""
def __init__(self, message, field=None):
super().__init__(message, 400)
self.field = field
class AuthorizationError(APIError):
"""Authorization error"""
def __init__(self, message="Unauthorized"):
super().__init__(message, 403)
# Global error handlers
@app.errorhandler(APIError)
def handle_api_error(error):
"""Handler for our custom API errors"""
response = {
'error': error.message,
'status_code': error.status_code,
'type': error.__class__.__name__
}
# Add additional info if exists
if hasattr(error, 'field') and error.field:
response['field'] = error.field
if error.payload:
response.update(error.payload)
logger.warning(f"API error: {error.message}")
return jsonify(response), error.status_code
@app.errorhandler(HTTPException)
def handle_http_exception(e):
"""Handler for Werkzeug HTTP errors"""
response = {
'error': e.description,
'status_code': e.code,
'type': 'HTTPException'
}
logger.error(f"HTTP error {e.code}: {e.description}")
return jsonify(response), e.code
@app.errorhandler(Exception)
def handle_generic_exception(e):
"""Handler for all other errors"""
logger.error(f"Unexpected error: {e}", exc_info=True)
# In production don't reveal details
if app.debug:
return jsonify({
'error': str(e),
'status_code': 500,
'type': 'InternalError'
}), 500
else:
return jsonify({
'error': 'Server error occurred',
'status_code': 500,
'type': 'InternalError'
}), 500
# Decorator for request logging
from functools import wraps
def log_request(f):
"""Decorator to log request details"""
@wraps(f)
def decorated_function(*args, **kwargs):
logger.info(f"Request: {request.method} {request.path}")
logger.debug(f"Request data: {request.get_json()}")
try:
return f(*args, **kwargs)
except Exception as e:
logger.error(
f"Error in {f.__name__}: {e}",
extra={
'endpoint': request.endpoint,
'method': request.method,
'path': request.path,
'remote_addr': request.remote_addr
},
exc_info=True
)
raise
return decorated_function
# Example endpoints with error handling
@app.route('/api/users', methods=['POST'])
@log_request
def create_user():
"""Endpoint to create user"""
data = request.get_json()
# Validate data presence
if not data:
raise ValidationError("No data in request")
# Validate required fields
required_fields = ['email', 'name']
missing_fields = [field for field in required_fields if not data.get(field)]
if missing_fields:
raise ValidationError(
f"Missing fields: {', '.join(missing_fields)}",
field=missing_fields[0]
)
# Validate email
email = data['email']
if '@' not in email:
raise ValidationError("Invalid email format", field='email')
# Simulate checking if user exists
if email == "[email protected]":
raise APIError(
f"User with email {email} already exists",
status_code=409,
payload={'existing_email': email}
)
# Simulate user creation
user_id = 12345
logger.info(f"Created user: {email}")
return jsonify({
'success': True,
'user_id': user_id,
'message': 'User created successfully'
}), 201
@app.route('/api/protected')
@log_request
def protected_endpoint():
"""Endpoint requiring authorization"""
auth_header = request.headers.get('Authorization')
if not auth_header:
raise AuthorizationError("Missing Authorization header")
if not auth_header.startswith('Bearer '):
raise AuthorizationError("Invalid token format")
token = auth_header[7:] # Remove "Bearer "
if token != "valid_token":
raise AuthorizationError("Invalid token")
return jsonify({'message': 'Access granted'})
if __name__ == '__main__':
app.run(debug=True)FastAPI – nowoczesna obsługa błędów
FastAPI ma wbudowaną walidację i automatyczną dokumentację. Dzięki dekoratorom @app.exception_handler możesz globalnie obsługiwać błędy w aplikacji:
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.exception_handlers import http_exception_handler
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError
import logging
from typing import Optional
app = FastAPI(title="API with advanced error handling")
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Pydantic models with validation
class UserCreate(BaseModel):
email: str
name: str
age: Optional[int] = None
class Config:
# Examples for API documentation
schema_extra = {
"example": {
"email": "[email protected]",
"name": "John Doe",
"age": 30
}
}
class UserResponse(BaseModel):
id: int
email: str
name: str
age: Optional[int]
# Custom FastAPI exceptions
class BusinessLogicException(Exception):
"""Business logic error"""
def __init__(self, message: str, error_code: str = None):
self.message = message
self.error_code = error_code
class UserAlreadyExistsException(BusinessLogicException):
"""User already exists"""
def __init__(self, email: str):
super().__init__(
f"User with email {email} already exists",
"USER_ALREADY_EXISTS"
)
self.email = email
# Global exception handlers
@app.exception_handler(BusinessLogicException)
async def business_logic_exception_handler(request: Request, exc: BusinessLogicException):
"""Handler for business logic errors"""
logger.warning(f"Business logic error: {exc.message}")
return JSONResponse(
status_code=422,
content={
"error": "Business logic error",
"message": exc.message,
"error_code": exc.error_code,
# In reality use datetime.utcnow().isoformat()
"timestamp": "2024-01-01T12:00:00Z"
}
)
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
"""Handler for Pydantic validation errors"""
logger.warning(f"Validation error: {exc.errors()}")
return JSONResponse(
status_code=422,
content={
"error": "Validation error",
"message": "Submitted data failed validation",
"details": exc.errors()
}
)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""Global handler for all other errors"""
logger.error(f"Unexpected error: {exc}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error": "Server error",
"message": "An unexpected error occurred"
}
)
# Dependency for request logging
async def log_request_info(request: Request):
"""Dependency to log request info"""
logger.info(f"Request: {request.method} {request.url.path}")
return request
# Endpoints with error handling
@app.post("/api/users", response_model=UserResponse)
async def create_user(
user_data: UserCreate,
request: Request = Depends(log_request_info)
):
"""
Create new user
- **email**: User email address
- **name**: User name
- **age**: User age (optional)
"""
# Additional business validation
if user_data.age and user_data.age < 0:
raise BusinessLogicException(
"Age cannot be negative",
"INVALID_AGE"
)
if user_data.age and user_data.age > 150:
raise BusinessLogicException(
"Age seems unrealistic",
"UNREALISTIC_AGE"
)
# Check if user already exists
if user_data.email == "[email protected]":
raise UserAlreadyExistsException(user_data.email)
# Simulate user creation
new_user = UserResponse(
id=12345,
email=user_data.email,
name=user_data.name,
age=user_data.age
)
logger.info(f"Created user: {user_data.email}")
return new_user
@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
"""Get user by ID"""
# Validate ID
if user_id <= 0:
raise HTTPException(
status_code=400,
detail="User ID must be positive"
)
# Simulate checking if user exists
if user_id == 999:
raise HTTPException(
status_code=404,
detail=f"User with ID {user_id} not found"
)
# Simulate server error
if user_id == 666:
raise Exception("Simulated server error")
# Return user
return UserResponse(
id=user_id,
email="[email protected]",
name="John Doe",
age=30
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Testowanie kodu z wyjątkami
Testowanie błędów jest równie ważne jak testowanie happy path. Pokażę wam, jak to robić z pytest (lub ręcznie, jeśli nie macie pytesta).
Podstawy testowania wyjątków z pytest
Poniżej prosta funkcja dzieląca liczby i walidująca e-mail, plus testy, które sprawdzają, czy zgłaszane są odpowiednie wyjątki:
import pytest
def divide(a, b):
"""Function to test"""
if b == 0:
raise ValueError("Cannot divide by zero")
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
raise TypeError("Arguments must be numbers")
return a / b
def validate_email(email):
"""Email validation function"""
if not email:
raise ValidationError("Email cannot be empty")
if "@" not in email:
raise ValidationError("Email must contain @ sign")
return email.lower().strip()
# Basic exception tests
def test_division_by_zero():
"""Test that division by zero raises ValueError"""
with pytest.raises(ValueError) as exc_info:
divide(10, 0)
# Check error message is correct
assert str(exc_info.value) == "Cannot divide by zero"
def test_division_wrong_type():
"""Test that wrong types raise TypeError"""
with pytest.raises(TypeError, match="Arguments must be numbers"):
divide("10", 5)
def test_empty_email_validation():
"""Test empty email validation"""
with pytest.raises(ValidationError) as exc_info:
validate_email("")
assert "empty" in str(exc_info.value)
def test_email_without_at_sign():
"""Test email without @ sign"""
with pytest.raises(ValidationError, match="must contain @ sign"):
validate_email("invalid-email")
# Test that NO exception is raised
def test_division_success():
"""Test that correct division doesn't raise exception"""
result = divide(10, 2)
assert result == 5.0
def test_valid_email_validation():
"""Test that valid email passes validation"""
email = validate_email(" [email protected] ")
assert email == "[email protected]"Testowanie z parametryzacją
Lubię parametryzowane testy – mniej kodu, więcej przypadków:
import pytest
# Parametrized exception tests
@pytest.mark.parametrize("a, b, expected_exception", [
(10, 0, ValueError), # Division by zero
("10", 5, TypeError), # Wrong type first argument
(10, "5", TypeError), # Wrong type second argument
(None, 5, TypeError), # None as argument
])
def test_division_errors(a, b, expected_exception):
"""Parametrized test for various division errors"""
with pytest.raises(expected_exception):
divide(a, b)
@pytest.mark.parametrize("email, expected_message", [
("", "empty"),
(" ", "empty"), # Only spaces
("invalid", "@"),
("@", "@"), # Only @
("user@", "@"), # @ at the end
])
def test_invalid_email_validation(email, expected_message):
"""Parametrized test for various invalid emails"""
with pytest.raises(ValidationError, match=expected_message):
validate_email(email)
# Test correct cases
@pytest.mark.parametrize("a, b, expected", [
(10, 2, 5.0),
(15, 3, 5.0),
(-10, 2, -5.0),
(0, 5, 0.0),
])
def test_division_success_cases(a, b, expected):
"""Test correct division cases"""
assert divide(a, b) == expectedTestowanie async kodu z wyjątkami
Async testy też są proste:
import pytest
import asyncio
async def async_fetch_data(user_id):
"""Async function to test"""
await asyncio.sleep(0.1) # Simulate delay
if user_id <= 0:
raise ValueError("User ID must be positive")
if user_id == 404:
raise FileNotFoundError(f"User {user_id} not found")
return f"User data for {user_id}"
# Async exception tests
@pytest.mark.asyncio
async def test_async_invalid_id():
"""Test invalid ID in async function"""
with pytest.raises(ValueError, match="positive"):
await async_fetch_data(-1)
@pytest.mark.asyncio
async def test_async_user_not_found():
"""Test non-existent user"""
with pytest.raises(FileNotFoundError):
await async_fetch_data(404)
@pytest.mark.asyncio
async def test_async_success_data():
"""Test successful data fetch"""
result = await async_fetch_data(123)
assert result == "User data for 123"Testowanie handlerów błędów w Flask/FastAPI
import pytest
from flask import Flask
from myapp import create_app, APIError
@pytest.fixture
def app():
"""Fixture creating Flask app for tests"""
app = create_app(testing=True)
return app
@pytest.fixture
def client(app):
"""Fixture creating test client"""
return app.test_client()
def test_api_error_handler(client):
"""Test API error handler"""
# Endpoint that raises APIError
response = client.post('/api/users', json={
'email': 'invalid-email' # Invalid email
})
assert response.status_code == 400
data = response.get_json()
assert data['error'] == 'Invalid email format'
assert data['type'] == 'ValidationError'
def test_validation_error_handler(client):
"""Test validation error handler"""
response = client.post('/api/users', json={
# Missing required fields
})
assert response.status_code == 400
data = response.get_json()
assert 'Missing fields' in data['error']
def test_internal_error_handler(client):
"""Test internal error handler"""
# Endpoint that raises regular Exception
response = client.get('/api/internal-error')
assert response.status_code == 500
data = response.get_json()
assert data['error'] == 'Server error'Mocking i testowanie integracji
import pytest
from unittest.mock import patch, Mock
import requests
def fetch_data_from_api(url):
"""Function that connects to external API"""
try:
response = requests.get(url, timeout=5)
response.raise_for_status() # Will raise HTTPError on error
return response.json()
except requests.exceptions.Timeout:
raise ConnectionError("API not responding in time")
except requests.exceptions.HTTPError as e:
raise ValueError(f"API error: {e}")
except requests.exceptions.RequestException as e:
raise ConnectionError(f"Connection error: {e}")
# Tests with mocking
@patch('myapp.services.requests.get')
def test_api_timeout(mock_get):
"""Test timeout when connecting to API"""
mock_get.side_effect = requests.exceptions.Timeout()
with pytest.raises(ConnectionError, match="not responding in time"):
fetch_data_from_api("https://api.example.com/data")
@patch('myapp.services.requests.get')
def test_api_http_error(mock_get):
"""Test HTTP error from API"""
mock_response = Mock()
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError("404 Not Found")
mock_get.return_value = mock_response
with pytest.raises(ValueError, match="API error"):
fetch_data_from_api("https://api.example.com/data")
@patch('myapp.services.requests.get')
def test_api_success(mock_get):
"""Test successful API connection"""
mock_response = Mock()
mock_response.json.return_value = {"data": "test"}
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
result = fetch_data_from_api("https://api.example.com/data")
assert result == {"data": "test"}Exception Groups – nowość w Python 3.11+
Python 3.11 przyniósł Exception Groups – sposób na grupowanie wielu wyjątków. Super przydatne w programowaniu równoległym.
Podstawy Exception Groups
# Available from Python 3.11
def demo_exception_groups():
"""Demonstrate Exception Groups"""
# Create group of exceptions
errors = [
ValueError("Email field validation error"),
TypeError("Age field type error"),
KeyError("Name field missing")
]
# ExceptionGroup groups multiple exceptions
group = ExceptionGroup("Form validation errors", errors)
try:
raise group
except* ValueError as vg:
# New syntax - except* catches groups
print("Handling ValueError errors:")
for error in vg.exceptions:
print(f" - {error}")
except* TypeError as tg:
print("Handling TypeError errors:")
for error in tg.exceptions:
print(f" - {error}")
except* KeyError as kg:
print("Handling KeyError errors:")
for error in kg.exceptions:
print(f" - {error}")
# Example with asyncio and Exception Groups
import asyncio
async def task_with_errors(name, error_type):
"""Task that can raise different errors"""
await asyncio.sleep(0.1)
if error_type == "value":
raise ValueError(f"Value error in task {name}")
elif error_type == "type":
raise TypeError(f"Type error in task {name}")
else:
return f"Success in task {name}"
async def demo_async_exception_groups():
"""Demo Exception Groups with asyncio"""
tasks = [
task_with_errors("A", "value"),
task_with_errors("B", "type"),
task_with_errors("C", "success"),
task_with_errors("D", "value"),
]
# gather with return_exceptions=True
results = await asyncio.gather(*tasks, return_exceptions=True)
# Separate successes and errors
successes = [r for r in results if not isinstance(r, Exception)]
errors = [r for r in results if isinstance(r, Exception)]
print(f"Successes: {successes}")
if errors:
# Create Exception Group from errors
group = ExceptionGroup("Errors in async tasks", errors)
try:
raise group
except* ValueError as vg:
print(f"ValueError errors: {len(vg.exceptions)}")
for error in vg.exceptions:
print(f" - {error}")
except* TypeError as tg:
print(f"TypeError errors: {len(tg.exceptions)}")
for error in tg.exceptions:
print(f" - {error}")
# Run demo (requires Python 3.11+)
if __name__ == "__main__":
print("=== Demo Exception Groups ===")
demo_exception_groups()
print("\n=== Demo Async Exception Groups ===")
asyncio.run(demo_async_exception_groups())Praktyczne zastosowanie Exception Groups
import asyncio
from typing import List, Tuple
class FieldValidator:
"""Class for single field validation"""
def __init__(self, field_name: str):
self.field_name = field_name
def validate(self, value) -> List[Exception]:
"""Returns list of validation errors"""
errors = []
return errors
class EmailValidator(FieldValidator):
def validate(self, value) -> List[Exception]:
errors = []
if not value:
errors.append(ValueError(f"{self.field_name}: Email cannot be empty"))
elif "@" not in str(value):
errors.append(ValueError(f"{self.field_name}: Email must contain @"))
elif len(str(value)) > 100:
errors.append(ValueError(f"{self.field_name}: Email too long (max 100 chars)"))
return errors
class AgeValidator(FieldValidator):
def validate(self, value) -> List[Exception]:
errors = []
try:
age = int(value)
if age < 0:
errors.append(ValueError(f"{self.field_name}: Age cannot be negative"))
elif age > 150:
errors.append(ValueError(f"{self.field_name}: Age seems unrealistic"))
except (ValueError, TypeError):
errors.append(TypeError(f"{self.field_name}: Age must be a number"))
return errors
class FormValidator:
"""Form validator using Exception Groups"""
def __init__(self):
self.validators = {
'email': EmailValidator('email'),
'age': AgeValidator('age'),
}
def validate(self, data: dict):
"""Validates form and raises ExceptionGroup if errors exist"""
all_errors = []
for field_name, validator in self.validators.items():
value = data.get(field_name)
field_errors = validator.validate(value)
all_errors.extend(field_errors)
if all_errors:
raise ExceptionGroup("Form validation errors", all_errors)
return True
def demo_form_validator():
"""Demo validator using Exception Groups"""
validator = FormValidator()
# Test with errors
invalid_data = {
'email': 'invalid-email', # Missing @
'age': -5 # Negative age
}
try:
validator.validate(invalid_data)
except* ValueError as vg:
print("Value errors:")
for error in vg.exceptions:
print(f" - {error}")
except* TypeError as tg:
print("Type errors:")
for error in tg.exceptions:
print(f" - {error}")
# Test with valid data
valid_data = {
'email': '[email protected]',
'age': 30
}
try:
validator.validate(valid_data)
print("✓ Validation passed successfully!")
except ExceptionGroup as eg:
print(f"Validation failed: {eg}")
demo_form_validator()Najczęstsze błędy i antywzorce
Na koniec pokażę najczęstsze błędy, które widzę w kodzie produkcyjnym:
Antywzorzec #1: ignorowanie wyjątków
# BAD - worst possible code!
try:
critical_operation()
except:
pass # Error vanishes without trace
# BETTER but still bad
try:
critical_operation()
except Exception:
print("Something went wrong") # Too little info
# GOOD - proper handling
try:
critical_operation()
except SpecificError as e:
logger.error(f"Specific error: {e}", exc_info=True)
handle_specific_error(e)
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
raise # Re-raise if you don't know what to doAntywzorzec #2: zbyt szerokie łapanie wyjątków
# BAD - catches everything, even SystemExit and KeyboardInterrupt
try:
operation()
except: # Bare except - never do this!
handle_error()
# BETTER but still too broad
try:
operation()
except Exception: # Too general
handle_error()
# GOOD - specific types
try:
operation()
except ValueError as e:
handle_value_error(e)
except TypeError as e:
handle_type_error(e)
except MyCustomError as e:
handle_custom_error(e)Antywzorzec #3: wyjątki jako control flow
# BAD - using exceptions for control flow
def find_element_bad(items, condition):
try:
for item in items:
if condition(item):
raise StopIteration(item) # Wrong usage!
except StopIteration as e:
return e.value
return None
# GOOD - normal flow
def find_element_good(items, condition):
for item in items:
if condition(item):
return item
return NoneAntywzorzec #4: gubienie kontekstu błędu
# BAD - losing original error
try:
dangerous_operation()
except OriginalError:
raise NewError("Something went wrong") # Lost context!
# GOOD - preserving context
try:
dangerous_operation()
except OriginalError as e:
raise NewError("Something went wrong") from e # Preserved contextPodsumowanie i najważniejsze wnioski
Po trzech częściach serii o wyjątkach w Pythonie znacie już wszystkie najważniejsze techniki i wzorce. Oto kluczowe wnioski:
Co zapamiętać na zawsze
1. Wyjątki to normalna część programowania
– Nie oznaczają błędu programisty, to mechanizm komunikacji
– Python zachęca do ich używania (EAFP vs LBYL) – ale pamiętajcie, że styl Easier to Ask Forgiveness than Permission nie zawsze jest szybszy
– W Python 3.11+ obsługa wyjątków jest bardziej wydajna, bo zaimplementowano tzw. „zero‑cost exceptions” – blok try nie niesie kosztu, gdy nie wystąpi błąd
2. Hierarchia ma znaczenie
– Projektujcie własne wyjątki na podstawie tego, jak będą łapane
– SystemExit, KeyboardInterrupt i GeneratorExit celowo nie dziedziczą z Exception
– Używajcie konkretnych typów zamiast ogólnych
3. Asyncio wymaga szczególnej uwagi
– gather() przy domyślnym return_exceptions=False propaguje pierwszą napotkaną wyjątkową sytuację i nie anuluje automatycznie pozostałych zadań
– return_exceptions=True to wasz przyjaciel, jeśli chcecie zebrać wszystkie wyniki
– Timeouty są kluczowe w aplikacjach produkcyjnych
– Asynchroniczne kontekst menedżery działają podobnie do synchronicznych
4. Frameworki mają swoje wzorce
– Django: middleware do globalnej obsługi błędów i process_exception() zwracające HttpResponse
– Flask: dekoratory @app.errorhandler() i własne klasy wyjątków
– FastAPI: automatyczna walidacja Pydantic + custom exception handlers
5. Testowanie błędów jest obowiązkowe
– pytest.raises() dla podstawowych testów
– Parametryzacja dla testowania wielu przypadków
– Mocking dla testowania integracji z zewnętrznymi systemami
6. Python 3.11+ przynosi nowości
– Exception Groups dla grupowania błędów
– Składnia except* dla obsługi grup
– Dokładniejsze lokalizacje błędów w tracebackach
Praktyczne rekomendacje
Dla juniorów:
– Zacznijcie od podstawowej składni try/except/finally
– Uczcie się czytać traceback od dołu do góry
– Zawsze używajcie konkretnych typów wyjątków
– Logujcie błędy z kontekstem
Dla mid‑level:
– Twórzcie własne hierarchie wyjątków
– Używajcie context managers do zarządzania zasobami
– Implementujcie profesjonalne logowanie
– Integrujcie monitoring (Sentry, DataDog)
Dla seniorów:
– Projektujcie API błędów dla całych systemów
– Używajcie Exception Groups w Python 3.11+
– Optymalizujcie wydajność w krytycznych ścieżkach
– Edukujcie zespół o dobrych praktykach
Ostatnia rada: wyjątki to potężne narzędzie, ale pamiętajcie zasadę – z wielką mocą wiąże się wielka odpowiedzialność. Używajcie ich mądrze, dokumentujcie zachowania i zawsze myślcie o programistach, którzy będą czytać wasz kod w przyszłości.
To koniec serii o wyjątkach w Pythonie! Mam nadzieję, że te trzy części dały wam solidne podstawy do profesjonalnej pracy z błędami w Pythonie.
Powodzenia w kodowaniu i pamiętajcie – każdy błąd to okazja do nauki! 🐍
Spodobał Ci się post? Sprawdź pozostałe części serii lub podziel się nim!
Masz uwagi do posta, chcesz porozmawiać, szukasz pomocy z Pythonem i Django? Napisz do mnie!
Zapraszam do zadawania pytań przez formularz kontaktowy. Pamiętaj, że jeśli potrzebujesz wsparcia, możesz napisać do mnie - pomogę.
Dla początkujących programistów przygotowałem darmowego ebooka "WARSZTAT JUNIORA: Przewodnik po kluczowych kompetencjach i narzędziach dla początkującego programisty Pythona".