You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
dep_graph and __getattr__ dependencies implementation #35
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"
)
"""
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`
Activity
miragecentury commentedon Sep 15, 2025
Maybe switch to first condition to this instead :
if inspect.getattr_static(executed_func, "dep_graph", False):