lablink_allocator_service.db.pool¶

`lablink_allocator_service.db.pool` ¶

Connection-pool primitives shared by every class in this package.

Kept separate from the persistence classes so that importing pool machinery never pulls in a module that opens connections or defines table-specific SQL.

Attributes¶

`POOL_MAX_SIZE = _pool_max_size_from_env(default=200)` `module-attribute` ¶

`POOL_MIN_SIZE = 2` `module-attribute` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

Classes¶

`PooledCursor(pool, dict_rows=False)` ¶

Checks out an autocommit connection from the pool, opens a cursor, and returns both to the pool/closes on exit. Preserves the per-call context-manager API previously provided by _LockedCursor.

dict_rows=True opens a RealDictCursor, so rows come back keyed by the database's own column names instead of as positional tuples — schema drift then fails loudly as a KeyError at the access instead of silently shifting values into the wrong keys. Opt-in because VmDatabase reads rows positionally throughout.

Source code in packages/allocator/src/lablink_allocator_service/db/pool.py

def __init__(self, pool, dict_rows: bool = False):
    self._pool = pool
    self._dict_rows = dict_rows
    self._conn = None
    self._cur = None

Functions:¶

`make_pool(password, *, pool_min_size=POOL_MIN_SIZE, pool_max_size=POOL_MAX_SIZE)` ¶

Build a ThreadedConnectionPool for the fixed allocator database.

For callers that need ONLY a pool and no persistence class — e.g. the APScheduler job in scheduler.py, which shares one pool across several handles. The pool is built with THIS module's psycopg2 binding; a caller that needs the binding to be its own (see validate_pool_sizes) should construct the pool itself rather than calling this.

Raises:

Type	Description
`ValueError`	If pool sizing is invalid.

Source code in packages/allocator/src/lablink_allocator_service/db/pool.py

def make_pool(password, *, pool_min_size=POOL_MIN_SIZE, pool_max_size=POOL_MAX_SIZE):
    """Build a ThreadedConnectionPool for the fixed allocator database.

    For callers that need ONLY a pool and no persistence class — e.g. the
    APScheduler job in scheduler.py, which shares one pool across several
    handles. The pool is built with THIS module's psycopg2 binding; a caller
    that needs the binding to be its own (see validate_pool_sizes) should
    construct the pool itself rather than calling this.

    Raises:
        ValueError: If pool sizing is invalid.
    """
    validate_pool_sizes(pool_min_size, pool_max_size)
    return psycopg2.pool.ThreadedConnectionPool(
        minconn=pool_min_size,
        maxconn=pool_max_size,
        dbname=DB_NAME,
        user=DB_USER,
        password=password,
        host=DB_HOST,
        port=DB_PORT,
    )

`validate_pool_sizes(pool_min_size, pool_max_size)` ¶

Reject invalid pool sizing. Pure — deliberately touches no psycopg2.

Callers that build their own pool (VmDatabase.init) use this instead of make_pool, so that pool CONSTRUCTION stays inside the caller's module. A module-level import psycopg2 binds the module object at import time, so whichever module actually builds the pool is the one whose binding of psycopg2 gets exercised. Keeping construction in VmDatabase rather than delegating it to a helper here keeps that binding local to VmDatabase — and therefore substitutable per-instance, e.g. by dependency-injecting a pre-built pool via the pool constructor argument instead of needing to intercept psycopg2 itself.

Raises:

Type	Description
`ValueError`	If pool sizing is invalid.

Source code in packages/allocator/src/lablink_allocator_service/db/pool.py

def validate_pool_sizes(pool_min_size: int, pool_max_size: int) -> None:
    """Reject invalid pool sizing. Pure — deliberately touches no psycopg2.

    Callers that build their own pool (VmDatabase.__init__) use this instead
    of make_pool, so that pool CONSTRUCTION stays inside the caller's module.
    A module-level `import psycopg2` binds the module object at import time,
    so whichever module actually builds the pool is the one whose binding of
    psycopg2 gets exercised. Keeping construction in VmDatabase rather than
    delegating it to a helper here keeps that binding local to VmDatabase —
    and therefore substitutable per-instance, e.g. by dependency-injecting a
    pre-built pool via the `pool` constructor argument instead of needing to
    intercept psycopg2 itself.

    Raises:
        ValueError: If pool sizing is invalid.
    """
    if pool_min_size < 1 or pool_max_size < pool_min_size:
        raise ValueError(
            f"Invalid pool sizes: min={pool_min_size}, max={pool_max_size}"
        )

lablink_allocator_service.db.pool¶