如何为azure函数python v2连接依赖注入

Question

我是一名 .net 开发人员，最近开始使用 python，我正在尝试使用 python 创建 Azure 函数 v2，并寻求有关如何将依赖项注入到我的函数中的帮助。我查看了 python 依赖注入的各种示例，并且一般能够做到这一点。但我不知道在哪里调用我的引导程序。

在.net中，这可以简单地通过扩展FunctionStarup类来完成，我在python azure-functions包中找不到这样的东西。 Microsoft 文档也没有在 python function v2programmingguidepython functionv2

中提到任何有关依赖注入的内容

这是我尝试过的示例的部分代码片段，但它不起作用。我在下面的例子中使用了 DI 的扭结

我正在尝试注入的类

       class IocService:
        def __init__(self, message: str):
            self.message = message
    
        def say_hello(self, name: str) -> str:
            return f"{self.message}, {name}!"

#使用蓝图定义函数

from kink import inject
import logging

from IocService import IocService

import azure.functions as func 

bp = func.Blueprint() 
@inject
@bp.route(route="default_template") 
def default_template(req: func.HttpRequest, context: func.Context, service: IocService) -> func.HttpResponse: 
    logging.info('Python HTTP trigger function processed a request.') 
    #ioc_service = context.get_service(IocService)

    name = req.params.get('name') 
    if not name: 
        try: 
            req_body = req.get_json() 
        except ValueError: 
            pass 
        else: 
            name = req_body.get('name') 

    if name: 
        return func.HttpResponse( 
            f"Hello, {name}. This HTTP-triggered function " 
            f"executed successfully.") 
    else: 
        return func.HttpResponse( 
            "This HTTP-triggered function executed successfully. " 
            "Pass a name in the query string or in the request body for a" 
            " personalized response.", 
            status_code=200 
        )

function_app.py

import azure.functions as func 
from http_blueprint import bp
from bootstrap import start_up

print("before start up")
start_up()
print("After start up")
app = func.FunctionApp() 
app.register_functions(bp)

以下是错误信息

>       File "<path>\function_app.py", line 2, in <module>
>     from http_blueprint import bp   File "<path>\http_blueprint.py", line 9, in <module>
>     @inject
>      ^^^^^^   File "<path>\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\kink\inject.py",
> line 203, in inject
>     return _decorator(_service)
>            ^^^^^^^^^^^^^^^^^^^^   File "<Path>\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\kink\inject.py",
> line 193, in _decorator
>     service_function = _decorate(bind or {}, _service, container)
>                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^   File
> "<Path>\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\kink\inject.py",
> line 105, in _decorate
>     service in [ABC.__init__, _no_init] or service.__name__ == "_no_init"
> 
>

https://learn.microsoft.com/en-us/azure/azure-functions/functions-reference-python?tabs=asgi%2Capplication-level&pivots=python-mode-decorators

当我通过从 httptrigger 方法中删除注入来运行该函数时，日志显示了非常有趣的事情：

它在函数中打印了日志
然后在function_app.py中打印日志

Answer 1

您可以直接在 Function Trigger 目录下的其他文件中添加依赖的类或方法，并在 Function App 中调用它。请参阅以下：-

我的

sidblueprint.py

:-

import azure.functions as func
import logging
from IocService import IocService



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


bp = func.Blueprint() 

@bp.route(route="HttpRequest") 
def main(req: func.HttpRequest) -> func.HttpResponse:
    message = "Hello from IocService"  # Define your message
    ioc_service = IocService(message)  # Instantiate the IocService

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        response = ioc_service.say_hello(name)
        return func.HttpResponse(response, status_code=200)
    else:
        return func.HttpResponse(
             "name with query string or body",
             status_code=400
        )

我的

function_app.py

:-

import azure.functions as func 
from sidblueprint import bp

app = func.FunctionApp() 

app.register_functions(bp)

我的

IocService.py

:-

class IocService:
    def __init__(self, message: str):
        self.message = message

    def say_hello(self, name: str) -> str:
        return f"{self.message}, {name}!"

输出：-

enter image description here

根据这个SO线程答案，您可以利用Python中的

dependency_injector

模块在Python脚本和函数中注入依赖注入。

我的

sidblueprint.py

与dependency_injector模块：-

import azure.functions as func
import logging
# from IocService import IocService

from dependency_injector import containers, providers

from dependency_injector.wiring import Provide, inject

class IocService:
    def __init__(self, message: str):
        self.message = message

    def say_hello(self, name: str) -> str:
        return f"{self.message}, {name}!"

class Container(containers.DeclarativeContainer):
    ioc_service = providers.Factory(
        IocService,
        message="Hello"  # Set your default message here
    )

container = Container()


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


bp = func.Blueprint() 
@inject
@bp.route(route="HttpRequest") 
def main(req: func.HttpRequest) -> func.HttpResponse:
    message = "Hello from IocService"  # Define your message
    ioc_service = IocService(message)  # Instantiate the IocService

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        response = ioc_service.say_hello(name)
        return func.HttpResponse(response, status_code=200)
    else:
        return func.HttpResponse(
             "name with query string or body",
             status_code=400
        )

输出：-

enter image description here

要求.txt：-

azure-functions
dependency_injector

Answer 2

您可以使用

dependency_injector

来注入依赖项。参考如下：

相关配套：

Azure 函数 - Python V2

依赖注入器：简介文档

项目布局

./
├── app/
│   ├── __init__.py
│   ├── services.py
│   ├── containers.py
│   └── hp_blueprint.py
├── venv/
├── function_app.py
└── requirements.txt

我的

hp_blueprint.py

：

在这些函数绑定触发器中，我们应该使用

Provide

来获取依赖服务而不是参数注入。如果使用依赖注入参数，我们会得到参数错误，需要在

function.json

中声明依赖服务参数。

在没有绑定触发器的函数中，我们应该使用

@inject

和注入参数来使用它们。查看更多详情：

import logging

import azure.functions as func

from dependency_injector.wiring import Provide

from app.containers import Container
from app.services import IocService

http_bp = func.Blueprint() 

@http_bp.route(route="test")
@http_bp.http_type("GET")
def testtrigger(
    req: func.HttpRequest) -> func.HttpResponse: 
    logging.info('Python HTTP trigger function processed a request.') 

    service: IocService = Provide[Container.iocservice]

    return func.HttpResponse( 
        body=service.test(),
        status_code= 200)

** 我的

services.py

和

container.py

:

参考文档：单容器-依赖注入器我们使用container来保存我们的依赖注入结构，包括Factory、Singleton、Configuration等。如果您想将容器拆分为多个模块或域，请参阅多个容器。

class IocService:
    def __init__(self):
        self.message = "Test"

    def test(self) -> str:
        return f"{self.message}"

from dependency_injector import containers, providers

from .services import IocService


class Container(containers.DeclarativeContainer):

    iocservice = providers.Factory(IocService)

容器注入

我们需要找到这个容器的使用范围。并在

__init__.py

中添加接线代码。您可以从

Wiring Document

和 Wire API 获取有关 wire

方法的更多详细信息

以下示例：

from .containers import Container
container = Container()

# init dependencies for Resource type.
container.init_resources()

container.wire(packages=["app"])

有两种连接使用范围的方法，包括modules和packages。我用包装线连接容器。容器递归地遍历包模块。这意味着

Container

可以用于

app.*

如何为azure函数python v2连接依赖注入

问题描述投票：0回答：2

2个回答

最新问题

如何为azure函数python v2连接依赖注入

问题描述 投票：0回答：2

2个回答

最新问题

问题描述投票：0回答：2