我们可以实例化多个顶级函数应用实例吗

问题描述 投票:0回答:1

我们正在尝试用 Python 创建一个单一的 Azure Function App,它结合了用于构建后端 API 的 FastAPI 框架和用于处理长时间运行的流程的 Azure Durable Functions。在下面的示例代码中,FastAPI 应用程序在第 7 行实例化。 该应用程序具有多个内部路由,并且按预期工作,提供稳定可靠的后端API。

我们还希望集成 Azure Durable Functions 以执行长时间运行的任务。如第 9 行所示,我们正在尝试实例化一个名为 myApp 的持久函数。

我们还想知道这样做的含义,因为持久函数可能会阻塞操作,而快速 api 主要是异步操作。

在此输入图片描述

我们尝试将它们分成两个功能应用程序,并且它工作正常,我们正在尝试检查是否可以在 Azure 中的单功能应用程序中运行。

python azure azure-functions fastapi azure-durable-functions
1个回答
0
投票

可以在 Azure Functions 中实例化多个顶级函数应用实例。每个实例都可以配置自己的一组功能、定价计划、部署方法和运行时版本。

关于将 FastAPI 和 Azure Durable Functions 组合在单个函数应用程序中的具体用例,是可以这样做的。

您将需要一个 main.py 来包含 FastAPI 应用程序实例(在我的例子中是应用程序)

from fastapi import FastAPI

  

app = FastAPI()

  

@app.get("/")

async  def read_root():

return {"message": "Hello from FastAPI"}

创建一个函数应用程序

import json

import azure.functions as func

import azure.durable_functions as DurableFunc

from app.main import app as fastapi_app

  

app = func.AsgiFunctionApp(app=fastapi_app, http_auth_level=func.AuthLevel.ANONYMOUS)

myApp = DurableFunc.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

  

@myApp.route("orchestrators/{functionName}")

@myApp.durable_client_input(client_name="client")

async  def http_start(req: func.HttpRequest, client):

function_name = req.route_params.get('functionName')

if req.get_body():

request_body = json.loads(req.get_body().decode())

else:

request_body = None

  

instance_id = await client.start_new(function_name, client_input=request_body)

response = client.create_check_status_response(req, instance_id)

return response

创建host.json和local.settings.json

主机.json

{
    "version": "2.0",
    "extensions": {
        "durableTask": {
            "hubName": "MyTaskHub"
        }
    }
}

本地设置

{
    "IsEncrypted": false,
    "Values": {
        "AzureWebJobsStorage": "UseDevelopmentStorage=true",
        "FUNCTIONS_WORKER_RUNTIME": "python"
    }
}

创建您的耐用功能

import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):
    result = yield context.call_activity('Hello', "world")
    return result

main = df.Orchestrator.create(orchestrator_function)

创建活动函数

import azure.functions as func

def main(name: str) -> str:
    return f"Hello {name}!"

你可以验证一下-

uvicorn main:app --reload

enter image description here

您现在可以将其部署到您的函数应用程序

func azure functionapp publish arkoFuncApp

参考资料:

© www.soinside.com 2019 - 2024. All rights reserved.