Flask 依赖注入

Question

我是 Flask 和 Python 的新手。我想实现一个依赖注入容器并访问不同模块内的依赖项。我的第一次尝试看起来像这样：

class AppModule(Module):
    def __init__(self, app):
        self.app = app

    """Configure the application."""
    def configure(self, binder):
        client = self.configure_cosmos_client()
        binder.bind(CosmosClient, to=client, scope=singleton)
        binder.bind(Dao, to=Dao, scope=singleton)

    def configure_cosmos_client(self) -> CosmosClient:
        return CosmosClient(
            url_connection=self.app.config.get('ENDPOINT'),
            auth={'masterKey': self.app.config.get('PRIMARYKEY')}
        )

app = Flask(__name__)

injector = Injector([AppModule(app)])
FlaskInjector(app=app, injector=injector)

app.run()

在模块内部，我想获得 CosmosClient 依赖项，例如：

class Dao:
    cosmos_client = None

    def __init__(self):
        self.cosmos_client = DI.get(CosmosClient)

有什么办法可以实现这一点吗？请注意“DI.get”只是一个示例，因为除了将依赖项注入到路由中之外，我找不到如何访问这些依赖项。

Answer 1

尝试依赖注入器。它有Flask 教程。你的容器看起来像这样：

from dependency_injector import containers, providers
from dependency_injector.ext import flask
from flask import Flask
from flask_bootstrap import Bootstrap
from github import Github

from . import views, services


class ApplicationContainer(containers.DeclarativeContainer):
    """Application container."""

    app = flask.Application(Flask, __name__)

    bootstrap = flask.Extension(Bootstrap)

    config = providers.Configuration()

    github_client = providers.Factory(
        Github,
        login_or_token=config.github.auth_token,
        timeout=config.github.request_timeout,
    )

    search_service = providers.Factory(
        services.SearchService,
        github_client=github_client,
    )

    index_view = flask.View(
        views.index,
        search_service=search_service,
        default_query=config.search.default_query,
        default_limit=config.search.default_limit,
    )

要运行该应用程序，您需要：

from .containers import ApplicationContainer


def create_app():
    """Create and return Flask application."""
    container = ApplicationContainer()
    container.config.from_yaml('config.yml')
    container.config.github.auth_token.from_env('GITHUB_TOKEN')

    app = container.app()
    app.container = container

    bootstrap = container.bootstrap()
    bootstrap.init_app(app)

    app.add_url_rule('/', view_func=container.index_view.as_view())

    return app

测试将如下所示：

from unittest import mock

import pytest
from github import Github
from flask import url_for

from .application import create_app


@pytest.fixture
def app():
    return create_app()


def test_index(client, app):
    github_client_mock = mock.Mock(spec=Github)
    # Configure mock

    with app.container.github_client.override(github_client_mock):
        response = client.get(url_for('index'))

    assert response.status_code == 200
    # Do more asserts

Answer 2

从注射器导入注射器，注射

cosmosClientInstance = Injector([AppModule(app)]).get(CosmosClient)

Answer 3

我不确定你到底想要实现什么，但不要以 Java 思维方式用 python 编写代码。

如果您尝试将

cosmos_client

添加到应用程序并从其他地方访问它，您可能希望将其作为 config attribute 保存到 Flask 应用程序中？