feat(plugin): add saq

src/Litestar/App/app.py.jinja

@@ -17,6 +17,9 @@ from litestar_vite import ViteConfig, VitePlugin
 {% if has_store %}
 from .config import redis_store
 {% endif %}
+{% if litestar_saq %}
+from .config import saq
+{% endif %}
 
 @get("/health")
 async def health() -> dict[str, str]:
@@ -52,11 +55,15 @@ app = Litestar(
         RepositoryError: exception_handler,
     },
 {% endif %}
-{% if advanced_alchemy and litestar_vite %}
-    plugins=[alchemy, VitePlugin(config=ViteConfig(dev_mode=settings.DEBUG))],
-{% elif advanced_alchemy %}
-    plugins=[alchemy],
-{% elif litestar_vite %}
-    plugins=[VitePlugin(config=ViteConfig(dev_mode=settings.DEBUG))],
+    plugins=[
+{% if advanced_alchemy %}
+        alchemy,
+{% endif %}
+{% if litestar_vite %}
+        VitePlugin(config=ViteConfig(dev_mode=settings.DEBUG)),
+{% endif %}
+{% if litestar_saq %}
+        saq,
 {% endif %}
+    ],
 )

src/Litestar/App/config.py.jinja

@@ -12,6 +12,9 @@ from litestar_vite import ViteConfig, PathConfig
 {% if has_store %}
 from litestar.stores.redis import RedisStore
 {% endif %}
+{% if litestar_saq %}
+from litestar_saq import QueueConfig, SAQConfig, SAQPlugin
+{% endif %}
 
 from .settings import get_settings
 
@@ -48,3 +51,14 @@ redis_store = RedisStore.with_client(
     db=0,
 )
 {% endif %}
+
+{% if litestar_saq %}
+saq = SAQPlugin(
+    config=SAQConfig(
+        use_server_lifespan=True,
+        queue_configs=[
+            QueueConfig(name="default", dsn=settings.REDIS_URL),
+        ],
+    )
+)
+{% endif %}

src/Litestar/Config/pyproject.toml.jinja

@@ -23,6 +23,9 @@ dependencies = [
 {% if has_store %}
     "redis>=5.2.1",
 {% endif %}
+{% if litestar_saq %}
+    "litestar-saq>=0.7.0",
+{% endif %}
 ]
 
 [tool.ruff]

src/Litestar/Plugins/LitestarSAQ/Templates/lib/tasks.py.jinja

@@ -0,0 +1,22 @@
+"""Background tasks."""
+
+import asyncio
+import logging
+from saq.types import Context
+
+logger = logging.getLogger(__name__)
+
+
+async def sample_task(ctx: Context) -> dict:
+    """A sample background task.
+
+    Args:
+        ctx: The SAQ job context.
+
+    Returns:
+        A dictionary with the task result.
+
+    """
+    logger.info("Running sample background task")
+    await asyncio.sleep(1)
+    return {"status": "complete"}

src/Litestar/Plugins/LitestarSAQ/__init__.py

@@ -1,3 +1,4 @@
+from src.models import MemoryStore, ProjectConfig
 from src.plugin import BasePlugin
 
 
@@ -11,3 +12,6 @@ def name(self) -> str:  # noqa: D102
     @property
     def description(self) -> str:  # noqa: D102
         return "SAQ integration for background tasks in Litestar"
+
+    def is_applicable(self, config: ProjectConfig) -> bool:  # noqa: D102, PLR6301
+        return config.memory_store != MemoryStore.NONE

tests/test_generator.py

@@ -124,3 +124,62 @@ def test_litestar_generator_docker_compose_rendering(tmp_path: Path) -> None:
     # Check services
     assert "image: postgres:17.7-alpine3.23" in content
     assert "image: redis:8.4.0-bookworm" in content
+
+
+def test_litestar_generator_saq_context(tmp_path: Path) -> None:
+    """Verify Litestar generator template context values with SAQ plugin enabled."""
+    config = ProjectConfig(
+        name="SAQ Test",
+        framework=Framework.LITESTAR,
+        database=Database.NONE,
+        memory_store=MemoryStore.REDIS,
+        plugins=["litestar_saq"],
+        docker=False,
+        docker_infra=False,
+    )
+
+    generator = LitestarGenerator(config, tmp_path)
+    context = generator._get_template_context()
+
+    assert context["litestar_saq"] is True
+    assert context["has_store"] is True
+    assert context["memory_store"] == MemoryStore.REDIS
+
+
+def test_litestar_generator_saq_rendering(tmp_path: Path) -> None:
+    """Verify that SAQ plugin templates are correctly rendered into the output directory."""
+    config = ProjectConfig(
+        name="SAQ Plugin Test",
+        framework=Framework.LITESTAR,
+        database=Database.NONE,
+        memory_store=MemoryStore.REDIS,
+        plugins=["litestar_saq"],
+        docker=False,
+        docker_infra=False,
+    )
+
+    generator = LitestarGenerator(config, tmp_path)
+    generator.generate()
+
+    # Verify base files
+    assert (tmp_path / "pyproject.toml").exists()
+    assert (tmp_path / "src" / "backend" / "app.py").exists()
+    assert (tmp_path / "src" / "backend" / "config.py").exists()
+
+    # Verify SAQ plugin files
+    assert (tmp_path / "src" / "backend" / "lib" / "tasks.py").exists()
+
+    # Verify SAQ config in config.py
+    config_content = (tmp_path / "src" / "backend" / "config.py").read_text()
+    assert "from litestar_saq import QueueConfig, SAQConfig, SAQPlugin" in config_content
+    assert "saq = SAQPlugin(" in config_content
+    assert 'QueueConfig(name="default"' in config_content
+
+    # Verify SAQ plugin in app.py
+    app_content = (tmp_path / "src" / "backend" / "app.py").read_text()
+    assert "from .config import saq" in app_content
+    assert "saq," in app_content
+
+    # Verify SAQ dependency in pyproject.toml
+    pyproject_content = (tmp_path / "pyproject.toml").read_text()
+    assert "litestar-saq>=0.7.0" in pyproject_content