Azure Functions ASGI应用(Python v2)对接存储队列输出绑定的方案问询

问题解答 #

1. ASGI应用是否支持输出绑定？

是的，AsgiFunctionApp确实不支持直接为FastAPI路由添加Azure Functions输出绑定。这是因为AsgiFunctionApp本质是将整个FastAPI应用封装成一个单一的HTTP触发函数，所有请求都会被路由到FastAPI处理，无法在FastAPI路由级别单独应用Functions的绑定装饰器（比如@app.queue_output）——这些装饰器仅适用于原生Azure Functions函数。

2. 仅用Azure Functions标准方法对接存储队列的方案

下面提供两种符合要求的实现方式，无需额外SDK：

方案一：用原生HTTP触发函数包装FastAPI，同时添加输出绑定

将FastAPI作为HTTP触发函数的处理逻辑一部分，这样就能直接使用队列输出绑定。所有请求先通过Azure Functions的HTTP触发器，再转发给FastAPI，同时在需要的场景下触发队列输出。

# function_app.py
import azure.functions as func
from src.main import app as fastapi
from fastapi import Request

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

# 用通配路由接收所有请求，转发给FastAPI
@app.route(route="{*route}", methods=["GET", "POST", "PUT", "DELETE"])
@app.queue_output(arg_name="msg", queue_name="test-queue", connection="Storage")
async def fastapi_handler(req: func.HttpRequest, msg: func.Out[str]) -> func.HttpResponse:
    # 将Azure Functions请求转换为FastAPI兼容的Request对象
    fastapi_req = Request(
        method=req.method,
        url=req.url,
        headers=req.headers,
        body=await req.get_body()
    )
    
    # 调用FastAPI处理请求
    response = await fastapi(fastapi_req.scope, fastapi_req.receive, lambda send: None)
    
    # 转换FastAPI响应为Azure Functions HttpResponse
    func_response = func.HttpResponse(
        body=response.body,
        status_code=response.status_code,
        headers=dict(response.headers)
    )
    
    # 针对特定路由（如/test POST）添加队列消息
    if req.route_params.get("route") == "test" and req.method == "POST":
        payload = await req.get_json()
        msg.set(str(payload))
    
    return func_response

方案二：用蓝图创建带HTTP触发器的队列输出函数，从FastAPI路由调用

既然Azure Functions要求每个函数必须有触发器，我们可以给队列输出逻辑添加一个HTTP触发器，然后在FastAPI路由中通过HTTP请求调用这个函数：

蓝图文件（blueprint.py）

import azure.functions as func

bp01 = func.Blueprint()

@bp01.route(route="send-to-queue", methods=["POST"])
@bp01.queue_output(arg_name="output", queue_name="test-queue", connection="Storage")
async def send_message_to_queue(req: func.HttpRequest, output: func.Out[str]) -> func.HttpResponse:
    payload = await req.get_json()
    output.set(str(payload))
    return func.HttpResponse("Message queued", status_code=200)

主函数文件（function_app.py）

import azure.functions as func
import httpx
from src.main import app as fastapi

app = func.AsgiFunctionApp(app=fastapi, http_auth_level=func.AuthLevel.FUNCTION)
app.register_blueprint(bp01)

@fastapi.post("/test")
async def receive_post_and_forward_to_queue(payload: dict):
    # 调用内部的队列输出函数（部署后可通过环境变量获取函数域名）
    base_url = "http://localhost:7071/api" if func.is_local() else f"https://{os.environ['WEBSITE_HOSTNAME']}/api"
    async with httpx.AsyncClient() as client:
        await client.post(f"{base_url}/send-to-queue", json=payload)
    return {"status": "message forwarded to queue"}

注意：部署后需要用WEBSITE_HOSTNAME环境变量构造正确的函数URL，避免硬编码本地地址。

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