taskiq-dependencies icon indicating copy to clipboard operation
taskiq-dependencies copied to clipboard

dep_graph and __getattr__ dependencies implementation

Open miragecentury opened this issue 4 months ago • 1 comments

Hi, I am using FastAPI, Taskiq and PyMongo. And i am requesting some hints to help me resolve it.

I have an AsyncIOMotorDatabase as dependencies but it's implement getattr magic to manage collection_name resulting in an exception raise through the resolver.

    def __bool__(self) -> NoReturn:
            raise NotImplementedError(
            f"{type(self).__name__} objects do not implement truth "
            "value testing or bool(). Please compare "
            "with None instead: collection is not None"
        )

The issue is in the resolver method :

if getattr(executed_func, "dep_graph", False):

        """
        Sync resolver.

        This function is used to execute functions
        to resolve dependencies.

        :param executed_func: function to resolve.
        :param initial_cache: cache to build a context if graph was passed.
        :raises RuntimeError: if async function is passed as the dependency.

        :return: dict with resolved kwargs.
        """
        if getattr(executed_func, "dep_graph", False):
            ctx = SyncResolveContext(executed_func, self.main_graph, initial_cache)
            self.sub_contexts.append(ctx)
            sub_result = ctx.resolve_kwargs()
        elif inspect.isgenerator(executed_func):
            sub_result = next(executed_func)
            self.opened_dependencies.append(executed_func)
        elif asyncio.iscoroutine(executed_func):
            raise RuntimeError(
                "Coroutines cannot be used in sync context. "
                "Please use async context instead.",
            )
        elif iscontextmanager(executed_func):
            sub_result = executed_func.__enter__()
            self.opened_dependencies.append(executed_func)
        elif inspect.isasyncgen(executed_func) or isasynccontextmanager(executed_func):
            raise RuntimeError(
                "Coroutines cannot be used in sync context. "
                "Please use async context instead.",
            )
        else:
            sub_result = executed_func
        return sub_result`

miragecentury avatar Sep 15 '25 22:09 miragecentury

Maybe switch to first condition to this instead : if inspect.getattr_static(executed_func, "dep_graph", False):

miragecentury avatar Sep 15 '25 22:09 miragecentury