PostgreSQL broker for taskiq

README.md

@@ -1,10 +1,10 @@
 # backend-python
 
-Copier template for a `backend/` part: 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`. Meant to be applied on top of the `infra` template.
+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.
 
 ```sh
 copier copy --trust -d project_name="My Thing" <this-repo> <dest>
-copier update -a .copier-answers.backend.yml
+copier update -a .copier/backend.yml
 ```

copier.yml

@@ -1,6 +1,6 @@
 _subdirectory: template
 _templates_suffix: .jinja
-_answers_file: .copier-answers.backend.yml
+_answers_file: .copier/backend.yml
 _envops:
   keep_trailing_newline: true
 
@@ -25,6 +25,14 @@ author_email:
   type: str
   default: ""
 
+layout:
+  type: str
+  help: Where the backend lives
+  choices:
+    In backend/ next to other parts: part
+    At the root of the project: root
+  default: part
+
 backend_services:
   type: str
   multiselect: true
@@ -63,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:
@@ -84,7 +95,7 @@ dynamic_config:
 
 serve_spa:
   type: bool
-  when: "{{ 'api' in backend_services }}"
+  when: "{{ layout == 'part' and 'api' in backend_services }}"
   help: Serve the frontend build from the api as a SPA
   default: false
 
@@ -93,6 +104,16 @@ install_deps:
   help: uv sync after generation
   default: true
 
+part_dir:
+  type: str
+  when: false
+  default: "{{ 'backend' if layout == 'part' else '.' }}"
+
+prefix:
+  type: str
+  when: false
+  default: "{{ 'backend/' if layout == 'part' else '' }}"
+
 _tasks:
-  - "{% if _copier_operation == 'copy' and install_deps %}sh -c 'cd backend && uv sync'{% else %}true{% endif %}"
+  - "{% if _copier_operation == 'copy' and install_deps %}sh -c 'cd {{ part_dir }} && uv sync'{% else %}true{% endif %}"
-  - "{% if install_deps %}sh -c 'cd backend && uv run ruff format -q'{% else %}true{% endif %}"
+  - "{% if install_deps %}sh -c 'cd {{ part_dir }} && uv run ruff format -q'{% else %}true{% endif %}"

template/ty.toml

@@ -1,5 +0,0 @@
-[environment]
-python = "backend/.venv"
-
-[src]
-exclude = ["backend/migrations"]

template/ty.toml.jinja

@@ -0,0 +1,5 @@
+[environment]
+python = "{{ prefix }}.venv"
+
+[src]
+exclude = ["{{ prefix }}migrations"]

template/{{ part_dir }}/.env.example.jinja

@@ -1,4 +1,5 @@
-RUN_ENVIRONMENT=dev
+{% if layout == 'root' %}COMPOSE_FILE=docker-compose.yml:docker-compose.local.yml
+{% endif %}RUN_ENVIRONMENT=dev
 LOG__LEVEL=INFO
 LOG__LEVEL_EXTERNAL=WARNING
 LOG__SHOW_TIME=false

template/{{ part_dir }}/Makefile.jinja

@@ -5,9 +5,11 @@
 {%- if 'kurigram_userbot' in backend_services %}{% set _ = svc_modules.append("userbot") %}{% endif -%}
 {%- if 'scheduler' in backend_services and 'taskiq_worker' not in backend_services %}{% set _ = svc_modules.append("scheduler") %}{% endif -%}
 {%- set has_script = dynamic_config and svc_modules -%}
-COMPOSE := docker compose --project-directory ..
+COMPOSE := docker compose{% if layout == 'part' %} --project-directory ..{% endif %}
-
+{% if layout == 'root' %}SERVICES := $(COMPOSE) --profile services
-.PHONY: fmt check build{% if database == 'postgres' %} migrate revision{% endif %}{% if has_script %} script{% endif %}{% if 'kurigram_userbot' in backend_services %} session{% endif %}
+EXTERNAL := $(COMPOSE) --profile external
+{% endif %}
+.PHONY: {% if layout == 'root' %}env external recreate rebuild restart logs down {% endif %}fmt check build{% if database == 'postgres' %} migrate revision{% endif %}{% if has_script %} script{% endif %}{% if 'kurigram_userbot' in backend_services %} session{% endif %}
 
 fmt:
 	uv run ruff format
@@ -20,7 +22,27 @@ check:
 
 build:
 	$(COMPOSE) build {{ svc_modules[0] }}
-{% if database == 'postgres' %}
+{% if layout == 'root' %}
+env:
+	@test -f .env || cp .env.example .env
+
+external:
+	$(EXTERNAL) up -d
+
+recreate: external
+	$(SERVICES) up -d --force-recreate
+
+rebuild: build recreate
+
+restart:
+	$(SERVICES) restart
+
+logs:
+	$(SERVICES) logs -f --tail=100
+
+down:
+	$(EXTERNAL) --profile services down
+{% endif %}{% if database == 'postgres' %}
 migrate:
 	$(COMPOSE) --profile external run --rm migrator
 

template/{{ part_dir }}/docker-compose.yml.jinja

@@ -9,21 +9,21 @@
 {%- if 'kurigram_userbot' in backend_services %}{% set _ = app_services.append({'name': 'userbot', 'command': '[userbot]'}) %}{% endif -%}
 {%- if 'scheduler' in backend_services %}{% set _ = app_services.append({'name': 'scheduler', 'command': ('[worker, scheduler]' if 'taskiq_worker' in backend_services else '[scheduler]')}) %}{% endif -%}
 {%- macro backend_service(name, command) %}  {{ name }}:
-    build: backend
+    build: {{ part_dir }}
     image: {{ project_slug }}/backend
     profiles: [{{ name }}, services]
     restart: unless-stopped
     env_file:
       - path: .env
         required: false
-      - path: backend/.env
+{% if layout == 'part' %}      - path: backend/.env
         required: false
-    environment:
+{% endif %}    environment:
       RUN_ENVIRONMENT: prod
     volumes:
-      - ./backend/src:/app/src
+      - ./{{ prefix }}src:/app/src
-{% if has_scripts %}      - ./backend/scripts:/app/scripts
+{% if has_scripts %}      - ./{{ prefix }}scripts:/app/scripts
-{% endif %}{% if name == 'userbot' %}      - ./backend/sessions:/app/sessions
+{% endif %}{% if name == 'userbot' %}      - ./{{ prefix }}sessions:/app/sessions
 {% endif %}{% if name == 'api' and serve_spa %}      - ./frontend/build:/app/static:ro
 {% endif %}    command: {{ command }}
 {% if svc_networks %}    networks:
@@ -70,20 +70,20 @@ services:
           - ${DB__HOST:-postgres}
 
   migrator:
-    build: backend
+    build: {{ part_dir }}
     image: {{ project_slug }}/backend
     profiles: [migrate]
     env_file:
       - path: .env
         required: false
-      - path: backend/.env
+{% if layout == 'part' %}      - path: backend/.env
         required: false
-    environment:
+{% endif %}    environment:
       RUN_ENVIRONMENT: prod
     volumes:
-      - ./backend/src:/app/src
+      - ./{{ prefix }}src:/app/src
-      - ./backend/migrations:/app/migrations
+      - ./{{ prefix }}migrations:/app/migrations
-      - ./backend/alembic.ini:/app/alembic.ini
+      - ./{{ prefix }}alembic.ini:/app/alembic.ini
     entrypoint: [alembic]
     command: [upgrade, head]
     depends_on:

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