ValueError：可调用<dependency_injector.providers.Singleton>不受MySQLDatabase签名支持

Question

在尝试实例化 MySQLDatabase 时，我的 Python 应用程序中遇到了 dependency_injector.providers.Singleton 的问题。这是我的代码的简化版本：

import logging
from dependency_injector import providers, containers
from app.repository.mysql.database import MySQLDatabase

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

class Containers(containers.DeclarativeContainer):
    config = providers.Configuration(yaml_files=["./conf/config.yaml"])

    mysql_db = providers.Singleton(
        MySQLDatabase,
        db_host=config.properties.database.mysql.host,
        db_port=config.properties.database.mysql.port,
        db_username=config.properties.database.mysql.user,
        db_password=config.properties.database.mysql.password,
        db_schema=config.properties.database.mysql.schema,
    )

# Attempting to access MySQLDatabase instance
db_instance = Containers.mysql_db()

运行上述代码时，遇到以下错误：

ValueError: callable <dependency_injector.providers.Singleton(<class 'app.repository.mysql.database.MySQLDatabase'>) at 0x1dc95622aa0> is not supported by signature

这是我的数据库设置代码供您参考。

import logging
import pymysql
from contextlib import contextmanager
from sqlalchemy import create_engine, orm
from sqlalchemy.orm import declarative_base

logger = logging.getLogger(__name__)

Base = declarative_base()

class MySQLDatabase:
    def __init__(self, db_host, db_port, db_username, db_password, db_schema):
        self.__db_host = db_host
        self.__db_port = db_port
        self.__db_username = db_username
        self.__db_password = db_password
        self.__db_schema = db_schema

        # Connection string for MySQL
        self.__db_connection_str = f"mysql+pymysql://{self.__db_username}:{self.__db_password}@{self.__db_host}:{self.__db_port}/{self.__db_schema}"

        logger.info(self.__db_connection_str)

        self._engine = create_engine(self.__db_connection_str)
        self._session_factory = orm.scoped_session(
            orm.sessionmaker(autocommit=False, autoflush=False, bind=self._engine)
        )

    def create_database(self):
        Base.metadata.create_all(self._engine)

    @contextmanager
    def session(self):
        session: orm.Session = self._session_factory()
        try:
            yield session
            session.commit()
        except Exception as ex:
            logger.exception("Session rollback because of exception: %s", ex)
            session.rollback()
            raise
        finally:
            session.close()

我期望 dependency_injector.providers.Singleton 能够使用 config.yaml 中提供的配置参数正确实例化 MySQLDatabase，而不会遇到与签名不支持的可调用相关的 ValueError。 MySQLDatabase 实例应该在我的 FastAPI 应用程序中作为单例进行初始化和访问，从而允许我无缝地使用它进行数据库操作。

Answer 1

如果我答对了你的问题，你可以按照以下方式进行操作：

container = Containers()
db = container.mysql_db()

或者在您需要的地方使用适当的装饰器将其注入（仅作为示例）：

@router.get(...)
@inject
async def get_data(db: MySQLDatabase = Depends(Provide[Containers.mysql_db])) -> ...:
    """Get data from db."""

    await db.do_some_stuff(...)