symfonic.agent.prompts.db_section

db_section

DBPromptSection -- versioned Jinja2 templates from PostgreSQL with Redis caching.

Conforms to the PromptSection protocol from symfonic.core.prompt. Unlike the HMS section (which uses {{VARIABLE}} plain substitution), this section renders through Jinja2's SandboxedEnvironment enabling conditionals, loops, and filters.

Render pipeline

1. Check Redis cache for section content
2. If miss, query prompt_section_versions for the enabled version
3. Render content through Jinja2 SandboxedEnvironment
4. If any step fails, render fallback_content through Jinja2

DBPromptSection dataclass

DBPromptSection(
    name: str,
    required: bool = False,
    priority: int = 50,
    cache_breakpoint: bool = False,
    fallback_content: str = "",
    section_key: str = "",
    cache_ttl_seconds: int = 300,
)

PromptSection that loads versioned Jinja2 templates from PostgreSQL.

The state dict is passed as Jinja2 template context, enabling: - {{ name }} variable substitution - {% if user_is_premium %}...{% endif %} conditionals - {% for item in items %}...{% endfor %} loops - {{ name | upper }} filters

Gracefully degrades when Redis or DB are unavailable, falling back to fallback_content rendered through the same Jinja2 pipeline.

Attributes:

Name	Type	Description
name	str	Section display name used by PromptBuilder.
required	bool	Whether a render failure should abort the build.
priority	int	Lower numbers appear earlier in the assembled prompt.
fallback_content	str	Jinja2 template string used when DB/cache are unavailable.
section_key	str	DB lookup key; defaults to name if empty.
cache_ttl_seconds	int	Redis TTL for cached template content (default 300s).

configure

configure(
    redis_client: Any = None, db_session_factory: Any = None
) -> DBPromptSection

Inject optional Redis and DB dependencies. Returns self for chaining.

Source code in src/symfonic/agent/prompts/db_section.py

def configure(
    self,
    redis_client: Any = None,
    db_session_factory: Any = None,
) -> DBPromptSection:
    """Inject optional Redis and DB dependencies. Returns self for chaining."""
    self._redis_client = redis_client
    self._db_session_factory = db_session_factory
    return self

render async

render(state: PromptState) -> str

Render the section: cache -> DB -> fallback, all through Jinja2.

Source code in src/symfonic/agent/prompts/db_section.py

async def render(self, state: PromptState) -> str:
    """Render the section: cache -> DB -> fallback, all through Jinja2."""
    key = self.section_key or self.name
    raw_template: str | None = None

    # 1. Try Redis cache
    raw_template = await self._try_cache(key)

    # 2. Try DB
    if raw_template is None:
        raw_template = await self._try_db(key)
        if raw_template is not None:
            await self._set_cache(key, raw_template)

    # 3. Fallback
    if raw_template is None:
        raw_template = self.fallback_content

    if not raw_template:
        return ""

    return self._render_jinja2(raw_template, dict(state))