symfonic.core.prompt.blocks.sources.database_sql

database_sql

Append-only guards, scope keys and dialect rewriting for the block schema.

The executable half of :mod:symfonic.core.prompt.blocks.sources.database_schema: that module declares the table and the statements, this one holds the rules those statements must satisfy and the small helpers a caller needs to use them safely. Splitting them keeps the SQL surface readable as a surface, and keeps these checks importable without pulling the statement list along.

Nothing here reads or writes a database. :func:assert_append_only is run at import by the schema module against its own ALL_STATEMENTS, so a statement added later is either append-only or the package fails to import.

assert_append_only

assert_append_only(statements: Iterable[str]) -> None

Raise unless every statement leaves existing revisions untouched.

Run against :data:ALL_STATEMENTS at import, so a mutating statement added to this module fails the import that introduced it rather than the first production write.

Source code in src/symfonic/core/prompt/blocks/sources/database_sql.py

def assert_append_only(statements: Iterable[str]) -> None:
    """Raise unless every statement leaves existing revisions untouched.

    Run against :data:`ALL_STATEMENTS` at import, so a mutating statement
    added to this module fails the import that introduced it rather than
    the first production write.
    """
    offenders = mutating_statements(statements)
    if offenders:
        raise AssertionError(
            "the block revision schema is append-only, but these statements would "
            f"change or remove existing rows: {offenders!r}"
        )

make_revision_id

make_revision_id(sequence: int, content_hash: str) -> str

Return a revision identifier unique within one block and scope.

Uniqueness rides on sequence, which the primary key already enforces per (scope_path, block_id). The digest tail is carried for legibility only -- it is what makes r7-9f2c... recognisable in a log -- and must never be the sole identity: restoring an earlier revision appends the same content, so the digest repeats by design.

Source code in src/symfonic/core/prompt/blocks/sources/database_sql.py

def make_revision_id(sequence: int, content_hash: str) -> str:
    """Return a revision identifier unique within one block and scope.

    Uniqueness rides on ``sequence``, which the primary key already
    enforces per ``(scope_path, block_id)``. The digest tail is carried
    for legibility only -- it is what makes ``r7-9f2c...`` recognisable
    in a log -- and must never be the sole identity: restoring an earlier
    revision appends the same content, so the digest repeats by design.
    """
    if sequence < 1:
        raise ValueError(f"revision sequence must be >= 1; got {sequence}")
    digest = content_hash.rsplit(":", 1)[-1]
    if not digest:
        raise ValueError(
            f"content_hash {content_hash!r} carries no digest to build a revision id from"
        )
    return f"r{sequence}-{digest[:16]}"

mutating_statements

mutating_statements(statements: Iterable[str]) -> tuple[str, ...]

Return every statement that would change or remove an existing row.

DROP TABLE is not counted: dropping the table is the declared reverse migration (:data:MIGRATION_DOWN), and it removes the schema rather than editing a revision inside it.

That exemption is unqualified rather than scoped to :data:MIGRATION_DOWN's own statement -- a DROP TABLE reached from anywhere else in the package (a reset() helper, say) would pass this check too. It is not scoped narrower because this checker also runs, unparametrised, against every SQL-looking string literal in the package's source (including partial fragments an f-string splits across ast.Constant nodes, which never equal a complete MIGRATION_DOWN statement); a membership test against :data:MIGRATION_DOWN would silently stop matching those fragments and reintroduce the exact gap ALTER TABLE was added above to close. Narrowing this safely needs the fragment-scanning test to cooperate, not just this function.

Source code in src/symfonic/core/prompt/blocks/sources/database_sql.py

def mutating_statements(statements: Iterable[str]) -> tuple[str, ...]:
    """Return every statement that would change or remove an existing row.

    ``DROP TABLE`` is not counted: dropping the table is the declared
    reverse migration (:data:`MIGRATION_DOWN`), and it removes the schema
    rather than editing a revision inside it.

    That exemption is unqualified rather than scoped to
    :data:`MIGRATION_DOWN`'s own statement -- a ``DROP TABLE`` reached
    from anywhere else in the package (a ``reset()`` helper, say) would
    pass this check too. It is not scoped narrower because this checker
    also runs, unparametrised, against every SQL-looking string literal in
    the package's source (including partial fragments an f-string splits
    across ``ast.Constant`` nodes, which never equal a complete
    ``MIGRATION_DOWN`` statement); a membership test against
    :data:`MIGRATION_DOWN` would silently stop matching those fragments
    and reintroduce the exact gap ``ALTER TABLE`` was added above to
    close. Narrowing this safely needs the fragment-scanning test to
    cooperate, not just this function.
    """
    return tuple(sql for sql in statements if _MUTATING_SQL.search(sql))

qmark

qmark(sql: str) -> str

Translate $1-style placeholders into the qmark paramstyle.

The statements are written once, in the style asyncpg speaks. A driver using qmark (sqlite3, and the schema's own conformance tests) runs the same text through this rather than keeping a second, silently divergent copy.

Positional rewriting is only sound when the placeholders appear in ascending order and each is used once, so that is checked rather than assumed.

Source code in src/symfonic/core/prompt/blocks/sources/database_sql.py

def qmark(sql: str) -> str:
    """Translate ``$1``-style placeholders into the qmark paramstyle.

    The statements are written once, in the style asyncpg speaks. A
    driver using qmark (``sqlite3``, and the schema's own conformance
    tests) runs the same text through this rather than keeping a second,
    silently divergent copy.

    Positional rewriting is only sound when the placeholders appear in
    ascending order and each is used once, so that is checked rather than
    assumed.
    """
    numbers = [int(n) for n in _PLACEHOLDER.findall(sql)]
    if numbers != list(range(1, len(numbers) + 1)):
        raise ValueError(
            f"cannot rewrite placeholders positionally: {sql!r} uses them out of order "
            "or more than once, so a qmark driver would bind different values"
        )
    return _PLACEHOLDER.sub("?", sql)

revision_key

revision_key(scope: TenantScope, block_id: str) -> tuple[str, str]

Return the (scope_path, block_id) values to bind into a statement.

Delegates to :func:~symfonic.core.prompt.blocks.protocol.block_isolation_key so the value stored in the scope_path column is exactly :attr:TenantScope.scope_path -- the same string the block layer keys its caches on. Deriving it here independently would be one more place for the canonical representation to drift.

Source code in src/symfonic/core/prompt/blocks/sources/database_sql.py

def revision_key(scope: TenantScope, block_id: str) -> tuple[str, str]:
    """Return the ``(scope_path, block_id)`` values to bind into a statement.

    Delegates to
    :func:`~symfonic.core.prompt.blocks.protocol.block_isolation_key` so
    the value stored in the ``scope_path`` column is exactly
    :attr:`TenantScope.scope_path` -- the same string the block layer
    keys its caches on. Deriving it here independently would be one more
    place for the canonical representation to drift.
    """
    return block_isolation_key(scope, block_id)