أفضل ممارسات معالجة أخطاء Python 2026: الاستثناءات والتسجيل وإعادة المحاولة

# Well-designed exception hierarchy for a FastAPI app
class AppError(Exception):
    def __init__(self, message: str, code: str = "UNKNOWN_ERROR"):
        super().__init__(message)
        self.message = message
        self.code = code

class ValidationError(AppError):
    def __init__(self, field: str, message: str):
        super().__init__(f"Validation error on '{field}': {message}", "VALIDATION_ERROR")
        self.field = field

class NotFoundError(AppError):
    def __init__(self, resource: str, id: str | int):
        super().__init__(f"{resource} not found: {id}", "NOT_FOUND")
        self.resource = resource
        self.resource_id = id

class AuthorizationError(AppError):
    def __init__(self, action: str = "perform this action"):
        super().__init__(f"Not authorized to {action}", "UNAUTHORIZED")

class DatabaseError(AppError):
    def __init__(self, operation: str, original: Exception):
        super().__init__(f"Database error during {operation}: {original}", "DATABASE_ERROR")
        self.original = original
        self.__cause__ = original  # sets __cause__ for proper chaining

# Usage
raise NotFoundError("User", user_id)
raise ValidationError("email", "must be a valid email address")

# Exception chaining (raise from)
try:
    await db.users.get(user_id)
except asyncpg.PostgresError as e:
    raise DatabaseError("user lookup", e) from e

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    status_map = {
        "VALIDATION_ERROR": 422,
        "NOT_FOUND": 404,
        "UNAUTHORIZED": 403,
        "DATABASE_ERROR": 500,
    }
    status = status_map.get(exc.code, 500)
    return JSONResponse(
        status_code=status,
        content={
            "error": exc.code,
            "message": exc.message,
            "path": str(request.url),
        }
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse:
    import logging
    logging.error(f"Unhandled error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": "INTERNAL_ERROR", "message": "An unexpected error occurred"}
    )

import asyncio

async def validate_all_fields(data: dict) -> dict:
    errors = []

    if not data.get("email"):
        errors.append(ValidationError("email", "required"))
    elif "@" not in data["email"]:
        errors.append(ValidationError("email", "invalid format"))

    if not data.get("name"):
        errors.append(ValidationError("name", "required"))

    if data.get("age") and (data["age"] < 13 or data["age"] > 120):
        errors.append(ValidationError("age", "must be between 13 and 120"))

    if errors:
        raise ExceptionGroup("Validation failed", errors)

    return data

# Handle exception groups
async def create_user(data: dict):
    try:
        validated = await validate_all_fields(data)
        return await db.users.create(validated)
    except* ValidationError as eg:
        # Handle only ValidationErrors from the group
        field_errors = {e.field: e.message for e in eg.exceptions}
        return {"errors": field_errors}
    except* DatabaseError as eg:
        for e in eg.exceptions:
            logger.error(f"DB error: {e}")
        raise

import asyncio
import functools
import logging
from typing import Type

logger = logging.getLogger(__name__)

def async_retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple[Type[Exception], ...] = (Exception,),
    log_errors: bool = True,
):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except asyncio.CancelledError:
                    raise  # never retry cancellation
                except exceptions as e:
                    last_error = e
                    if attempt == max_attempts:
                        break
                    wait = delay * (backoff ** (attempt - 1))
                    if log_errors:
                        logger.warning(
                            f"{func.__name__} attempt {attempt}/{max_attempts} failed: {e}. "
                            f"Retrying in {wait:.1f}s"
                        )
                    await asyncio.sleep(wait)
            raise last_error
        return wrapper
    return decorator

@async_retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
async def fetch_external_api(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=10)
        r.raise_for_status()
        return r.json()

import structlog
import uuid
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="")

# Configure structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ]
)

logger = structlog.get_logger()

async def process_order(order_id: str, user_id: str):
    # Bind context that appears in all logs from this request
    structlog.contextvars.bind_contextvars(
        request_id=str(uuid.uuid4()),
        order_id=order_id,
        user_id=user_id,
    )

    try:
        order = await db.orders.get(order_id)
        logger.info("order.fetched", amount=order["total"])

        payment = await payment_service.charge(order)
        logger.info("payment.success", transaction_id=payment["id"])

        return payment
    except PaymentError as e:
        logger.error("payment.failed", error_code=e.code, reason=str(e))
        raise
    except Exception as e:
        logger.exception("order.unexpected_error")
        raise

from dataclasses import dataclass
from typing import TypeVar, Generic

T = TypeVar("T")
E = TypeVar("E", bound=Exception)

@dataclass
class Ok(Generic[T]):
    value: T
    success: bool = True

@dataclass
class Err(Generic[E]):
    error: E
    success: bool = False

Result = Ok[T] | Err[E]

async def safe_fetch_user(user_id: int) -> Result:
    try:
        user = await db.users.get(user_id)
        if not user:
            return Err(NotFoundError("User", user_id))
        return Ok(user)
    except DatabaseError as e:
        return Err(e)

# Usage - explicit handling, no try/except at call site
result = await safe_fetch_user(user_id)
if result.success:
    process(result.value)
else:
    handle_error(result.error)