用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/agusmdev/burntop --skill fastapi-exceptions命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
OpenProse is a programming language for AI sessions. An AI session is a Turing-complete computer; OpenProse structures English into unambiguous control flow. More pattern than framework—it ships as a skill with no dependencies. Activate when: running .prose files, mentioning OpenProse, or orchestrating multi-agent workflows from a script. Use this skill if you ever want to kick off more than one subagent at a time, or orchestrate anything interesting between more than one subagent. Write a .prose file and save it in the .claude-plugin/ directory. Then embody the OpenProse VM, as described in prose.md, and execute it.
Configure Alembic for async SQLAlchemy migrations with PostgreSQL
Create FastAPI application factory with lifespan, middleware, pagination, and router configuration
基于 SOC 职业分类
| name | fastapi-exceptions |
| description | Create custom exceptions, error response schemas, and centralized exception handlers for FastAPI |
This skill covers creating a comprehensive exception handling system with custom exceptions, standardized error responses, and centralized exception handlers.
Create src/app/exceptions.py:
from typing import Any
from uuid import UUID
class AppException(Exception):
"""
Base exception for all application errors.
All custom exceptions should inherit from this class.
Provides consistent error structure across the application.
Attributes:
message: Human-readable error description
error_code: Machine-readable error code (e.g., "NOT_FOUND")
status_code: HTTP status code
details: Additional error context
"""
def __init__(
self,
message: str,
error_code: str,
status_code: int = 500,
details: dict[str, Any] | None = None,
):
self.message = message
.error_code = error_code
.status_code = status_code
.details = details {}
().__init__(.message)
():
():
:
message =
details = {: resource, : ()}
field :
message =
details = {: resource, : field, : (value)}
:
message =
details = {: resource}
().__init__(
message=message,
error_code=,
status_code=,
details=details,
)
():
():
default_message =
().__init__(
message=message default_message,
error_code=,
status_code=,
details={
: resource,
: field,
: (value),
},
)
():
():
error_details = details {}
field:
error_details[] = field
().__init__(
message=message,
error_code=,
status_code=,
details=error_details,
)
():
():
details = {}
resource:
details[] = resource
action:
details[] = action
().__init__(
message=message,
error_code=,
status_code=,
details=details,
)
():
():
().__init__(
message=message,
error_code=,
status_code=,
)
():
():
().__init__(
message=message,
error_code=,
status_code=,
details=details {},
)
():
():
().__init__(
message=message,
error_code=,
status_code=,
details=details {},
)
():
():
().__init__(
message=message ,
error_code=,
status_code=,
details={: service},
)
Create src/app/schemas/error.py:
from datetime import datetime
from typing import Any
from pydantic import BaseModel, Field
class ErrorResponse(BaseModel):
"""
Standardized error response schema.
All API errors return this structure for consistency.
"""
detail: str = Field(
...,
description="Human-readable error message",
examples=["Item with id '123' not found"],
)
error_code: str = Field(
...,
description="Machine-readable error code",
examples=["NOT_FOUND", "VALIDATION_ERROR", "CONFLICT"],
)
correlation_id: str = Field(
...,
description="Request correlation ID for tracing",
examples=["550e8400-e29b-41d4-a716-446655440000"],
)
timestamp: datetime = Field(
...,
description="When the error occurred (UTC)",
)
details: dict[str, Any] = Field(
default_factory=dict,
description="Additional error context",
examples=[{"resource": "Item", "id": "123"}],
)
class ValidationErrorDetail(BaseModel):
"""Detail for a single validation error."""
loc: list[str | int] = Field(
...,
description="Location of the error (field path)",
examples=[["body", "name"]],
)
msg: str = Field(
...,
description="Error message",
examples=["field required"],
)
type: str = Field(
...,
description="Error type",
examples=["value_error.missing"],
)
class ValidationErrorResponse(BaseModel):
"""
Response schema for Pydantic validation errors.
Maintains compatibility with FastAPI's default validation error format
while adding correlation_id and timestamp.
"""
detail: list[ValidationErrorDetail]
error_code: str = "VALIDATION_ERROR"
correlation_id: str
timestamp: datetime
Create src/app/exception_handlers.py:
import logging
from datetime import UTC, datetime
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import ValidationError as PydanticValidationError
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from app.exceptions import AppException, ConflictError, DatabaseError
from app.middleware.correlation_id import get_correlation_id
from app.schemas.error import ErrorResponse, ValidationErrorResponse
logger = logging.getLogger(__name__)
async def app_exception_handler(
request: Request,
exc: AppException,
) -> JSONResponse:
"""
Handle all AppException subclasses.
Converts application exceptions to standardized JSON responses.
"""
correlation_id = get_correlation_id()
# Log the error
logger.warning(
"Application error occurred",
extra={
"error_code": exc.error_code,
"status_code": exc.status_code,
"message": exc.message,
"details": exc.details,
"correlation_id": correlation_id,
"path": str(request.url.path),
},
)
error_response = ErrorResponse(
detail=exc.message,
error_code=exc.error_code,
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
details=exc.details,
)
return JSONResponse(
status_code=exc.status_code,
content=error_response.model_dump(mode="json"),
)
async def validation_exception_handler(
request: Request,
exc: RequestValidationError,
) -> JSONResponse:
"""
Handle Pydantic/FastAPI validation errors.
Converts validation errors to standardized format while
preserving the detailed error information.
"""
correlation_id = get_correlation_id()
logger.warning(
"Validation error",
extra={
"errors": exc.errors(),
"correlation_id": correlation_id,
"path": str(request.url.path),
},
)
error_response = ValidationErrorResponse(
detail=[
{
"loc": list(err["loc"]),
"msg": err["msg"],
"type": err["type"],
}
for err in exc.errors()
],
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=422,
content=error_response.model_dump(mode="json"),
)
async def pydantic_validation_exception_handler(
request: Request,
exc: PydanticValidationError,
) -> JSONResponse:
"""
Handle raw Pydantic validation errors (not from FastAPI).
"""
correlation_id = get_correlation_id()
error_response = ValidationErrorResponse(
detail=[
{
"loc": list(err["loc"]),
"msg": err["msg"],
"type": err["type"],
}
for err in exc.errors()
],
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=422,
content=error_response.model_dump(mode="json"),
)
async def integrity_error_handler(
request: Request,
exc: IntegrityError,
) -> JSONResponse:
"""
Handle SQLAlchemy IntegrityError (constraint violations).
Attempts to parse the error and return a user-friendly message.
"""
correlation_id = get_correlation_id()
logger.error(
"Database integrity error",
extra={
"error": str(exc.orig),
"correlation_id": correlation_id,
"path": str(request.url.path),
},
)
# Try to extract constraint name for better error message
error_str = str(exc.orig)
# Common patterns for PostgreSQL unique constraint violations
if "unique constraint" in error_str.lower():
error_response = ErrorResponse(
detail="A record with this value already exists",
error_code="CONFLICT",
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=409,
content=error_response.model_dump(mode="json"),
)
# Foreign key violation
if "foreign key constraint" in error_str.lower():
error_response = ErrorResponse(
detail="Referenced record does not exist",
error_code="VALIDATION_ERROR",
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=422,
content=error_response.model_dump(mode="json"),
)
# Generic database error
error_response = ErrorResponse(
detail="A database constraint was violated",
error_code="DATABASE_ERROR",
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=500,
content=error_response.model_dump(mode="json"),
)
async def sqlalchemy_error_handler(
request: Request,
exc: SQLAlchemyError,
) -> JSONResponse:
"""
Handle generic SQLAlchemy errors.
Logs the full error but returns a generic message to avoid
exposing database internals.
"""
correlation_id = get_correlation_id()
logger.error(
"Database error",
extra={
"error": str(exc),
"correlation_id": correlation_id,
"path": str(request.url.path),
},
exc_info=True,
)
error_response = ErrorResponse(
detail="A database error occurred",
error_code="DATABASE_ERROR",
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=500,
content=error_response.model_dump(mode="json"),
)
async def unhandled_exception_handler(
request: Request,
exc: Exception,
) -> JSONResponse:
"""
Catch-all handler for unhandled exceptions.
Logs the full exception but returns a generic error to the client.
"""
correlation_id = get_correlation_id()
logger.exception(
"Unhandled exception",
extra={
"correlation_id": correlation_id,
"path": str(request.url.path),
},
)
error_response = ErrorResponse(
detail="An unexpected error occurred",
error_code="INTERNAL_ERROR",
correlation_id=correlation_id,
timestamp=datetime.now(UTC),
)
return JSONResponse(
status_code=500,
content=error_response.model_dump(mode="json"),
)
def register_exception_handlers(app: FastAPI) -> None:
"""
Register all exception handlers with the FastAPI app.
Call this in your app factory:
register_exception_handlers(app)
"""
# Application exceptions
app.add_exception_handler(AppException, app_exception_handler)
# Validation exceptions
app.add_exception_handler(RequestValidationError, validation_exception_handler)
app.add_exception_handler(PydanticValidationError, pydantic_validation_exception_handler)
# Database exceptions
app.add_exception_handler(IntegrityError, integrity_error_handler)
app.add_exception_handler(SQLAlchemyError, sqlalchemy_error_handler)
# Catch-all (must be last)
app.add_exception_handler(Exception, unhandled_exception_handler)
from app.exceptions import NotFoundError, ConflictError, ValidationError
class ItemService:
async def get_by_id_or_raise(self, id: UUID) -> Item:
item = await self._repository.get_by_id(id)
if not item:
raise NotFoundError(resource="Item", id=id)
return item
async def create(self, obj_in: ItemCreate) -> Item:
existing = await self._repository.get_by_name(obj_in.name)
if existing:
raise ConflictError(
resource="Item",
field="name",
value=obj_in.name,
)
return await self._repository.create(obj_in)
async def activate(self, id: UUID) -> Item:
item = await self.get_by_id_or_raise(id)
if item.is_deleted:
raise ValidationError(
message="Cannot activate a deleted item",
field="deleted_at",
)
# ... activation logic
404 Not Found:
{
"detail": "Item with id '550e8400-e29b-41d4-a716-446655440000' not found",
"error_code": "NOT_FOUND",
"correlation_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"timestamp": "2025-01-05T12:00:00Z",
"details": {
"resource": "Item",
"id": "550e8400-e29b-41d4-a716-446655440000"
}
}
409 Conflict:
{
"detail": "Item with name='Widget' already exists",
"error_code": "CONFLICT",
"correlation_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"timestamp": "2025-01-05T12:00:00Z",
"details": {
"resource": "Item",
"field": "name",
"value": "Widget"
}
}
422 Validation Error:
{
"detail": [
{
"loc": ["body", "name"],
"msg": "String should have at least 1 character",
"type": "string_too_short"
}
],
"error_code": "VALIDATION_ERROR",
"correlation_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"timestamp": "2025-01-05T12:00:00Z"
}