Dependency injection in background callbacks

i am using dependency_injector library to inject db connection. With normal callbacks the wiring is happening correctly and DB injection works. However, there seems to be a problem while using background callbacks.

I am instantiating background callbacks like below in app.py. However, using @inject decorator does not convert session_factory into dynamic session_factory object. Any guidance, where to do the wiring so that it works with background callbacks?

session_factory: db.AppContainer = Provide[db.AppContainer.session_factory]

if os.name == “nt”:
launch_uid = uuid4()
disk_cache = diskcache.Cache(“./cache”)
long_callback_manager = DiskcacheLongCallbackManager(disk_cache, cache_by=[lambda: launch_uid], expire=60)
background_callback_manager = DiskcacheManager(disk_cache)
else:
celery_app = Celery(
name, broker=“redis://localhost:6379/1”, backend=“redis://localhost:6379/2”, include=[‘callbacks’]
)
long_callback_manager = CeleryLongCallbackManager(celery_app)
background_callback_manager = CeleryManager(celery_app)

Did you ever find the solution to this problem?

I used a workaround by doing the wiring in the container itself like below
wiring_config = containers.WiringConfiguration(modules=[“XYZ”])

This allowed injection in the background callbacks.
Ideally it will be great to know where exactly should the wiring happen possibly in celery since this is what background callback is using i suppose.

Is the code open s.t. it’s possible to have a look at the entire container setup?

i may not be able to provide complete code has it has sensitive information. However, see if below redacted code serves the purpose

from dependency_injector import containers, providers
...

class AppContainer(containers.DeclarativeContainer):
    my_keepass = providers.Singleton(MyKeePassContainer)
    session_factory = providers.Singleton(DatabaseContainer)
    wiring_config = containers.WiringConfiguration(modules=["callbacks"])
	

class DatabaseContainer(containers.DeclarativeContainer):
    config = providers.Configuration()
    config_schema = providers.Configuration()
	schema_list = [...]
	
	db_ro_schema_selector = providers.Selector(
        config_schema.schema_name,
        **{
            schema_name: providers.Singleton(
                Database,
                schema_name=schema_name,
                db_conn_uri=providers.Singleton(
                    DbConnectionURI,
                    password_manager=providers.Callable(my_keepass.password_manager),
                    read_only=True
                )
            )
            for schema_name in schema_list
        },
        **{
            "mssql": providers.Singleton(
                MSSQLDatabase,
                password_manager=providers.Callable(my_keepass.password_manager)
            )
        }
    )

    db_rw_schema_selector = providers.Selector(
        config_schema.schema_name,
        **{
            schema_name: providers.Singleton(
                Database,
                schema_name=schema_name if schema_name not in [""] else 'public',
                db_conn_uri=providers.Singleton(
                    DbConnectionURI,
                    password_manager=providers.Callable(my_keepass.password_manager),
                    read_only=False
                )
            )
            for schema_name in schema_list
        },
        **{
            "mssql": providers.Singleton(
                MSSQLDatabase,
                password_manager=providers.Callable(my_keepass.password_manager)
            )
        }
    )

    db_conn_provider = providers.Selector(
        config.ro_or_rw,
        ro=db_ro_schema_selector,
        rw=db_rw_schema_selector
    )