Cannot send task to the queue #34
Description
taskiq-postgres version
0.6.1
Python version
Python 3.11
PostgreSQL driver
asyncpg
PostgreSQL version
15.3
What happened?
Ошибка в подготовке SQL выражения (см. ошибку ниже)
Expected behavior
Задача записывается в таблицу taskiq_messages
Steps to reproduce
- Запустить scheduler
- Получить ошибку в логах
Minimal code example
Relevant log output
[2025-12-01 14:21:00,003][INFO ][run:send:168] Sending task app.entrypoints.jobs.modules.example.endpoints.jobs:hello_world_job with schedule_id 21e22e1a-0fc6-42cc-9225-3b4530907f84.
[2025-12-01 14:21:10,001][ERROR ][base_events:default_exception_handler:1771] Task exception was never retrieved
future: <Task finished name='schedule_21e22e1a-0fc6-42cc-9225-3b4530907f84' coro=<send() done, defined at /opt/venv/lib/python3.11/site-packages/taskiq/cli/scheduler/run.py:156> exception=taskiq.exceptions.SendTaskError()>
Traceback (most recent call last):
File "asyncpg/protocol/prepared_stmt.pyx", line 175, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
File "asyncpg/protocol/codecs/base.pyx", line 227, in asyncpg.protocol.protocol.Codec.encode
File "asyncpg/protocol/codecs/base.pyx", line 129, in asyncpg.protocol.protocol.Codec.encode_scalar
File "asyncpg/pgproto/./codecs/bytea.pyx", line 19, in asyncpg.pgproto.pgproto.bytea_encode
TypeError: a bytes-like object is required, not 'str'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/opt/venv/lib/python3.11/site-packages/taskiq/kicker.py", line 164, in kiq
await self.broker.kick(self.broker.formatter.dumps(message))
File "/opt/venv/lib/python3.11/site-packages/taskiq_pg/asyncpg/broker.py", line 97, in kick
await conn.fetchval(
File "/opt/venv/lib/python3.11/site-packages/asyncpg/connection.py", line 714, in fetchval
data = await self._execute(query, args, 1, timeout)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/venv/lib/python3.11/site-packages/asyncpg/connection.py", line 1864, in _execute
result, _ = await self.__execute(
^^^^^^^^^^^^^^^^^^^^^
File "/opt/venv/lib/python3.11/site-packages/asyncpg/connection.py", line 1961, in __execute
result, stmt = await self._do_execute(
^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/venv/lib/python3.11/site-packages/asyncpg/connection.py", line 2024, in _do_execute
result = await executor(stmt, None)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "asyncpg/protocol/protocol.pyx", line 185, in bind_execute
File "asyncpg/protocol/prepared_stmt.pyx", line 204, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $3: '{"task_id": "f3aa373a9946412baf02c61783... (a bytes-like object is required, not 'str')
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/opt/venv/lib/python3.11/site-packages/taskiq/cli/scheduler/run.py", line 173, in send
await scheduler.on_ready(source, task)
File "/opt/venv/lib/python3.11/site-packages/taskiq/scheduler/scheduler.py", line 50, in on_ready
await (
File "/opt/venv/lib/python3.11/site-packages/taskiq/kicker.py", line 166, in kiq
raise SendTaskError from exc
taskiq.exceptions.SendTaskError: Cannot send task to the queueAdditional context
Кажется ошибка в файле asyncpg/broker.py в методе kick (строки 97-103)
await conn.fetchval(
INSERT_MESSAGE_QUERY.format(self.table_name),
message.task_id,
message.task_name,
message.message.decode(),
json.dumps(message.labels),
)Я пробовал убрать .decode() и задачи начали записываться в таблицу
await conn.fetchval(
INSERT_MESSAGE_QUERY.format(self.table_name),
message.task_id,
message.task_name,
message.message,
json.dumps(message.labels),
)