非请求处理器场景下如何获取aiohttp当前请求上下文

我之前也碰到过类似的需求——让aiohttp接口根据客户端的Accept请求头自动返回对应格式的响应，用装饰器来处理确实是个优雅的方案。其实核心点就是让装饰器能拿到视图函数的request参数，因为aiohttp的视图函数本身就会接收request作为第一个参数，我们只需要在装饰器的包装逻辑里把它接住就行。

下面给你一套完整的实现思路和代码示例：

1. 实现格式转换装饰器 #

这个装饰器的作用是：

• 接收视图函数返回的原始数据（比如字典、列表）
• 从request中读取Accept请求头
• 根据指定格式把数据转换成对应的响应内容
• 返回符合aiohttp要求的web.Response对象

from aiohttp import web
import json
import csv
from io import StringIO

def format_response():
    def decorator(view_func):
        async def wrapped(request, *args, **kwargs):
            # 先调用原视图函数，拿到原始业务数据
            data = await view_func(request, *args, **kwargs)
            
            # 获取请求头的Accept字段，默认返回JSON格式
            accept_header = request.headers.get('Accept', 'application/json')
            
            # 根据不同格式处理响应
            if accept_header == 'application/json':
                return web.json_response(data)
            elif accept_header == 'text/csv':
                # 处理CSV格式（假设data是列表字典结构）
                if not data:
                    return web.Response(text='', content_type='text/csv')
                output = StringIO()
                writer = csv.DictWriter(output, fieldnames=data[0].keys())
                writer.writeheader()
                writer.writerows(data)
                return web.Response(text=output.getvalue(), content_type='text/csv')
            elif accept_header == 'text/html':
                # 简单生成HTML表格（复杂场景可以用模板引擎）
                if isinstance(data, list):
                    html = '<table border="1">'
                    # 渲染表头
                    html += '<tr>' + ''.join([f'<th>{key}</th>' for key in data[0].keys()]) + '</tr>'
                    # 渲染内容行
                    for item in data:
                        html += '<tr>' + ''.join([f'<td>{value}</td>' for value in item.values()]) + '</tr>'
                    html += '</table>'
                else:
                    html = f'<h1>Response</h1><p>{json.dumps(data)}</p>'
                return web.Response(text=html, content_type='text/html')
            else:
                # 不支持的格式返回406错误
                return web.Response(
                    text='Unsupported media type',
                    status=406,
                    content_type='text/plain'
                )
        return wrapped
    return decorator

2. 装饰你的视图函数 #

现在你只需要在视图函数上加上这个装饰器，视图函数只需要专注于返回原始业务数据，不用管响应格式：

@format_response()
async def get_users(request):
    # 模拟从数据库获取用户数据
    users = [
        {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
        {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
    ]
    return users

# 注册路由并启动应用
app = web.Application()
app.add_routes([web.get('/users', get_users)])

if __name__ == '__main__':
    web.run_app(app)

3. 测试不同格式的请求 #

你可以用curl或者Postman来测试不同格式的响应：

• 请求JSON格式：curl -H "Accept: application/json" http://localhost:8080/users
• 请求CSV格式：curl -H "Accept: text/csv" http://localhost:8080/users
• 请求HTML格式：curl -H "Accept: text/html" http://localhost:8080/users

一些额外的优化建议 #

• 支持模糊匹配Accept头：比如客户端发送Accept: application/json;q=0.9, text/csv;q=0.8，可以用aiohttp的web.Request.accept属性来解析权重，选择优先级最高的格式。
• 模板渲染HTML：如果HTML格式复杂，可以用aiohttp-jinja2这类模板引擎，在装饰器里调用模板渲染方法。
• 抽象格式转换逻辑：把不同格式的转换函数抽离成单独的模块，比如json_renderer、csv_renderer，让装饰器的代码更简洁易维护。

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