Task cannot generate pydantic-core schema for methods with TaskiqDepends

[kai-nashi]

taskiq == 0.11.18

import asyncio
from typing import Annotated

from taskiq import Context
from taskiq import TaskiqDepends
from taskiq import InMemoryBroker

broker = InMemoryBroker()


@broker.task("my_task")
async def my_task(context: Annotated[Context, TaskiqDepends()], x: int) -> None:
    print(f'{x=}')


if __name__ == "__main__":
    asyncio.run(my_task.kiq(1))

Expected:

$ x=1

Real:

pydantic.errors.PydanticSchemaGenerationError: Unable to generate pydantic-core schema for <class 'taskiq.context.Context'>. Set `arbitrary_types_allowed=True` in the model_config to ignore this error or implement `__get_pydantic_core_schema__` on your type to fully support it.

BUT:

@broker.task("my_task")
async def my_task(x: int, context: Annotated[Context, TaskiqDepends()]) -> None:
    print(f'{x=}')

Will work as expected.

Problem: When parsing params in https://github.com/taskiq-python/taskiq/blob/master/taskiq/receiver/params_parser.py#L70 Dependency might be ignored.

[kai-nashi]

Dependency is broken and unusable. How tests passed?

@broker.task("my_task")
async def my_task(*values, context: Context = TaskiqDepends(), **kwargs) -> None:
    print(values, context, kwargs)


if __name__ == "__main__":
    asyncio.run(my_task.kiq(1, 2, 3, z=4))

Same broken.

[kai-nashi]

Wow. It's much more broken i think.

import asyncio
from typing import Annotated

from taskiq import Context
from taskiq import TaskiqDepends
from taskiq import InMemoryBroker

broker = InMemoryBroker()


@broker.task("my_task")
async def my_task(context: Context = TaskiqDepends()) -> None:
    print(context)


if __name__ == "__main__":
    asyncio.run(my_task())

In this example context is instance of TaskiqDepends, not Context.

Any real reason to use DI besides following trends? Untested complexity is the fastest way to kill a project.

[s3rius]

The way you declare a function is incorrect. It's a syntax error. You define arg parameter after kwarg parameter.

It's prohibited by python itself.

import asyncio
from typing import Annotated

from taskiq import Context
from taskiq import TaskiqDepends
from taskiq import InMemoryBroker

broker = InMemoryBroker()


@broker.task("my_task")
async def my_task(x: int, context: Annotated[Context, TaskiqDepends()]) -> None:
    print(f"{x=}")


if __name__ == "__main__":
    asyncio.run(my_task.kiq(1))

This works, because it's correct python syntax. You can test out on this example:

def test(a : int = 1, b: int):
    pass

If you will try to execute it, then you will get the following error:

❯ python a.py
  File "/home/s3rius/ttest/a.py", line 1
    def test(a : int = 1, b: int):
                          ^^^^^^
SyntaxError: parameter without a default follows parameter with a default

Dependency injection is heavily tested though. You can check all test cases here:

https://github.com/taskiq-python/taskiq-dependencies/tree/master/tests

[kai-nashi]

@s3rius, maybe you are right — without a dedicated PEP, DI in Python is still a grey zone.

However, I still disagree with the idea of not resolving dependencies in a direct call. It should work as expected: we expect a Context from DI and must receive a Context from DI.

Right now, AsyncTaskiqDecoratedTask.call doesn’t implement any logic for this — and that’s not great.