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