feat(redis): restore redis consumers and taskiq

Author: dlivxpr

docs/migration-traceability.toml

@@ -313,3 +313,176 @@ allowed_transformations = ["删除 owner 外键与 index", "重置为初始 migr
 removed_references = ["owner_id、users.id ForeignKey、ix_items_owner_id、legacy down_revision"]
 approved_deviation = ["保留 revision 0021_create_items 身份且 down_revision=None"]
 verification = ["test_item_crud_switch_controls_the_complete_generated_slice", "test_both_orms_pass_real_postgresql_contract"]
+
+# Issue #9 Redis consumers 与 Taskiq 独立 profile。
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/pyproject.toml"
+target = "pyproject.toml"
+allowed_transformations = ["Copier 模板语法", "仅保留 Redis、cache 与 rate limit 依赖", "依赖版本兼容修改"]
+removed_references = ["V1 排除能力依赖"]
+approved_deviation = ["Logfire 关闭时不生成 instrumentation extras"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/.env.example"
+target = ".env.example"
+allowed_transformations = ["Copier 模板语法", "仅保留启用 consumer 的环境变量"]
+removed_references = ["V1 排除能力与 Taskiq 变量"]
+approved_deviation = ["RATE_LIMIT_STORAGE 显式记录 redis"]
+verification = ["test_issue_9_generation_boundaries"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/main.py"
+target = "app/main.py"
+allowed_transformations = ["删除 V1 排除生命周期", "保留 Redis、cache 与 limiter 生命周期"]
+removed_references = ["用户、RAG、消息渠道、admin 与其他 V1 排除能力"]
+approved_deviation = ["FastAPICache shutdown 显式 reset", "SlowAPI 使用 ASGI middleware 与统一异常 envelope"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/config.py"
+target = "app/core/config.py"
+allowed_transformations = ["删除 V1 排除 Settings", "保留 Redis 与 limiter Settings"]
+removed_references = ["身份、billing、文件、RAG 与 provider Settings"]
+approved_deviation = ["RATE_LIMIT_STORAGE 作为显式 Settings"]
+verification = ["test_issue_9_generation_boundaries"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/clients/redis.py"
+target = "app/clients/redis.py"
+allowed_transformations = ["Copier 模板语法", "当前 redis asyncio close API"]
+removed_references = ["无"]
+approved_deviation = ["raw client 保持 binary 供 FastAPICache 使用，get 执行 UTF-8 decode"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/cache.py"
+target = "app/core/cache.py"
+allowed_transformations = ["Copier 模板语法", "增加 shutdown reset"]
+removed_references = ["disabled no-op 分支"]
+approved_deviation = ["文件仅在启用 cache 时生成"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/rate_limit.py"
+target = "app/core/rate_limit.py"
+allowed_transformations = ["Copier 模板语法", "增加 Redis storage URI", "保留 low/medium/high decorators"]
+removed_references = ["disabled 空模块"]
+approved_deviation = ["Redis rate limit 使用 settings.REDIS_URL 共享服务端状态"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/api/deps.py"
+target = "app/api/deps.py"
+allowed_transformations = ["删除 V1 排除 dependencies", "deployment API Key 替换", "保留 Redis dependency"]
+removed_references = ["用户、JWT、teams 与排除 service dependencies"]
+approved_deviation = ["Redis client 从 app lifespan state 注入"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "redis_consumers"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/api/routes/v1/health.py"
+target = "app/api/health.py"
+allowed_transformations = ["根路径迁移", "删除排除资源 probes", "保留 Redis readiness"]
+removed_references = ["RAG、provider、billing、Celery probes"]
+approved_deviation = ["Redis probe 使用 dependency-injected RedisClient"]
+verification = ["test_redis_consumers_pass_real_redis_contract"]
+
+[[target]]
+profile = "rate_limit_memory"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/pyproject.toml"
+target = "pyproject.toml"
+allowed_transformations = ["Copier 模板语法", "仅保留 SlowAPI 依赖"]
+removed_references = ["V1 排除能力与 Redis 依赖"]
+approved_deviation = ["无"]
+verification = ["test_memory_rate_limit_uses_no_redis"]
+
+[[target]]
+profile = "rate_limit_memory"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/.env.example"
+target = ".env.example"
+allowed_transformations = ["Copier 模板语法", "仅保留 limiter 环境变量"]
+removed_references = ["V1 排除能力与 Redis 变量"]
+approved_deviation = ["RATE_LIMIT_STORAGE 显式记录 memory"]
+verification = ["test_memory_rate_limit_uses_no_redis"]
+
+[[target]]
+profile = "rate_limit_memory"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/config.py"
+target = "app/core/config.py"
+allowed_transformations = ["删除 V1 排除 Settings", "保留 limiter Settings"]
+removed_references = ["Redis 与其他未启用能力 Settings"]
+approved_deviation = ["RATE_LIMIT_STORAGE 作为显式 Settings"]
+verification = ["test_memory_rate_limit_uses_no_redis"]
+
+[[target]]
+profile = "rate_limit_memory"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/rate_limit.py"
+target = "app/core/rate_limit.py"
+allowed_transformations = ["Copier 模板语法", "显式 memory storage URI", "保留 low/medium/high decorators"]
+removed_references = ["disabled 空模块"]
+approved_deviation = ["memory:// 明确限制为 process-local storage"]
+verification = ["test_memory_rate_limit_uses_no_redis"]
+
+[[target]]
+profile = "rate_limit_memory"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/main.py"
+target = "app/main.py"
+allowed_transformations = ["删除 V1 排除生命周期", "保留 SlowAPI middleware"]
+removed_references = ["Redis 与其他未启用资源生命周期"]
+approved_deviation = ["使用 SlowAPIASGIMiddleware 与统一异常 envelope"]
+verification = ["test_memory_rate_limit_uses_no_redis"]
+
+[[target]]
+profile = "taskiq"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/pyproject.toml"
+target = "pyproject.toml"
+allowed_transformations = ["Copier 模板语法", "仅保留 Taskiq 与 taskiq-redis 依赖", "依赖版本兼容修改"]
+removed_references = ["V1 排除 queue 与 API Redis client 依赖"]
+approved_deviation = ["无"]
+verification = ["test_taskiq_process_commands_start_and_stop_against_real_redis"]
+
+[[target]]
+profile = "taskiq"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/.env.example"
+target = ".env.example"
+allowed_transformations = ["Copier 模板语法", "仅保留 Taskiq 环境变量"]
+removed_references = ["V1 排除能力与 API Redis client 变量"]
+approved_deviation = ["result backend 变量命名为 TASKIQ_RESULT_BACKEND_URL"]
+verification = ["test_issue_9_generation_boundaries"]
+
+[[target]]
+profile = "taskiq"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/core/config.py"
+target = "app/core/config.py"
+allowed_transformations = ["删除 V1 排除 Settings", "保留独立 Taskiq URLs"]
+removed_references = ["API Redis client 与其他未启用能力 Settings"]
+approved_deviation = ["result backend 变量命名为 TASKIQ_RESULT_BACKEND_URL"]
+verification = ["test_issue_9_generation_boundaries"]
+
+[[target]]
+profile = "taskiq"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/app/worker/taskiq_app.py"
+target = "app/worker/taskiq_app.py"
+allowed_transformations = ["Copier 模板语法", "删除 legacy schedule import", "startup 与 shutdown 执行 Redis health"]
+removed_references = ["RAG schedule 与虚构业务 task"]
+approved_deviation = ["生命周期 hooks 同时验证 broker 与 result backend，pool 关闭仍由 Taskiq owner 管理"]
+verification = ["test_taskiq_process_commands_start_and_stop_against_real_redis"]
+
+[[target]]
+profile = "taskiq"
+legacy_source = "legacy/template/{{cookiecutter.project_slug}}/backend/cli/commands.py"
+target = "cli/commands.py"
+allowed_transformations = ["删除 V1 排除 commands", "保留 server 与 Taskiq commands"]
+removed_references = ["用户、Celery、database 与 custom commands"]
+approved_deviation = ["subprocess.run 使用 check=True"]
+verification = ["test_taskiq_process_commands_start_and_stop_against_real_redis"]

template/.env.example.jinja

@@ -16,6 +16,10 @@ DB_POOL_SIZE=[[ db_pool_size ]]
 DB_MAX_OVERFLOW=[[ db_max_overflow ]]
 DB_POOL_TIMEOUT=[[ db_pool_timeout ]]
 [% endif %]
+[% if background_tasks == "taskiq" %]
+TASKIQ_BROKER_URL=redis://localhost:6379/1
+TASKIQ_RESULT_BACKEND_URL=redis://localhost:6379/1
+[% endif %]
 [% if enable_redis %]
 REDIS_HOST=localhost
 REDIS_PORT=6379

template/AGENTS.md.jinja

@@ -14,6 +14,11 @@
 - `alembic/`：PostgreSQL migration；通过 `uv run [[ project_slug ]] db ...` 管理。
 [% endif %][% if database == "postgresql" and include_example_crud %]- `app/db/models/item.py` → `app/repositories/item.py` → `app/services/item.py` → `app/api/items.py`：可选 Item CRUD。
 [% endif %]
+[% if enable_redis %]- `app/clients/redis.py`：API lifespan 拥有的 async Redis client；通过 `app.api.deps.Redis` 注入。
+[% endif %][% if enable_caching %]- `app/core/cache.py`：复用 API Redis client 的 FastAPICache backend。
+[% endif %][% if enable_rate_limiting %]- `app/core/rate_limit.py`：SlowAPI limiter 与 low/medium/high decorators。
+[% endif %][% if background_tasks == "taskiq" %]- `app/worker/taskiq_app.py`：独立 Taskiq broker、result backend 与 scheduler。
+[% endif %]
 
 ```bash
 uv sync --all-groups
@@ -36,6 +41,12 @@ uv run pytest
 - migration 使用 `uv run [[ project_slug ]] db upgrade` 执行；readiness 通过数据库 session 执行 `SELECT 1`。
 [% endif %][% if database == "postgresql" and include_example_crud %]- Item 仅提供 create/get/update/delete；不添加 owner/user 外键、list endpoint 或分页 schema。
 [% endif %]
+[% if enable_redis %]- Redis client 必须由 API lifespan `connect()`/`close()`；readiness 通过注入的 client 执行 `ping()`。
+[% endif %][% if enable_caching %]- FastAPICache 与 Redis client 共享 binary connection；shutdown 时调用 `FastAPICache.reset()`。
+[% endif %][% if enable_rate_limiting %]- Rate limiting 使用 [[ "Redis shared storage" if rate_limit_storage == "redis" else "process-local memory storage" ]]；业务 route 按成本选择 `rate_limit_low`、`rate_limit_medium` 或 `rate_limit_high`。
+[% endif %][% if background_tasks == "taskiq" %]- Taskiq 只使用 `TASKIQ_BROKER_URL` 与 `TASKIQ_RESULT_BACKEND_URL`，不复用 API Redis client 生命周期。
+- 使用 `uv run [[ project_slug ]] taskiq worker` 和 `uv run [[ project_slug ]] taskiq scheduler` 启动进程；业务 task 在调用方模块中通过 `broker.task` 定义。
+[% endif %]
 ## 约束
 
 - I/O 和生命周期路径使用 async。

template/README.md.jinja

@@ -26,3 +26,27 @@ uv run [[ project_slug ]] db current
 
 Item 示例 API：`POST /api/v1/items`、`GET /api/v1/items/{item_id}`、`PATCH /api/v1/items/{item_id}`、`DELETE /api/v1/items/{item_id}`。
 [% endif %][% endif %]
+[% if enable_redis %]
+
+## Redis
+
+配置 `REDIS_HOST`、`REDIS_PORT`、`REDIS_PASSWORD` 和 `REDIS_DB`。API lifespan 管理连接，`/health/ready` 验证 Redis。
+[% endif %][% if enable_caching %]
+
+FastAPICache 使用相同的 API Redis connection，并以 `[[ project_slug ]]:cache:` 为 key prefix。
+[% endif %][% if enable_rate_limiting %]
+
+## Rate limiting
+
+SlowAPI 默认限制为 `RATE_LIMIT_REQUESTS` / `RATE_LIMIT_PERIOD`，storage 为 `[[ rate_limit_storage ]]`。业务 route 使用 `rate_limit_low`、`rate_limit_medium` 或 `rate_limit_high` decorator。
+[% endif %][% if background_tasks == "taskiq" %]
+
+## Taskiq
+
+Taskiq 使用独立的 `TASKIQ_BROKER_URL` 与 `TASKIQ_RESULT_BACKEND_URL`：
+
+```bash
+uv run [[ project_slug ]] taskiq worker
+uv run [[ project_slug ]] taskiq scheduler
+```
+[% endif %]

template/[% if ci_type == 'github' %].github[% endif %]/workflows/ci.yml.jinja

@@ -7,7 +7,7 @@ on:
 jobs:
   quality:
     runs-on: ubuntu-latest
-[% if database == "postgresql" or enable_redis %]    services:
+[% if database == "postgresql" or enable_redis or background_tasks == "taskiq" %]    services:
 [% if database == "postgresql" %]      postgres:
         image: postgres:17-alpine
         env:
@@ -21,7 +21,7 @@ jobs:
           --health-interval 2s
           --health-timeout 3s
           --health-retries 15
-[% endif %][% if enable_redis %]      redis:
+[% endif %][% if enable_redis or background_tasks == "taskiq" %]      redis:
         image: redis:8-alpine
         ports:
           - 6379:6379
@@ -38,6 +38,8 @@ jobs:
       POSTGRES_DB: app
 [% endif %][% if enable_redis %]      REDIS_HOST: 127.0.0.1
       REDIS_PORT: 6379
+[% endif %][% if background_tasks == "taskiq" %]      TASKIQ_BROKER_URL: redis://127.0.0.1:6379/1
+      TASKIQ_RESULT_BACKEND_URL: redis://127.0.0.1:6379/1
 [% endif %][% if enable_logfire %]      LOGFIRE_SEND_TO_LOGFIRE: "false"
 [% endif %]    steps:
       - uses: actions/checkout@v4

template/[% if enable_docker %]deploy[% endif %]/compose.yaml.jinja

@@ -6,7 +6,10 @@ services:
     environment:
       API_KEY: "${API_KEY:-[[ deployment_api_key ]]}"
 [% if database == "postgresql" %]      DATABASE_URL: postgresql+asyncpg://postgres:postgres@postgres:5432/app
-[% endif %][% if enable_redis %]      REDIS_URL: redis://redis:6379/0
+[% endif %][% if enable_redis %]      REDIS_HOST: redis
+      REDIS_PORT: 6379
+[% endif %][% if background_tasks == "taskiq" %]      TASKIQ_BROKER_URL: redis://redis:6379/1
+      TASKIQ_RESULT_BACKEND_URL: redis://redis:6379/1
 [% endif %][% if ai_framework == "pydantic_ai" %]      LLM_BASE_URL: "${LLM_BASE_URL:-https://api.example.com/v1}"
       LLM_API_KEY: "${LLM_API_KEY:-model-key-CHANGEME}"
       LLM_MODEL: "${LLM_MODEL:-model-name-CHANGEME}"
@@ -28,9 +31,10 @@ services:
     build:
       context: ..
       dockerfile: deploy/Dockerfile
-    command: ["uv", "run", "--no-sync", "taskiq", "worker", "app.tasks.taskiq:broker"]
+    command: ["uv", "run", "--no-sync", "taskiq", "worker", "app.worker.taskiq_app:broker"]
     environment:
-      REDIS_URL: redis://redis:6379/0
+      TASKIQ_BROKER_URL: redis://redis:6379/1
+      TASKIQ_RESULT_BACKEND_URL: redis://redis:6379/1
 [% if enable_logfire %]      LOGFIRE_TOKEN: "${LOGFIRE_TOKEN:-}"
 [% endif %]
     depends_on:
@@ -41,9 +45,10 @@ services:
     build:
       context: ..
       dockerfile: deploy/Dockerfile
-    command: ["uv", "run", "--no-sync", "taskiq", "scheduler", "app.tasks.taskiq:scheduler"]
+    command: ["uv", "run", "--no-sync", "taskiq", "scheduler", "app.worker.taskiq_app:scheduler"]
     environment:
-      REDIS_URL: redis://redis:6379/0
+      TASKIQ_BROKER_URL: redis://redis:6379/1
+      TASKIQ_RESULT_BACKEND_URL: redis://redis:6379/1
 [% if enable_logfire %]      LOGFIRE_TOKEN: "${LOGFIRE_TOKEN:-}"
 [% endif %]
     depends_on:

template/app/[% if background_tasks == 'taskiq' %]tasks[% endif %]/__init__.py

@@ -0,0 +1 @@
+"""Taskiq worker process configuration."""

template/app/[% if background_tasks == 'taskiq' %]tasks[% endif %]/taskiq.py.jinja

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from redis.asyncio import Redis
+from taskiq import TaskiqEvents, TaskiqScheduler, TaskiqState
+from taskiq.schedule_sources import LabelScheduleSource
+from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
+
+from app.core.config import settings
+[% if enable_logfire %]from app.core.telemetry import configure_telemetry
+[% endif -%]
+[[ "\n" -]]
+result_backend = RedisAsyncResultBackend(redis_url=settings.TASKIQ_RESULT_BACKEND_URL)
+broker = ListQueueBroker(url=settings.TASKIQ_BROKER_URL).with_result_backend(result_backend)
+scheduler = TaskiqScheduler(broker=broker, sources=[LabelScheduleSource(broker)])
+
+
+async def ping_redis_consumers() -> None:
+    """Fail when either Taskiq Redis consumer is unhealthy."""
+    async with Redis(connection_pool=broker.connection_pool) as broker_redis:
+        await broker_redis.ping()
+    async with Redis(connection_pool=result_backend.redis_pool) as result_redis:
+        await result_redis.ping()
+
+
+@broker.on_event(TaskiqEvents.CLIENT_STARTUP, TaskiqEvents.WORKER_STARTUP)
+async def startup(_state: TaskiqState) -> None:
+    """Configure integrations and validate Taskiq Redis connections."""
+[% if enable_logfire %]    configure_telemetry()
+[% endif %]    await ping_redis_consumers()
+
+
+@broker.on_event(TaskiqEvents.CLIENT_SHUTDOWN, TaskiqEvents.WORKER_SHUTDOWN)
+async def shutdown(_state: TaskiqState) -> None:
+    """Validate Redis before Taskiq closes its broker resources."""
+    await ping_redis_consumers()