Search code examples
pythondependency-injectionsqlalchemyfastapiuvicorn

ValueError: callable <dependency_injector.providers.Singleton> is not supported by signature with MySQLDatabase


I'm encountering an issue with dependency_injector.providers.Singleton in my Python application when trying to instantiate MySQLDatabase. Here's a simplified version of my code:

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()

When running the above code, I encounter the following error:

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

Here is the my database setup code for your reference.

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()


I expected that dependency_injector.providers.Singleton would correctly instantiate MySQLDatabase with the provided configuration parameters from config.yaml without encountering the ValueError related to the callable not being supported by signature. The MySQLDatabase instance should be initialized and accessible as a singleton throughout my FastAPI application, allowing me to use it for database operations seamlessly.


Solution

  • If I got your question correctly, you can do it in the following way:

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

    Or inject it using the appropriate decorator exactly where you need it (just an example):

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