Faust webview配置CORS解决跨域报错的实现方法

前置依赖安装 #

首先安装官方文档提到的aiohttp-cors库：

pip install aiohttp-cors

配置实现逻辑 #

Faust的Web服务底层基于aiohttp实现，CORS配置不能直接在视图类里零散添加，需要通过Faust提供的web初始化钩子，在服务启动时把CORS规则挂载到底层aiohttp应用实例上。

完整修改后代码 #

import faust
from faust.web import Request, Response, View
import aiohttp_cors
from datetime import datetime
from models.models_2 import AlarmFilterRequest

BROKER_LIST=['kafka://localhost:9092', 'kafka://localhost:9093', 'kafka://localhost:9094']

app = faust.App('alarm-filter-stream', broker=BROKER_LIST, store='memory://')

alarm_filter_requests_topic = app.topic('alarm-filter-requests', value_type=AlarmFilterRequest, value_serializer='json')

# 服务初始化时配置CORS
@app.on_webserver_init.connect
def configure_cors(app, web_app):
    # 配置CORS规则，开发环境可先放开所有限制，上线请替换为具体允许的源
    cors = aiohttp_cors.setup(web_app, defaults={
        "*": aiohttp_cors.ResourceOptions(
            allow_credentials=True,
            expose_headers="*",
            allow_headers="*",
            allow_methods=["GET", "POST", "OPTIONS"]
        )
    })
    # 给所有已注册的路由自动添加CORS规则，不用逐个视图配置
    for route in web_app.router.routes():
        cors.add(route)

@app.agent(alarm_filter_requests_topic)
async def new_filter_request(stream):
  async for event in stream.events():
    # 原有业务处理逻辑
    print(event.value)

@app.page('/filter')
class test_page_view(View):  
  async def post(self, request: Request) -> Response:
    # 解析请求体
    req_data = await request.json()
    filter_req = AlarmFilterRequest(**req_data)
    # 往Kafka主题提交数据
    await alarm_filter_requests_topic.send(value=filter_req)
    return self.json({"timestamp":f"{datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S.%f')}"})

配置说明 #

• 开发阶段可以直接用示例里的"*"通配符允许所有源访问，上线时需要替换为实际允许的前端域名，比如"http://your-frontend-domain.com"，避免安全风险
• allow_methods可以根据接口实际支持的请求方法调整，POST接口必须放行OPTIONS方法，否则浏览器的预检请求会失败
• 示例里的自动遍历路由添加CORS规则的写法，不用给每个视图单独写配置，新增页面路由也会自动生效
• 原示例代码漏了datetime模块的导入，直接运行会报错，修改版已经补上了对应导入

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