Глобальные и локальные обработчики исключений, кастомные HTTP-ошибки, логирование ошибок, middleware для ошибок
Глобальные и локальные обработчики исключений, кастомные HTTP-ошибки, логирование
Starlite имеет собственную иерархию HTTP-исключений:
from starlite.exceptions import (
HTTPException,
ClientException,
ServerException,
ValidationException,
NotFoundException,
PermissionDeniedException,
InternalServerException,
ServiceUnavailableException,
)HTTPException (500)
├── ClientException (4xx)
│ ├── ValidationException (400)
│ ├── UnauthorizedException (401)
│ ├── PermissionDeniedException (403)
│ ├── NotFoundException (404)
│ └── ConflictException (409)
└── ServerException (5xx)
├── InternalServerException (500)
├── NotImplementedException (501)
└── ServiceUnavailableException (503)
from starlite import NotFoundException, get
@get("/users/{user_id:int}")
def get_user(user_id: int) -> User:
user = db.get_user(user_id)
if not user:
raise NotFoundException(detail=f"User {user_id} not found")
return userfrom starlite import ValidationException
@post("/users")
def create_user(data: UserCreate) -> User:
if db.user_exists(data.email):
raise ValidationException(
detail="Email already registered",
extra={"field": "email"}
)
...from starlite import PermissionDeniedException
@delete("/users/{user_id:int}")
def delete_user(
request: Request,
user_id: int,
current_user: User,
) -> None:
if current_user.id != user_id and not current_user.is_admin:
raise PermissionDeniedException(
detail="You can only delete your own account"
)
...from starlite import UnauthorizedException
def check_auth(request: Request):
if not request.session.get("user_id"):
raise UnauthorizedException(detail="Authentication required")Можно создавать собственные исключения:
from starlite.exceptions import HTTPException
class RateLimitExceeded(HTTPException):
status_code = 429
detail = "Rate limit exceeded"
def __init__(self, retry_after: int):
super().__init__(
headers={"Retry-After": str(retry_after)}
)
class PaymentRequired(HTTPException):
status_code = 402
detail = "Payment required to access this resource"Использование:
@get("/premium-content")
def get_premium_content(user: User) -> Content:
if not user.has_subscription:
raise PaymentRequired()
...
@get("/api/data")
def get_data(request: Request) -> dict:
if request.state.rate_limit_exceeded:
raise RateLimitExceeded(retry_after=60)
...Регистрируется на уровне приложения:
from starlite import Request, Response
from starlite.exceptions import HTTPException
from starlite.middleware import ExceptionResponseContent
async def http_exception_handler(
request: Request,
exc: HTTPException,
) -> Response:
return Response(
content=ExceptionResponseContent(
status_code=exc.status_code,
detail=exc.detail,
headers=exc.headers,
),
status_code=exc.status_code,
headers=exc.headers,
)
app = Starlite(
route_handlers=[...],
exception_handlers={
HTTPException: http_exception_handler,
},
)from starlite import exception_handler, Request, Response
@exception_handler(ValidationException)
async def validation_exception_handler(
request: Request,
exc: ValidationException,
) -> Response:
return Response(
content={
"error": "validation_failed",
"details": exc.extra,
},
status_code=400,
)
app = Starlite(
route_handlers=[...],
exception_handlers={
ValidationException: validation_exception_handler,
},
)from starlite.exceptions import NotFoundException, PermissionDeniedException
async def client_error_handler(
request: Request,
exc: NotFoundException | PermissionDeniedException,
) -> Response:
return Response(
content={"error": str(exc)},
status_code=exc.status_code,
)
app = Starlite(
exception_handlers={
NotFoundException: client_error_handler,
PermissionDeniedException: client_error_handler,
}
)from starlite.exceptions import InternalServerException
async def global_exception_handler(
request: Request,
exc: Exception,
) -> Response:
# Логирование ошибки
logger.exception(f"Unhandled exception: {exc}")
return Response(
content=ExceptionResponseContent(
status_code=500,
detail="Internal server error",
),
status_code=500,
)
app = Starlite(
exception_handlers={
Exception: global_exception_handler,
}
)from starlite import Controller, exception_handler
class ItemController(Controller):
path = "/items"
@exception_handler(ItemNotFoundError)
async def handle_item_not_found(
self,
request: Request,
exc: ItemNotFoundError,
) -> Response:
return Response(
content={"error": "item_not_found", "message": str(exc)},
status_code=404,
)
@get("/{item_id:int}")
def get_item(self, item_id: int) -> Item:
item = db.get_item(item_id)
if not item:
raise ItemNotFoundError(f"Item {item_id} not found")
return itemapi_router = Router(
path="/api",
route_handlers=[...],
exception_handlers={
ApiError: api_error_handler,
},
)import logging
from starlite import Request, Response
from starlite.exceptions import HTTPException
logger = logging.getLogger(__name__)
async def logged_exception_handler(
request: Request,
exc: HTTPException,
) -> Response:
logger.warning(
"HTTP error: %s %s -> %d",
request.method,
request.url,
exc.status_code,
extra={
"method": request.method,
"url": str(request.url),
"status_code": exc.status_code,
}
)
return Response(
content={"error": exc.detail},
status_code=exc.status_code,
)import sentry_sdk
from starlite import Request, Response
async def sentry_exception_handler(
request: Request,
exc: Exception,
) -> Response:
# Отправка в Sentry
sentry_sdk.capture_exception(exc, scope={
"request": {
"method": request.method,
"url": str(request.url),
"headers": dict(request.headers),
}
})
return Response(
content={"error": "Internal server error"},
status_code=500,
)
app = Starlite(
exception_handlers={
Exception: sentry_exception_handler,
}
)Middleware может перехватывать ошибки до exception handlers:
from starlite import Middleware
from starlite.types import ASGIApp, Receive, Scope, Send
class ErrorLoggingMiddleware:
def __init__(self, app: ASGIApp):
self.app = app
async def __call__(
self,
scope: Scope,
receive: Receive,
send: Send,
) -> None:
try:
await self.app(scope, receive, send)
except Exception as exc:
logger.exception("Error in middleware")
raise # Пробрасываем дальше для exception handler
app = Starlite(
route_handlers=[...],
middleware=[ErrorLoggingMiddleware],
)from starlite.middleware import ExceptionResponseContent
content = ExceptionResponseContent(
status_code=404,
detail="User not found",
extra={"user_id": 123},
)
# Результат:
# {
# "status_code": 404,
# "detail": "User not found",
# "extra": {"user_id": 123}
# }async def custom_exception_handler(
request: Request,
exc: HTTPException,
) -> Response:
return Response(
content={
"success": False,
"error": {
"code": f"ERR_{exc.status_code}",
"message": exc.detail,
"path": request.url.path,
},
},
status_code=exc.status_code,
)# ✅ Хорошо
raise NotFoundException(detail="User not found")
raise PermissionDeniedException(detail="Access denied")
# ❌ Плохо
raise HTTPException(status_code=404, detail="User not found")# ✅ Хорошо
raise ValidationException(
detail="Invalid email format",
extra={"field": "email", "value": email}
)
# ❌ Плохо
raise ValidationException(detail="Invalid input")# ✅ Хорошо
async def exception_handler(request, exc):
logger.exception("Unhandled error")
return Response(content={"error": "Internal error"}, status_code=500)
# ❌ Плохо — тишина в логах
async def exception_handler(request, exc):
return Response(content={"error": "Error"}, status_code=500)# ✅ Хорошо
raise InternalServerException(detail="Internal server error")
# ❌ Плохо — утечка информации
raise InternalServerException(
detail=f"Database connection failed: {db_password}"
)# ✅ Хорошо
client_handler = create_handler(400)
server_handler = create_handler(500)
app = Starlite(
exception_handlers={
ClientException: client_handler,
ServerException: server_handler,
}
)
# ❌ Плохо — дублирование
app = Starlite(
exception_handlers={
NotFoundException: handler1,
PermissionDeniedException: handler2,
ValidationException: handler3,
...
}
)Обработка ошибок в Starlite предлагает:
В следующей теме мы изучим WebSocket и Server-Sent Events.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.