symfonic.memory.backends.pool

pool

PostgresPoolManager -- asyncpg connection pool wrapper.

Provides a reusable async context manager around asyncpg's Pool. Shared by both PostgresGraphBackend and PostgresVectorBackend.

Optional dependency: asyncpg must be installed.

pgvector codec registration (v7.3, Item 13.3)

Every physical connection created by the pool runs the pgvector asyncpg codec registration in its init=... callback. Once registered, asyncpg returns vector columns as pgvector.Vector instances (iterable of floats) instead of the slower text form "[0.1,0.2]".

The graph backend's _decode_embedding accepts BOTH shapes — text form AND iterable — so this PR can land without breaking the 7.2.1 hotfix. When pgvector is not importable (degraded mode) the init callback logs once and the text-form path keeps working.

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)

connection_has_pgvector_codec

connection_has_pgvector_codec(conn: Any) -> bool

Return True if conn had pgvector's codec successfully registered.

Set by :func:_register_pgvector_on_connection per physical connection. Defaults to False when the connection has not been seen by the init hook — e.g. on mock connections in unit tests, or on a connection that was not created by a pool we control.

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

def connection_has_pgvector_codec(conn: Any) -> bool:
    """Return True if ``conn`` had pgvector's codec successfully registered.

    Set by :func:`_register_pgvector_on_connection` per physical
    connection. Defaults to ``False`` when the connection has not been
    seen by the init hook — e.g. on mock connections in unit tests, or
    on a connection that was not created by a pool we control.
    """
    raw = _raw_connection(conn)
    try:
        return bool(_CODEC_REGISTRY.get(raw, False))
    except TypeError:  # raw is not hashable (mocks, etc.) — degrade safely
        return False