Repository navigation
Unable to serialize unknown type: <class 'taskiq.task.AsyncTaskiqTask'> #587
Description
Taskiq version
0.11.14
Python version
Python 3.13
OS
Linux
What happened?
After startup a worker with the result backend as RedisAsyncResultBackend the application receives an exception: PydanticSerializationError
Relevant log output
2026-02-03 15:00:00,067 - taskiq.receiver.receiver - ERROR - Can't set result in result backend. Cause: Unable to serialize unknown type: <class 'taskiq.task.AsyncTaskiqTask'>
2026-02-03 17:00:00.069
Traceback (most recent call last):
2026-02-03 17:00:00.069
File "/usr/local/lib/python3.13/site-packages/taskiq/receiver/receiver.py", line 165, in callback
2026-02-03 17:00:00.069
await self.broker.result_backend.set_result(taskiq_msg.task_id, result)
2026-02-03 17:00:00.069
File "/usr/local/lib/python3.13/site-packages/taskiq_redis/redis_backend.py", line 126, in set_result
2026-02-03 17:00:00.069
"value": self.serializer.dumpb(model_dump(result)),
2026-02-03 17:00:00.069
~~~~~~~~~~^^^^^^^^
2026-02-03 17:00:00.069
File "/usr/local/lib/python3.13/site-packages/taskiq/compat.py", line 31, in model_dump
2026-02-03 17:00:00.069
return instance.model_dump(mode="json")
2026-02-03 17:00:00.069
~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^
2026-02-03 17:00:00.069
File "/usr/local/lib/python3.13/site-packages/pydantic/main.py", line 464, in model_dump
2026-02-03 17:00:00.069
return self.__pydantic_serializer__.to_python(
2026-02-03 17:00:00.069
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
2026-02-03 17:00:00.069
self,
2026-02-03 17:00:00.069
^^^^^
2026-02-03 17:00:00.069
...<12 lines>...
2026-02-03 17:00:00.069
serialize_as_any=serialize_as_any,
2026-02-03 17:00:00.069
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-02-03 17:00:00.069
)
2026-02-03 17:00:00.069
^
2026-02-03 17:00:00.069
pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'taskiq.task.AsyncTaskiqTask'>Broker initialization code
from taskiq import PrometheusMiddleware, SimpleRetryMiddleware, TaskiqEvents, TaskiqScheduler, TaskiqState
from taskiq.schedule_sources import LabelScheduleSource
from taskiq_pipelines import PipelineMiddleware
from taskiq_redis import ListRedisScheduleSource, RedisAsyncResultBackend, RedisStreamBroker
result_backend = RedisAsyncResultBackend( # type: ignore
redis_url=settings.redis.result_dsn,
result_ex_time=settings.redis.CACHE_RESULT_EX_TIME,
keep_results=False
)
schedule_source = ListRedisScheduleSource(url=settings.redis.schedule_dsn)
worker = (
RedisStreamBroker(url=settings.redis.broker_dsn)
.with_result_backend(result_backend)
.with_middlewares(
SimpleRetryMiddleware(default_retry_count=5),
PipelineMiddleware(),
PrometheusMiddleware(server_addr="0.0.0.0", server_port=9000),
)
)
# And here's the scheduler that is used to query scheduled sources
scheduler = TaskiqScheduler(broker=worker, sources=[schedule_source, LabelScheduleSource(worker)])
@worker.on_event(TaskiqEvents.WORKER_STARTUP)
async def startup(state: TaskiqState) -> None:
redis_pool = aioredis.ConnectionPool.from_url(url=settings.redis.cache_dsn, decode_responses=True)
state.redis = aioredis.Redis(connection_pool=redis_pool)
@worker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
async def shutdown(state: TaskiqState) -> None:
try:
await state.redis.disconnect()
except AttributeError:
await state.redis.aclose()
Activity
Vansh-Sharma27 commentedon Jul 19, 2026
Hi @s3rius !
I traced #587 through the 0.11.14 path and reproduced the reported Pydantic 2 error with a serializing result backend. A task that starts a pipeline and returns
await pipeline.kiq()returns anAsyncTaskiqTask; the receiver places that handle inTaskiqResult, andRedisAsyncResultBackendcallsmodel_dump(..., mode="json")before its serializer or Redis I/O.This also appears connected to the scheduling approach suggested in #353, where
pipeline.kiqis registered as a task. One related finding is that currentmasterhas the separate bound-method registration issue tracked in #436, so I would keep that scope independent.My proposed direction is a small receiver-level policy that does not persist returned task handles, while leaving the final pipeline task result available under its own ID. I would add a serializing-backend regression test for a task that returns a nested Taskiq handle, plus integration coverage only if a cross-repository pipeline change is preferred. The reporter's task registration code is not included, so I cannot confirm that exact route.
Would that behavior and scope align with your expectations? If so, may I prepare a focused patch?