feat: TaskIQ Integration option (#23)

Author: mslaursen

fastapi_forge/dtos.py

@@ -295,6 +295,7 @@ class ProjectSpec(_Base):
     use_builtin_auth: bool = False
     use_redis: bool = False
     use_rabbitmq: bool = False
+    use_taskiq: bool = False
     models: list[Model] = []
 
     @model_validator(mode="after")
@@ -329,6 +330,18 @@ def _validate_models(self) -> Self:
             msg = "Only one model can be an auth user."
             raise ValueError(msg)
 
+        if self.use_taskiq and not (self.use_redis and self.use_rabbitmq):
+            missing = []
+            if not self.use_rabbitmq:
+                missing.append("RabbitMQ")
+            if not self.use_redis:
+                missing.append("Redis")
+
+            if missing:
+                raise ValueError(
+                    "TaskIQ is enabled, but the following are missing and required "
+                    f"for its operation: {', '.join(missing)}."
+                )
         return self
 
     @model_validator(mode="after")

fastapi_forge/example-projects/game_zone.yaml

@@ -5,6 +5,7 @@ project:
   use_builtin_auth: true
   use_redis: true
   use_rabbitmq: true
+  use_taskiq: true
 
   models:
     - name: auth_user

fastapi_forge/frontend/panels/project_config_panel.py

@@ -22,6 +22,7 @@ def __init__(self):
         super().__init__(value=True, elevated=False, bottom_corner=True)
         self._build()
         self._bind_state_to_ui()
+        self._update_taskiq_state()
 
     def _build(self) -> None:
         with (
@@ -83,16 +84,16 @@ def _build(self) -> None:
                 self.use_rabbitmq = ui.checkbox(
                     "RabbitMQ",
                     value=state.use_rabbitmq,
+                    on_change=self._update_taskiq_state,
                 ).classes("w-full")
 
             with ui.column().classes("w-full gap-2"):
                 ui.label("Task Queues").classes("text-lg font-bold")
-                self.use_taskiq = (
-                    ui.checkbox("Taskiq")
-                    .classes("w-full")
-                    .tooltip("Coming soon!")
-                    .set_enabled(False)
-                )
+                self.use_taskiq = ui.checkbox(
+                    "Taskiq",
+                    value=state.use_taskiq,
+                    on_change=self._update_taskiq_state,
+                ).classes("w-full")
                 self.use_celery = (
                     ui.checkbox("Celery")
                     .classes("w-full")
@@ -123,6 +124,7 @@ def _build(self) -> None:
                 self.use_redis = ui.checkbox(
                     "Redis",
                     value=state.use_redis,
+                    on_change=self._update_taskiq_state,
                 ).classes("w-full")
 
             with ui.column().classes("w-full gap-2"):
@@ -141,7 +143,23 @@ def _bind_state_to_ui(self) -> None:
         self.use_postgres.bind_value_to(state, "use_postgres")
         self.use_alembic.bind_value_to(state, "use_alembic")
         self.use_builtin_auth.bind_value_to(state, "use_builtin_auth")
-        self.use_rabbitmq.bind_value_to(state, "use_rabbitmq")
+        self.use_rabbitmq.bind_value_to(state, "use_rabbitmq").on(
+            "change", self._update_taskiq_state
+        )
+        self.use_redis.bind_value_to(state, "use_redis").on(
+            "change", self._update_taskiq_state
+        )
+        self.use_taskiq.bind_value_to(state, "use_taskiq")
+
+    def _update_taskiq_state(self, *_) -> None:
+        """Enable or disable Taskiq based on Redis and RabbitMQ."""
+        self.use_taskiq.set_enabled(self.use_redis.value and self.use_rabbitmq.value)
+        if (
+            not (self.use_redis.value and self.use_rabbitmq.value)
+            and self.use_taskiq.value
+        ):
+            self.use_taskiq.value = False
+            state.use_taskiq = False
 
     def _handle_builtin_auth_change(self, event: ValueChangeEventArguments) -> None:
         """Handle JWT Auth checkbox changes"""
@@ -245,6 +263,7 @@ async def _create_project(self) -> None:
             state.use_builtin_auth = self.use_builtin_auth.value
             state.use_redis = self.use_redis.value
             state.use_rabbitmq = self.use_rabbitmq.value
+            state.use_taskiq = self.use_taskiq.value
 
             project_spec = state.get_project_spec()
             await build_project(project_spec)

fastapi_forge/frontend/state.py

@@ -37,6 +37,7 @@ class ProjectState(BaseModel):
     use_builtin_auth: bool = False
     use_redis: bool = False
     use_rabbitmq: bool = False
+    use_taskiq: bool = False
 
     def initialize_from_project(self, project: ProjectSpec) -> None:
         """Initialize state from an existing project specification."""
@@ -46,6 +47,7 @@ def initialize_from_project(self, project: ProjectSpec) -> None:
         self.use_builtin_auth = project.use_builtin_auth
         self.use_redis = project.use_redis
         self.use_rabbitmq = project.use_rabbitmq
+        self.use_taskiq = project.use_taskiq
         self.models = project.models.copy()
 
         self._trigger_ui_refresh()
@@ -111,6 +113,7 @@ def get_project_spec(self) -> ProjectSpec:
             use_builtin_auth=self.use_builtin_auth,
             use_redis=self.use_redis,
             use_rabbitmq=self.use_rabbitmq,
+            use_taskiq=self.use_taskiq,
             models=self.models,
         )
 

fastapi_forge/template/cookiecutter.json

@@ -17,6 +17,9 @@
     "use_rabbitmq": {
         "default": true
     },
+    "use_taskiq": {
+        "default": true
+    },
     "models": {
         "models": []
     },

fastapi_forge/template/{{cookiecutter.project_name}}/docker-compose.yaml

@@ -1,5 +1,6 @@
 services:
   api:
+    &api
     build:
       context: .
     image: {{ cookiecutter.project_name }}:latest
@@ -71,6 +72,19 @@ services:
           path: pyproject.toml
         - action: rebuild
           path: uv.lock
+  {% if cookiecutter.use_taskiq %}
+  taskiq-worker:
+    <<: *api
+    container_name: game_zone-taskiq-worker
+    ports: []
+    command: [ taskiq, worker, -fsd, src.services.taskiq.broker:broker, -w, "1", --max-fails, "1"]
+
+  taskiq-scheduler:
+    <<: *api
+    container_name: game_zone-taskiq-scheduler
+    ports: []
+    command: [ taskiq, scheduler, -fsd, src.services.taskiq.scheduler:scheduler ]
+  {%- endif %}
   {% if cookiecutter.use_postgres %}
   postgres:
     image: postgres:13.8-bullseye
@@ -108,7 +122,7 @@ services:
   {% endif %}
   {%- if cookiecutter.use_rabbitmq -%}
   rabbitmq:
-    image: rabbitmq:3-management
+    image: rabbitmq:3.8.27-management-alpine
     container_name: {{ cookiecutter.project_name }}-rabbitmq
     ports:
       - "15672:15672"

fastapi_forge/template/{{cookiecutter.project_name}}/forge-config.yaml

@@ -28,11 +28,17 @@ paths:
     enabled: {{cookiecutter.use_rabbitmq | lower}}
     paths:
       - src/services/rabbitmq
-      - src/routes/demo_routes.py
+
+  use_taskiq:
+    enabled: {{cookiecutter.use_taskiq | lower}}
+    paths:
+      - src/services/taskiq
 
   constants:
     requires_all:
       - use_builtin_auth
       - use_rabbitmq
     paths:
-      - src/constants.py
+      - src/constants.py
+
+  

fastapi_forge/template/{{cookiecutter.project_name}}/pyproject.toml

@@ -14,8 +14,6 @@ dependencies = [
     "ruff>=0.9.4",
     {%- if cookiecutter.use_postgres -%}
     "sqlalchemy[asyncio]>=2.0.37",
-    {% endif %}
-    {%- if cookiecutter.use_postgres -%}
     "asyncpg>=0.30.0",
     {% endif %}
     {%- if cookiecutter.use_builtin_auth -%}
@@ -31,10 +29,18 @@ dependencies = [
     "factory-boy>=3.3.3",
     {%- if cookiecutter.use_redis -%}
     "redis>=5.2.1",
+    "fakeredis>=2.28.1",
     {% endif %}
     {%- if cookiecutter.use_rabbitmq -%}
     "aio-pika>=9.5.5",
     {%- endif %}
+    {%- if cookiecutter.use_taskiq -%}
+    "taskiq>=0.11.16",
+    "taskiq-aio-pika>=0.4.1",
+    "taskiq-redis>=1.0.4",
+    "taskiq-fastapi>=0.3.4",
+    "orjson>=3.10.16",
+    {%- endif %}
 ]
 
 [tool.pytest.ini_options]
@@ -64,7 +70,9 @@ ignore = [
     #### specific rules
     "A001",
     "A002",
+    "ARG002",
     "ARG001",
+    "B008",
     "B904",
     "BLE001",
     "D100",

fastapi_forge/template/{{cookiecutter.project_name}}/src/main.py

@@ -16,6 +16,9 @@
 from src.services.rabbitmq import rabbitmq_lifetime
 from src.constants import QUEUE_CONFIGS
 {% endif %}
+{% if cookiecutter.use_taskiq %}
+from src.services.taskiq import taskiq_lifetime
+{% endif %}
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
@@ -29,6 +32,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     {%- if cookiecutter.use_rabbitmq -%}
     await rabbitmq_lifetime.setup_rabbitmq(app, configs=QUEUE_CONFIGS)
     {% endif %}
+    {%- if cookiecutter.use_taskiq %}
+    await taskiq_lifetime.setup_taskiq()
+    {% endif %}
 
     yield
 
@@ -41,6 +47,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     {%- if cookiecutter.use_rabbitmq -%}
     await rabbitmq_lifetime.shutdown_rabbitmq(app)
     {% endif %}
+    {%- if cookiecutter.use_taskiq %}
+    await taskiq_lifetime.shutdown_taskiq()
+    {% endif %}
 
 
 def get_app() -> FastAPI:

fastapi_forge/template/{{cookiecutter.project_name}}/src/routes/demo_routes.py

@@ -7,6 +7,12 @@
 from pydantic import BaseModel
 from typing import Any
 
+{% if cookiecutter.use_taskiq %}
+from datetime import UTC, datetime, timedelta
+from src.services.taskiq import tasks
+from src.services.taskiq.scheduler import redis_source
+{% endif %}
+
 router = APIRouter(prefix="/demo")
 
 {% if cookiecutter.use_rabbitmq %}
@@ -17,10 +23,8 @@ class RabbitMQDemoMessage(BaseModel):
 
 {% if cookiecutter.use_redis %}
 @router.post("/set-redis")
-async def set_redis_value(key: str, value: str, redis: GetRedis,) -> dict[str, Any]:
+async def set_redis_value(key: str, value: str, redis: GetRedis,) -> None:
     await redis.set(key, value)
-    return {"message": "Value set successfully", "key": key, "value": value}
-
 
 @router.get("/get-redis")
 async def get_redis_value(key: str, redis: GetRedis,) -> dict[str, Any]:
@@ -35,11 +39,22 @@ async def get_redis_value(key: str, redis: GetRedis,) -> dict[str, Any]:
 async def send_rabbitmq_message(
     message: RabbitMQDemoMessage,
     rabbitmq: GetRabbitMQ,
-) -> dict[str, Any]:
+) ->  None:
     await rabbitmq.send_demo_message(message)
-    return {
-        "message": "RabbitMQ message sent successfully",
-        "key": message.key,
-        "value": message.value,
-    }
+{% endif %}
+
+{% if cookiecutter.use_taskiq %}
+@router.post("/taskiq-kiq")
+async def kick_taskiq_message() -> None:
+    await tasks.demo_task.kiq(hello="hello taskiq", world="world taskiq")
+
+
+@router.post("/taskiq-scheduled")
+async def schedule_taskiq_message(delay_seconds: int = 10) -> None:
+    await tasks.demo_task.schedule_by_time(
+        redis_source,
+        datetime.now(UTC) + timedelta(seconds=delay_seconds),
+        hello="hello taskiq scheduled",
+        world="world taskiq scheduled",
+    )
 {% endif %}