aiohttp服务端API错误处理框架及参数校验方案咨询

我们已经基于aiohttp开发了若干服务端API，这些API会调用一个承载全部业务逻辑的Python类。现需为所有API搭建错误处理框架，同时要实现请求参数校验，特此咨询框架的实现思路，以及参数校验应一次性返回所有错误还是逐个校验返回？

示例API代码如下：

async def new_user(request):
    try:
        # happy path where name is set
        user = request.query['name']
        # Process our new user
        print("Creating new user with name: ", user)
        response_obj = {
            'status' : 'success'
        }
        # return a success json response with status code 200 i.e. 'OK'
        return web.Response(text=json.dumps(response_obj), status=200)
    except Exception as e:
        # Bad path where name is not set
        response_obj = {
            'status' : 'failed',
            'reason': str(e),
            'code' : 400
        }
        # return failed with a status code of 500 i.e. 'Server Error'
        return web.Response(text=json.dumps(response_obj), status=400)

错误处理框架实现思路 #

我在aiohttp项目里做过类似的错误处理，分享下我的实践思路：

1. 用aiohttp中间件统一拦截异常 #

不要在每个API视图里写重复的try-except，而是用aiohttp的**中间件（middleware）**来全局捕获异常。这样所有API的错误处理逻辑都集中在一处，维护起来更方便，也避免了代码冗余。

2. 定义自定义异常类 #

把业务中可能出现的错误分类，比如客户端参数错误、资源不存在、权限不足、服务端内部错误等，分别定义对应的自定义异常类，让错误语义更清晰：

class APIError(Exception):
    def __init__(self, status_code: int, reason: str, error_code: str = None):
        self.status_code = status_code
        self.reason = reason
        self.error_code = error_code

class ValidationError(APIError):
    def __init__(self, reason: str, error_code: str = "VALIDATION_FAILED"):
        super().__init__(400, reason, error_code)

class ResourceNotFoundError(APIError):
    def __init__(self, reason: str, error_code: str = "RESOURCE_NOT_FOUND"):
        super().__init__(404, reason, error_code)

class InternalServerError(APIError):
    def __init__(self, reason: str = "Internal server error", error_code: str = "INTERNAL_ERROR"):
        super().__init__(500, reason, error_code)

3. 中间件捕获异常并返回统一格式响应 #

写一个全局异常处理中间件，捕获自定义异常和未预料到的异常，转换成统一的JSON响应格式：

import json
from aiohttp import web

@web.middleware
async def error_middleware(request, handler):
    try:
        return await handler(request)
    except APIError as e:
        response_obj = {
            "status": "failed",
            "reason": e.reason,
            "error_code": e.error_code
        }
        return web.Response(
            text=json.dumps(response_obj),
            status=e.status_code,
            content_type="application/json"
        )
    except Exception as e:
        # 捕获未定义的异常，返回通用500错误
        response_obj = {
            "status": "failed",
            "reason": "Internal server error",
            "error_code": "INTERNAL_ERROR"
        }
        return web.Response(
            text=json.dumps(response_obj),
            status=500,
            content_type="application/json"
        )

然后在创建aiohttp应用时注册这个中间件：

app = web.Application(middlewares=[error_middleware])
# 可以把业务逻辑类挂载到app实例，方便视图调用
app['user_service'] = UserService()

4. 业务逻辑层抛出自定义异常 #

在你的业务逻辑类里，遇到错误时直接抛出对应的自定义异常，不用处理响应，交给中间件统一处理：

class UserService:
    def create_new_user(self, name: str):
        if not name:
            raise ValidationError("User name cannot be empty")
        # 其他业务逻辑...

此时API视图可以大幅简化，不用写繁琐的异常捕获：

async def new_user(request):
    user_service = request.app['user_service']
    name = request.query.get('name')
    user_service.create_new_user(name)
    return web.Response(
        text=json.dumps({"status": "success"}),
        status=200,
        content_type="application/json"
    )

参数校验策略：一次性返回所有错误 #

我强烈推荐一次性返回所有参数错误，原因很简单：用户体验更好。如果逐个校验返回，用户需要多次提交请求才能修正所有问题，效率极低，很容易失去耐心。

用Pydantic实现批量参数校验 #

Pydantic是Python里非常好用的数据校验库，它能一次性收集所有参数错误，完美适配这个需求。

比如定义一个用户创建的参数模型：

from pydantic import BaseModel, Field, ValidationError

class CreateUserRequest(BaseModel):
    name: str = Field(..., min_length=2, max_length=50, description="User name (2-50 characters)")
    age: int = Field(..., ge=18, le=120, description="User age (18-120)")
    email: str = Field(..., description="Valid user email address")

然后在API视图里用这个模型校验请求参数，捕获ValidationError并转换成我们的自定义异常：

async def new_user(request):
    try:
        # 解析请求参数（这里以query参数为例，也可以是JSON Body）
        query_params = dict(request.query)
        # 用Pydantic模型校验所有参数
        validated_data = CreateUserRequest(**query_params)
    except ValidationError as e:
        # 把Pydantic的错误信息整理成友好的格式
        error_details = []
        for err in e.errors():
            error_details.append({
                "field": err["loc"][0],
                "message": err["msg"]
            })
        raise ValidationError(f"Invalid parameters: {error_details}")
    
    user_service = request.app['user_service']
    user_service.create_new_user(validated_data.name, validated_data.age, validated_data.email)
    
    return web.Response(
        text=json.dumps({"status": "success"}),
        status=200,
        content_type="application/json"
    )

这样当用户提交的参数有多个错误时（比如name太短、age小于18、email格式不对），会一次性返回所有错误信息，用户可以一次修正所有问题。

总结 #

• 错误处理：用aiohttp中间件+自定义异常类实现全局统一处理，避免重复代码，提高可维护性。
• 参数校验：用Pydantic实现批量校验，一次性返回所有错误，大幅提升用户体验。

内容的提问来源于stack exchange，提问作者venkat b