Single part at the root, COMPOSE_FILE in .env

template/{{ part_dir }}/src/{% if 'taskiq_worker' in backend_services %}worker{% endif %}/__main__.py.jinja

@@ -0,0 +1,23 @@
+import os
+{% if 'scheduler' in backend_services %}import sys
+{% endif %}
+BROKER = "worker.broker:broker"
+{% if 'scheduler' in backend_services %}SCHEDULER = "worker.broker:scheduler"
+{% endif %}TASKS = "worker.tasks"
+
+
+def main() -> None:
+    {%- if 'scheduler' in backend_services %}
+    mode = sys.argv[1] if len(sys.argv) > 1 else "worker"
+    if mode == "scheduler":
+        argv = ["taskiq", "scheduler", SCHEDULER, TASKS]
+    else:
+        argv = ["taskiq", "worker", BROKER, TASKS]
+    {%- else %}
+    argv = ["taskiq", "worker", BROKER, TASKS]
+    {%- endif %}
+    os.execvp("taskiq", argv)  # noqa: S606, S607
+
+
+if __name__ == "__main__":
+    main()

template/{{ part_dir }}/src/{% if 'taskiq_worker' in backend_services %}worker{% endif %}/broker.py.jinja

@@ -0,0 +1,54 @@
+{% if taskiq_broker == 'redis' -%}
+from collections.abc import AsyncGenerator
+
+{% if use_dishka %}from dishka.integrations.taskiq import setup_dishka
+{% endif %}from redis.asyncio import Redis
+from redis.exceptions import ConnectionError as RedisConnectionError
+from redis.exceptions import TimeoutError as RedisTimeoutError
+{% if 'scheduler' in backend_services %}from taskiq import TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+{% endif %}from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
+
+{% if use_dishka %}from dependencies.container import container
+{% endif %}from utils.env import env
+from utils.logging import logger
+
+BRPOP_TIMEOUT = 2
+
+
+class ResilientListQueueBroker(ListQueueBroker):
+    async def listen(self) -> AsyncGenerator[bytes]:
+        while True:
+            try:
+                async with Redis(connection_pool=self.connection_pool) as conn:
+                    result = await conn.brpop(self.queue_name, timeout=BRPOP_TIMEOUT)
+                    if result is None:
+                        continue
+                    value = result[1]
+                    yield value if isinstance(value, bytes) else value.encode()
+            except RedisTimeoutError:
+                continue
+            except (RedisConnectionError, OSError) as exc:
+                logger.warning("redis listen error: %s", exc)
+
+
+broker = ResilientListQueueBroker(env.redis.url).with_result_backend(
+    RedisAsyncResultBackend(env.redis.url)
+)
+{% if 'scheduler' in backend_services %}scheduler = TaskiqScheduler(broker, sources=[LabelScheduleSource(broker)])
+{% endif %}{% if use_dishka %}
+setup_dishka(container, broker)
+{% endif %}
+{%- else -%}
+{% if use_dishka %}from dishka.integrations.taskiq import setup_dishka
+{% endif %}from taskiq import InMemoryBroker{% if 'scheduler' in backend_services %}, TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource{% endif %}
+
+{% if use_dishka %}from dependencies.container import container
+{% endif %}
+broker = InMemoryBroker()
+{% if 'scheduler' in backend_services %}scheduler = TaskiqScheduler(broker, sources=[LabelScheduleSource(broker)])
+{% endif %}{% if use_dishka %}
+setup_dishka(container, broker)
+{% endif %}
+{%- endif %}

template/{{ part_dir }}/src/{% if 'taskiq_worker' in backend_services %}worker{% endif %}/tasks/__init__.py.jinja

@@ -0,0 +1,3 @@
+from . import example
+
+__all__ = ["example"]

template/{{ part_dir }}/src/{% if 'taskiq_worker' in backend_services %}worker{% endif %}/tasks/example.py.jinja

@@ -0,0 +1,8 @@
+from utils.logging import logger
+from worker.broker import broker
+
+
+@broker.task{% if 'scheduler' in backend_services %}(schedule=[{"cron": "*/5 * * * *"}]){% endif %}
+async def example() -> str:
+    logger.info("example task ran")
+    return "ok"