symfonic.memory.backends.postgres_conversation_export

postgres_conversation_export

Privacy export of a conversation scope and its descendants.

export async

export(pool: Any, scope: MemoryScope) -> tuple[dict[str, Any], ...]

Rows owned by scope or its descendants.

Source code in src/symfonic/memory/backends/postgres_conversation_export.py

async def export(pool: Any, scope: MemoryScope) -> tuple[dict[str, Any], ...]:
    """Rows owned by ``scope`` or its descendants."""
    if scope.identity_version not in (0, 1):
        raise MemoryContractError("unsupported conversation scope identity version")
    conn = await pool.acquire()
    try:
        if scope.identity_version == 1:
            rows = await conn.fetch(_EXPORT_NATIVE, scope.path, native_child_prefix(scope))
            rows = [row for row in rows if native_covers(scope, row["scope_path"])]
        else:
            rows = await conn.fetch(_EXPORT, scope.path)
    except Exception as exc:  # noqa: BLE001
        raise MemoryUnavailable("conversation history could not be exported") from exc
    finally:
        await pool.release(conn)
    return tuple(dict(row) for row in rows)

native_child_prefix

native_child_prefix(scope: MemoryScope) -> str

End at a complete level and the next pair separator.

Source code in src/symfonic/memory/backends/postgres_conversation_export.py

def native_child_prefix(scope: MemoryScope) -> str:
    """End at a complete level and the next pair separator."""
    return scope.path[:-2] + ","

native_covers

native_covers(scope: MemoryScope, path: str) -> bool

A lexical candidate counts only if its full stored path is canonical.

Source code in src/symfonic/memory/backends/postgres_conversation_export.py

def native_covers(scope: MemoryScope, path: str) -> bool:
    """A lexical candidate counts only if its full stored path is canonical."""
    try:
        return scope.covers(scope_from_path(path))
    except MemoryContractError:
        return False