PostgreSQL broker for taskiq

README.md

@@ -1,6 +1,6 @@
 # backend-python
 
-Copier template for a Python backend, either as a `backend/` part next to other parts or alone at the root of the project (`layout`): a uv project with one Docker image and one `python -m <module>` entrypoint per service (`api`, `bot`, `userbot`, `worker`, `scheduler`), pydantic-settings, rich logging, optional Dishka, MongoDB or PostgreSQL, Redis, taskiq, CryptoPay, pydantic-ai and a DynamicConfig stored in the database.
+Copier template for a Python backend, either as a `backend/` part next to other parts or alone at the root of the project (`layout`): a uv project with one Docker image and one `python -m <module>` entrypoint per service (`api`, `bot`, `userbot`, `worker`, `scheduler`), pydantic-settings, rich logging, optional Dishka, MongoDB or PostgreSQL, Redis, taskiq (in-memory, Redis or PostgreSQL broker), CryptoPay, pydantic-ai and a DynamicConfig stored in the database.
 
 Ships `backend/docker-compose.yml` with root-relative paths, `docker-compose.local.yml` (published ports) and `docker-compose.prod.yml` (the `caddy` network), a `Makefile` with `fmt`, `check`, `build` and service-specific targets, `.env.example` and `CLAUDE.md`. Applied on top of the `infra` template, `layout: parts` or `layout: single` respectively.
 

copier.yml

@@ -71,6 +71,9 @@ taskiq_broker:
   choices:
     in-memory: memory
     Redis: redis
+    PostgreSQL (taskiq-pg):
+      value: postgres
+      validator: "{% if database != 'postgres' %}Requires the PostgreSQL database{% endif %}"
   default: memory
 
 include_payments:

template/{{ part_dir }}/pyproject.toml.jinja

@@ -4,7 +4,7 @@
 {%- if 'aiogram_bot' in backend_services %}{% set _ = deps.append("aiogram>=3.29.0") %}{% endif -%}
 {%- if include_payments %}{% set _ = deps.append("aiosend>=3.0.7") %}{% endif -%}
 {%- if 'kurigram_userbot' in backend_services %}{% set _ = deps.append("anyio>=4.0.0") %}{% set _ = deps.append("kurigram>=2.2.7") %}{% set _ = deps.append("tgcrypto>=1.2.5") %}{% set _ = deps.append("uvloop>=0.21.0") %}{% endif -%}
-{%- if 'taskiq_worker' in backend_services %}{% set _ = deps.append("taskiq>=0.12.4") %}{% if taskiq_broker == 'redis' %}{% set _ = deps.append("taskiq-redis>=1.2.3") %}{% endif %}{% endif -%}
+{%- if 'taskiq_worker' in backend_services %}{% set _ = deps.append("taskiq>=0.12.4") %}{% if taskiq_broker == 'redis' %}{% set _ = deps.append("taskiq-redis>=1.2.3") %}{% endif %}{% if taskiq_broker == 'postgres' %}{% set _ = deps.append("taskiq-pg>=0.2.0") %}{% endif %}{% endif -%}
 {%- if 'scheduler' in backend_services %}{% set _ = deps.append("apscheduler>=3.11.0") %}{% endif -%}
 {%- if database == 'mongo' %}{% set _ = deps.append("beanie>=2.0.1") %}{% endif -%}
 {%- if database == 'postgres' %}{% set _ = deps.append("sqlmodel>=0.0.38") %}{% set _ = deps.append("asyncpg>=0.31.0") %}{% set _ = deps.append("alembic>=1.18.4") %}{% set _ = deps.append("greenlet>=3.1.0") %}{% endif -%}

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

@@ -39,6 +39,22 @@ broker = ResilientListQueueBroker(env.redis.url).with_result_backend(
 {% endif %}{% if use_dishka %}
 setup_dishka(container, broker)
 {% endif %}
+{%- elif taskiq_broker == 'postgres' -%}
+{% if use_dishka %}from dishka.integrations.taskiq import setup_dishka
+{% endif %}{% if 'scheduler' in backend_services %}from taskiq import TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+{% endif %}from taskiq_pg import AsyncpgBroker, AsyncpgResultBackend
+
+{% if use_dishka %}from dependencies.container import container
+{% endif %}from utils.env import env
+
+broker = AsyncpgBroker(env.db.connection_url).with_result_backend(
+    AsyncpgResultBackend(env.db.connection_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