symfonic.memory.backends

backends

Backend implementations for graph and vector storage.

InMemoryGraphBackend

InMemoryGraphBackend()

In-memory implementation of the GraphBackend protocol.

Stores nodes and edges in dicts keyed by (tenant_id, node_id/edge_id). Suitable for testing and local development only.

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

def __init__(self) -> None:
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)

add_edge async

add_edge(
    scope: TenantScope, edge: MemoryEdge
) -> MemoryEdge

Add an edge to the in-memory store.

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

async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the in-memory store."""
    edge.tenant_id = scope.tenant_id
    self._edges[scope.tenant_id][str(edge.id)] = edge
    return edge

add_node async

add_node(
    scope: TenantScope, node: MemoryNode
) -> MemoryNode

Add a node to the in-memory store.

v8.0: stamps the materialised scope_path into the node's property bag (zero-migration) so the prefix-isolation filter (§5.d) can be enforced on every read.

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

async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the in-memory store.

    v8.0: stamps the materialised ``scope_path`` into the node's
    property bag (zero-migration) so the prefix-isolation filter (§5.d)
    can be enforced on every read.
    """
    node.tenant_id = scope.tenant_id
    node.properties[SCOPE_PATH_KEY] = materialise_scope_path(scope)
    self._nodes[scope.tenant_id][str(node.id)] = node
    return node

delete_edge async

delete_edge(scope: TenantScope, edge_id: EdgeId) -> None

Delete a single edge by its ID.

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

async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_edges.pop(str(edge_id), None)

delete_node async

delete_node(
    scope: TenantScope,
    node_id: NodeId,
    cascade: bool = False,
) -> None

Delete a node, optionally cascading to connected edges.

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

async def delete_node(
    self, scope: TenantScope, node_id: NodeId, cascade: bool = False
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        return
    # v8.7.7: scope-gate deletes by prefix visibility so a sibling sub-scope
    # cannot delete another sub-scope's node (mirrors the PG delete_node and
    # get_node read filter; previously tenant-only). Round-2 fix.
    stored = stored_scope_path(node.properties, scope.tenant_id)
    if not is_visible(stored, scope.ancestor_prefix_paths()):
        return
    tenant_nodes.pop(str(node_id), None)
    if cascade:
        tenant_edges = self._edges.get(scope.tenant_id, {})
        to_remove = [
            eid
            for eid, edge in tenant_edges.items()
            if str(edge.source) == str(node_id)
            or str(edge.target) == str(node_id)
        ]
        for eid in to_remove:
            del tenant_edges[eid]

get_neighbors async

get_neighbors(
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]

Get nodes connected to the given node by edges.

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

async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Get nodes connected to the given node by edges."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    neighbor_ids: set[str] = set()
    for edge in tenant_edges.values():
        if relationship and edge.relationship != relationship:
            continue
        if str(edge.source) == str(node_id):
            neighbor_ids.add(str(edge.target))
        elif str(edge.target) == str(node_id):
            neighbor_ids.add(str(edge.source))
    return [
        tenant_nodes[nid]
        for nid in neighbor_ids
        if nid in tenant_nodes and self._node_visible(scope, tenant_nodes[nid])
    ]

get_node async

get_node(
    scope: TenantScope, node_id: NodeId
) -> MemoryNode | None

Get a node by ID within the scope's visible prefix chain.

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

async def get_node(
    self, scope: TenantScope, node_id: NodeId
) -> MemoryNode | None:
    """Get a node by ID within the scope's visible prefix chain."""
    node = self._nodes.get(scope.tenant_id, {}).get(str(node_id))
    if node is None or not self._node_visible(scope, node):
        return None
    return node

query_edges async

query_edges(
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]

Query edges with optional relationship filter and pagination.

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

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges with optional relationship filter and pagination."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    relationship = (filters or {}).get("relationship")

    results: list[MemoryEdge] = []
    for edge in tenant_edges.values():
        if relationship is not None and edge.relationship != relationship:
            continue
        results.append(edge)

    # Stable ordering: newest first (mirrors Postgres behaviour)
    results.sort(key=lambda e: e.created_at, reverse=True)
    return results[offset : offset + limit]

query_nodes async

query_nodes(
    scope: TenantScope,
    filters: dict[str, Any],
    limit: int | None = 50,
) -> list[MemoryNode]

Filter nodes by properties within the tenant scope.

limit=None returns every matching node (no truncation); limit=0 returns no rows (v8.7.1 — unified across backends).

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

async def query_nodes(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = 50
) -> list[MemoryNode]:
    """Filter nodes by properties within the tenant scope.

    ``limit=None`` returns every matching node (no truncation);
    ``limit=0`` returns no rows (v8.7.1 — unified across backends).
    """
    if limit == 0:
        return []
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    results: list[MemoryNode] = []
    for node in tenant_nodes.values():
        if not self._node_visible(scope, node):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

traverse async

traverse(
    scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]

BFS traversal from start node up to max_depth.

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

async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth."""
    visited: set[str] = set()
    queue: list[tuple[str, int]] = [(str(start), 0)]
    result: list[MemoryNode] = []
    tenant_nodes = self._nodes.get(scope.tenant_id, {})

    while queue:
        current_id, depth = queue.pop(0)
        if current_id in visited:
            continue
        visited.add(current_id)
        node = tenant_nodes.get(current_id)
        if node is not None and self._node_visible(scope, node):
            result.append(node)
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for neighbor in neighbors:
                if str(neighbor.id) not in visited:
                    queue.append((str(neighbor.id), depth + 1))
    return result

update_node async

update_node(
    scope: TenantScope,
    node_id: NodeId,
    updates: dict[str, Any],
) -> MemoryNode

Partial update of node properties.

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

async def update_node(
    self, scope: TenantScope, node_id: NodeId, updates: dict[str, Any]
) -> MemoryNode:
    """Partial update of node properties."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    for key, value in updates.items():
        if key == "updated_at" and isinstance(value, str):
            value = datetime.fromisoformat(value)
        if hasattr(node, key):
            object.__setattr__(node, key, value)
        else:
            node.properties[key] = value
    if "updated_at" not in updates:
        node.updated_at = datetime.now(UTC)
    return node

upsert_edge async

upsert_edge(
    scope: TenantScope, edge: MemoryEdge
) -> MemoryEdge

Insert edge or increment weight by 1 if (tenant, source, target, rel) matches.

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

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if (tenant, source, target, rel) matches."""
    edge.tenant_id = scope.tenant_id
    tenant_edges = self._edges[scope.tenant_id]
    for existing in tenant_edges.values():
        if (
            str(existing.source) == str(edge.source)
            and str(existing.target) == str(edge.target)
            and existing.relationship == edge.relationship
        ):
            object.__setattr__(existing, "weight", existing.weight + 1)
            return existing
    tenant_edges[str(edge.id)] = edge
    return edge

InMemoryVectorBackend

InMemoryVectorBackend()

In-memory implementation of the VectorBackend protocol.

Stores embeddings in a list and uses brute-force cosine similarity for search. Suitable for testing only.

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

def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async

add(
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None

Add vectors with metadata to the store.

v8.0: stamps the materialised scope_path into the metadata bag so the prefix-isolation filter (§5.d) can run on every search.

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

async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with metadata to the store.

    v8.0: stamps the materialised ``scope_path`` into the metadata bag so
    the prefix-isolation filter (§5.d) can run on every search.
    """
    scope_path = materialise_scope_path(scope)
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        meta.setdefault(SCOPE_PATH_KEY, scope_path)
        self._store[scope.tenant_id].append({
            "id": doc_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
        })

count async

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

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

async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    return len(self._store.get(scope.tenant_id, []))

delete async

delete(scope: TenantScope, ids: list[str]) -> None

Delete vectors by their IDs.

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

async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    id_set = set(ids)
    if scope.tenant_id in self._store:
        self._store[scope.tenant_id] = [
            e for e in self._store[scope.tenant_id] if e["id"] not in id_set
        ]

search async

search(
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]

Brute-force cosine similarity search.

v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored scope_path is a prefix of the query path are scored. Dual-reads pre-v8.0 metadata (no scope_path) as a 1-level root path.

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

async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Brute-force cosine similarity search.

    v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored
    ``scope_path`` is a prefix of the query path are scored.  Dual-reads
    pre-v8.0 metadata (no ``scope_path``) as a 1-level root path.
    """
    entries = self._store.get(scope.tenant_id, [])
    prefixes = scope.ancestor_prefix_paths()
    scored = []
    for entry in entries:
        entry_path = metadata_scope_path(entry["metadata"], scope.tenant_id)
        if not is_visible(entry_path, prefixes):
            continue
        sim = _cosine_similarity(query_embedding, entry["embedding"])
        scored.append({
            "id": entry["id"],
            "score": sim,
            "metadata": entry["metadata"],
            "document": entry["document"],
        })
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]

PostgresGraphBackend

PostgresGraphBackend(pool: PostgresPoolManager)

GraphBackend backed by PostgreSQL with JSONB storage.

All operations enforce tenant isolation via WHERE tenant_id = $N.

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

def __init__(self, pool: PostgresPoolManager) -> None:
    _require_asyncpg("PostgresGraphBackend")
    self._pool = pool

backfill_scope_paths async

backfill_scope_paths() -> int

Eagerly backfill NULL scope_path rows (design §6.b migration).

Pre-v8.0 rows have scope_path IS NULL; they read as a 1-level root path tenant\x1f<tenant_id> via the query-time dual-read, so this eager backfill is OPTIONAL (online-safe, idempotent) — it just materialises the same value into the column so the B-tree index and the scorer's dual-read agree. Returns the number of rows updated. Uses a parameterised expression so adopter tenant_ids cannot inject.

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

async def backfill_scope_paths(self) -> int:
    """Eagerly backfill NULL ``scope_path`` rows (design §6.b migration).

    Pre-v8.0 rows have ``scope_path IS NULL``; they read as a 1-level root
    path ``tenant\\x1f<tenant_id>`` via the query-time dual-read, so this
    eager backfill is OPTIONAL (online-safe, idempotent) — it just
    materialises the same value into the column so the B-tree index and
    the scorer's dual-read agree.  Returns the number of rows updated.
    Uses a parameterised expression so adopter tenant_ids cannot inject.
    """
    # The 1-level root path is f"tenant\x1f{tenant_id}"; build it in SQL.
    delim = "\x1f"
    conn = await self._pool.acquire()
    try:
        result = await conn.execute(
            "UPDATE memory_nodes "
            "SET scope_path = 'tenant' || $1 || tenant_id "
            "WHERE scope_path IS NULL",
            delim,
        )
    finally:
        await self._pool.release(conn)
    # asyncpg returns e.g. "UPDATE 42"
    try:
        return int(str(result).split()[-1])
    except (ValueError, IndexError):
        return 0

ensure_schema async

ensure_schema() -> None

Create graph tables (nodes + edges) and their indexes.

Does NOT create the memory_vectors table — that is the responsibility of :class:PostgresVectorBackend.ensure_schema.

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

async def ensure_schema(self) -> None:
    """Create graph tables (nodes + edges) and their indexes.

    Does NOT create the ``memory_vectors`` table — that is the
    responsibility of :class:`PostgresVectorBackend.ensure_schema`.
    """
    conn = await self._pool.acquire()
    try:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(NODES_DDL)
        await conn.execute(EDGES_DDL)
        # v8.0 online-safe upgrade: add scope_path to pre-existing tables.
        await conn.execute(NODES_SCOPE_PATH_MIGRATION)
        # Online-safe upgrade: add edge model fields to pre-existing tables.
        await conn.execute(EDGES_FIELDS_MIGRATION)
        for idx in GRAPH_INDEXES_DDL:
            await conn.execute(idx)
    finally:
        await self._pool.release(conn)

query_edges async

query_edges(
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]

Query edges directly from memory_edges table with pagination.

Supports optional relationship filter. All other keys in filters are ignored to avoid SQL injection risk.

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

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges directly from memory_edges table with pagination.

    Supports optional ``relationship`` filter. All other keys in ``filters``
    are ignored to avoid SQL injection risk.
    """
    conditions: list[str] = ["tenant_id=$1"]
    params: list[Any] = [scope.tenant_id]

    relationship = (filters or {}).get("relationship")
    if relationship is not None:
        params.append(relationship)
        conditions.append(f"relationship=${len(params)}")

    params.append(limit)
    limit_ph = f"${len(params)}"
    params.append(offset)
    offset_ph = f"${len(params)}"

    sql = (
        f"SELECT * FROM memory_edges WHERE {' AND '.join(conditions)}"
        f" ORDER BY created_at DESC"
        f" LIMIT {limit_ph} OFFSET {offset_ph}"
    )

    conn = await self._pool.acquire()
    try:
        rows = await conn.fetch(sql, *params)
    finally:
        await self._pool.release(conn)

    return [_row_to_edge(row) for row in rows]

traverse async

traverse(
    scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]

BFS graph traversal up to max_depth hops from start node.

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

async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS graph traversal up to max_depth hops from start node."""
    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(str(start), 0)])
    result: list[MemoryNode] = []
    while queue:
        current_id, depth = queue.popleft()
        if current_id in visited:
            continue
        visited.add(current_id)
        node = await self.get_node(scope, NodeId(current_id))
        if node is not None:
            result.append(node)
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for nb in neighbors:
                if str(nb.id) not in visited:
                    queue.append((str(nb.id), depth + 1))
    return result

upsert_edge async

upsert_edge(
    scope: TenantScope, edge: MemoryEdge
) -> MemoryEdge

Insert edge or increment weight by 1 if the same edge already exists.

Uniqueness is determined by (tenant_id, source, target, relationship). The unique index idx_memory_edges_upsert must exist for ON CONFLICT to resolve correctly (created by ensure_schema).

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

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if the same edge already exists.

    Uniqueness is determined by (tenant_id, source, target, relationship).
    The unique index ``idx_memory_edges_upsert`` must exist for ON CONFLICT
    to resolve correctly (created by ``ensure_schema``).
    """
    edge.tenant_id = scope.tenant_id
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            """
            INSERT INTO memory_edges
                (id, source, target, relationship, tenant_id, weight,
                 created_at, uses, provenance, source_episodic_ids,
                 scope_path)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
            ON CONFLICT (tenant_id, source, target, relationship)
            DO UPDATE SET weight = memory_edges.weight + 1
            RETURNING *
            """,
            str(edge.id), str(edge.source), str(edge.target),
            edge.relationship, scope.tenant_id, edge.weight, edge.created_at,
            edge.uses, edge.provenance, json.dumps(edge.source_episodic_ids),
            materialise_scope_path(scope),
        )
    finally:
        await self._pool.release(conn)

    if row is None:  # pragma: no cover — RETURNING always yields a row
        return edge
    return _row_to_edge(row)

PostgresPoolManager

PostgresPoolManager(
    dsn: str, min_size: int = 2, max_size: int = 10
)

Async connection pool manager wrapping asyncpg.Pool.

Usage::

pool = PostgresPoolManager(dsn="postgresql://user:pw@host/db")
await pool.open()
async with pool as conn:
    await conn.execute("SELECT 1")
await pool.close()

Or as a context manager for the pool itself::

async with PostgresPoolManager(dsn=...) as pool:
    conn = await pool.acquire()
    try:
        await conn.execute("SELECT 1")
    finally:
        await pool.release(conn)

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

def __init__(
    self,
    dsn: str,
    min_size: int = 2,
    max_size: int = 10,
) -> None:
    if not _ASYNCPG_AVAILABLE:
        raise ImportError(
            "PostgresPoolManager requires asyncpg. "
            "Install it with: pip install symfonic-core[postgres]"
        )
    self._dsn = dsn
    self._min_size = min_size
    self._max_size = max_size
    self._pool: Any = None  # asyncpg.Pool at runtime

dsn property

dsn: str

Return the configured connection string (read-only).

Used by the v7.1.1 PostgresCheckpointerFactory to build a separate psycopg pool against the same database. The two pools intentionally do not share connections (asyncpg vs. psycopg protocols are incompatible inside langgraph-checkpoint-postgres).

raw_pool property

raw_pool: Any

Expose the underlying asyncpg.Pool for direct use (advanced).

acquire async

acquire() -> Any

Acquire a connection from the pool.

Caller is responsible for releasing via :meth:release.

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

async def acquire(self) -> Any:
    """Acquire a connection from the pool.

    Caller is responsible for releasing via :meth:`release`.
    """
    if self._pool is None:
        raise RuntimeError(
            "Pool is not open. Call await pool.open() first "
            "or use 'async with pool' context manager."
        )
    return await self._pool.acquire()

close async

close() -> None

Gracefully close all connections in the pool.

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

async def close(self) -> None:
    """Gracefully close all connections in the pool."""
    if self._pool is not None:
        await self._pool.close()
        self._pool = None

open async

open() -> None

Create the underlying asyncpg connection pool.

Passes init=_register_pgvector_on_connection so every new physical connection registers the pgvector binary codec before the pool hands it out. See module docstring for the rationale.

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

async def open(self) -> None:
    """Create the underlying asyncpg connection pool.

    Passes ``init=_register_pgvector_on_connection`` so every new
    physical connection registers the pgvector binary codec before
    the pool hands it out. See module docstring for the rationale.
    """
    if self._pool is not None:
        return
    self._pool = await asyncpg.create_pool(
        self._dsn,
        min_size=self._min_size,
        max_size=self._max_size,
        init=_register_pgvector_on_connection,
    )

release async

release(conn: Any) -> None

Return a connection to the pool.

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

async def release(self, conn: Any) -> None:
    """Return a connection to the pool."""
    if self._pool is not None:
        await self._pool.release(conn)

PostgresVectorBackend

PostgresVectorBackend(
    pool: PostgresPoolManager, embedding_dim: int = 1536
)

VectorBackend backed by PostgreSQL + pgvector.

Uses cosine distance (<=> operator) for similarity search. All operations enforce tenant isolation via WHERE tenant_id = $N.

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

def __init__(self, pool: PostgresPoolManager, embedding_dim: int = 1536) -> None:
    _require_asyncpg("PostgresVectorBackend")
    self._pool = pool
    self._embedding_dim = embedding_dim

count async

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

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

async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT COUNT(*)::int AS n FROM memory_vectors WHERE tenant_id=$1",
            scope.tenant_id,
        )
        return int(row["n"]) if row else 0
    finally:
        await self._pool.release(conn)

ensure_schema async

ensure_schema() -> None

Create the memory_vectors table and indexes if not present.

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

async def ensure_schema(self) -> None:
    """Create the memory_vectors table and indexes if not present."""
    conn = await self._pool.acquire()
    try:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(VECTORS_DDL)
        await conn.execute(VECTORS_SCOPE_PATH_MIGRATION)
        for idx in VECTOR_INDEXES_DDL:
            await conn.execute(idx)
    finally:
        await self._pool.release(conn)