如何在 Flask 应用程序中使用依赖注入器正确实例化类？

Question

在我的 Flask 应用程序中，我正在使用依赖注入，这是我的应用程序的样子。我有一个使用 S3 作为数据存储的服务，我正在尝试使用注入的服务（通过 S3 客户端注入）来实例化我的应用程序。然而，S3 客户端似乎没有正确实例化，或者我正在做一些完全不同的事情。

containers.py

class Container(containers.DeclarativeContainer):

    wiring_config = containers.WiringConfiguration(modules=[".routes", ".scheduler"])

    config = providers.Configuration()
    s3_config = dict()
    s3_config["path"], s3_config["filters"] = config.get("s3_bucket"), [("member_status_nm", "=", "ACTIVE")]
    
    s3_repository = providers.Singleton(S3Repository, s3_config)

    my_service = providers.Factory(
        MyService, config, S3Repository
    )

这是我的

S3Repository

：

import logging
import sys
import time

import some_library as lib

class S3Repository:
    def __init__(self, s3_config):
        self.path, self.columns, self.filters = \
            s3_config.get("path", ""), s3_config.get("columns", []), s3_config.get("filters", [])

    def fetch(self):
        # execute fetch
        result = lib.some_fetch_method(self.path, self.columns, self.filters)

        return result

和

MyService

：

import #all relevant imports here

class MyService:
    
    def __init__(self, config: dict, s3_repository: S3Repository) -> None:
        
        logging.info("HealthSignalService(): initializing")

        self.config = config["app"]["health_signal_service"]

        
        # prepare s3_repository for the service
        self.s3_repository = s3_repository
        self.s3_repository.columns, self.s3_repository.filters, self.s3_repository.path = \
            ["x","y"], ["x1","y1"], "file_path"

        
    def fetch_data(self) -> None:

        try:
            summary_result = self.s3_repository.fetch()
        except (FileNotFoundError, IOError) as e:
            
           print("failure")
        return summary_result
   
   def get_data(memberId):
        sth = self.fetch_data()
        return sth.get(memberId)

最后把它绑在我的

routes.py

：

@inject
@auth.login_required
def get_signals(
        my_service: MyService = Provide[
            Container.my_service
        ],
):
    content = request.json
    member_id = content["memberId"]
    result = my_service.get_signals(member_id)
    return jsonify(result)

当我点击 API 端点时，我看到此错误：

summary_result = self.s3_repository.fetch()
TypeError: fetch() missing 1 required positional argument: 'self'

如何在使用依赖注入时正确初始化 S3 客户端？

Answer 1

从本文中提取的简单解决方案。 https://snyk.io/blog/dependency-injection-python/ ./app.py

pip install Flask-Injector

from flask import Flask, jsonify
from flask_injector import FlaskInjector, singleton
from injector import Injector, inject

app = Flask(__name__)

class Database:
    def __init__(self):
        self.data = {"message": "Data from the Database!"}

class Logger:
    def log(self, message):
        print("Logging:", message)

class DataService:
    def __init__(self, database: Database, logger: Logger):
        self.database = database
        self.logger = logger

    def fetch_data(self):
        self.logger.log("Fetching data...")
        return self.database.data

@app.route('/')
@inject
def index(data_service: DataService):
    return jsonify(data_service.fetch_data())

# Dependency configurations
def configure(binder):
    binder.bind(Database, to=Database(), scope=singleton)
    binder.bind(Logger, to=Logger(), scope=singleton)
    binder.bind(DataService, to=DataService(Database(), Logger()), scope=singleton)

if __name__ == '__main__':
    FlaskInjector(app=app, modules=[configure])
    app.run(debug=True)

调用存储的实例

a = Injector(configure).get(Database)

这个配置你可以把它放在其他文件中并将其作为模块导入