connection

connection

Connection pooling, transactions, and audit logging for ccr_metadata.

Public API

get_connection() -- context manager yielding a pooled mariadb.Connection transaction() -- context manager yielding an audit-logging cursor execute() -- one-shot query helper returning list[dict] init_pool() -- explicit pool configuration close_pool() -- shutdown / test teardown

Credentials are read from ~/.my.cnf by default. The section name and any individual fields can be overridden via init_pool(...) keyword arguments. The database name additionally honors the NOXDB_DATABASE env variable (env var loses to an explicit init_pool(database=...) override).

SSH tunneling

The production database lives on a Galera cluster inside the LiSC network and is only reachable by SSH-ing through the project VM at ccr-lab.lisc.univie.ac.at. To connect from outside LiSC, supply SSH parameters and init_pool() will open a local-port-forwarding tunnel before creating the pool. The DB host/port you configure are interpreted as the remote DB endpoint (i.e. the Galera cluster as seen from the VM).

SSH parameters (kwargs > NOXDB_SSH_* env vars > [noxdb-ssh] INI section): ssh_host, ssh_port (default 22), ssh_user, ssh_password, ssh_pkey, ssh_pkey_password.

If ssh_host is unset the tunnel is skipped and the driver connects directly to host:port (useful when running on the VM itself).

Write statements (INSERT/UPDATE/DELETE/REPLACE) issued via execute() or via the cursor yielded by transaction() are appended to an audit log at ~/.noxdb/audit.log (override with NOXDB_AUDIT_LOG).

init_pool

init_pool(pool_size: int = DEFAULT_POOL_SIZE, *, config_path: str | Path | None = DEFAULT_CONFIG_PATH, section: str = DEFAULT_SECTION, host: str | None = None, port: int | None = None, user: str | None = None, password: str | None = None, database: str | None = None, ssh_host: str | None = None, ssh_port: int | None = None, ssh_user: str | None = None, ssh_password: str | None = None, ssh_pkey: str | None = None, ssh_pkey_password: str | None = None) -> None

Create the connection pool.

When ssh_host resolves to a non-empty value (via kwarg, NOXDB_SSH_HOST, or the [noxdb-ssh] config section), an SSH tunnel is opened to that host and the pool connects through it; the configured DB host:port is the tunnel's remote bind target.

Parameters:

Name	Type	Description	Default
pool_size	int	Number of pooled connections.	DEFAULT_POOL_SIZE
config_path	str | Path | None	Path to the MariaDB-style config file. Pass None to skip the INI file entirely and rely solely on keyword overrides (useful for CI / tests).	DEFAULT_CONFIG_PATH
section	str	INI section to read DB credentials from.	DEFAULT_SECTION
host	str | None	DB host override. Wins over the config file.	None
port	int | None	DB port override.	None
user	str | None	DB user override.	None
password	str | None	DB password override.	None
database	str | None	DB name override. Also honours the NOXDB_DATABASE env var (env var loses to an explicit kwarg).	None
ssh_host	str | None	SSH jump host. When non-empty, a tunnel is opened.	None
ssh_port	int | None	SSH port (default 22).	None
ssh_user	str | None	SSH username.	None
ssh_password	str | None	SSH password (used if ssh_pkey not set).	None
ssh_pkey	str | None	Path to private key. Tried first for auth.	None
ssh_pkey_password	str | None	Passphrase for ssh_pkey, if any.	None

Raises:

Type	Description
RuntimeError	If the pool is already initialized (call close_pool first to reconfigure), or if SSH credentials are incomplete, or if required DB credentials are missing.
FileNotFoundError	If config_path is set but missing.

Source code in src/noxdb/connection.py

def init_pool(
    pool_size: int = DEFAULT_POOL_SIZE,
    *,
    config_path: str | Path | None = DEFAULT_CONFIG_PATH,
    section: str = DEFAULT_SECTION,
    host: str | None = None,
    port: int | None = None,
    user: str | None = None,
    password: str | None = None,
    database: str | None = None,
    ssh_host: str | None = None,
    ssh_port: int | None = None,
    ssh_user: str | None = None,
    ssh_password: str | None = None,
    ssh_pkey: str | None = None,
    ssh_pkey_password: str | None = None,
) -> None:
    """Create the connection pool.

    When ``ssh_host`` resolves to a non-empty value (via kwarg,
    ``NOXDB_SSH_HOST``, or the ``[noxdb-ssh]`` config section), an SSH
    tunnel is opened to that host and the pool connects through it;
    the configured DB ``host:port`` is the tunnel's remote bind target.

    Args:
        pool_size: Number of pooled connections.
        config_path: Path to the MariaDB-style config file. Pass
            ``None`` to skip the INI file entirely and rely solely on
            keyword overrides (useful for CI / tests).
        section: INI section to read DB credentials from.
        host: DB host override. Wins over the config file.
        port: DB port override.
        user: DB user override.
        password: DB password override.
        database: DB name override. Also honours the ``NOXDB_DATABASE``
            env var (env var loses to an explicit kwarg).
        ssh_host: SSH jump host. When non-empty, a tunnel is opened.
        ssh_port: SSH port (default 22).
        ssh_user: SSH username.
        ssh_password: SSH password (used if ``ssh_pkey`` not set).
        ssh_pkey: Path to private key. Tried first for auth.
        ssh_pkey_password: Passphrase for ``ssh_pkey``, if any.

    Raises:
        RuntimeError: If the pool is already initialized (call
            [`close_pool`][noxdb.connection.close_pool] first
            to reconfigure), or if SSH credentials are incomplete, or
            if required DB credentials are missing.
        FileNotFoundError: If ``config_path`` is set but missing.
    """
    global _pool, _pool_counter, _tunnel
    if _pool is not None:
        raise RuntimeError(
            "pool already initialized; call close_pool() before re-initializing"
        )

    creds = _resolve_credentials(
        config_path,
        section,
        {
            "host": host,
            "port": port,
            "user": user,
            "password": password,
            "database": database,
        },
    )

    ssh_creds = _resolve_ssh_credentials(
        config_path,
        DEFAULT_SSH_SECTION,
        {
            "ssh_host": ssh_host,
            "ssh_port": ssh_port,
            "ssh_user": ssh_user,
            "ssh_password": ssh_password,
            "ssh_pkey": ssh_pkey,
            "ssh_pkey_password": ssh_pkey_password,
        },
    )

    effective_local_port: int | None = None

    cfg_local_port = int(ssh_creds.get("local_port", 0))
    if cfg_local_port and _is_port_open(cfg_local_port):
        effective_local_port = cfg_local_port

    if effective_local_port is not None:
        creds["host"] = "127.0.0.1"
        creds["port"] = effective_local_port
    elif ssh_creds.get("ssh_host"):
        _tunnel = _open_tunnel(ssh_creds, creds["host"], creds["port"])
        local_host, local_port = _tunnel.local_bind_address
        creds["host"] = local_host
        creds["port"] = int(local_port)

    try:
        _pool_counter += 1
        pool_name = f"noxdb_{os.getpid()}_{_pool_counter}"
        _pool = mariadb.ConnectionPool(
            pool_name=pool_name,
            pool_size=pool_size,
            autocommit=False,
            read_timeout=15,
            write_timeout=15,
            connect_timeout=10,
            **creds,
        )
    except Exception:
        if _tunnel is not None:
            try:
                _tunnel.stop()
            except Exception:
                pass
            _tunnel = None
        raise
    _setup_audit_logger()

close_pool

close_pool() -> None

Close the pool, tear down the SSH tunnel, and release audit handlers.

Safe to call when no pool exists. Used by test teardown and at shutdown. After this returns, init_pool can be called again.

Source code in src/noxdb/connection.py

def close_pool() -> None:
    """Close the pool, tear down the SSH tunnel, and release audit handlers.

    Safe to call when no pool exists. Used by test teardown and at
    shutdown. After this returns,
    [`init_pool`][noxdb.connection.init_pool] can be called
    again.
    """
    global _pool, _tunnel
    if _pool is not None:
        try:
            _pool.close()
        except Exception:
            pass
        _pool = None
    if _tunnel is not None:
        try:
            _tunnel.stop()
        except Exception:
            pass
        _tunnel = None
    _teardown_audit_logger()

get_connection

get_connection() -> Iterator[mariadb.Connection]

Yield a pooled connection.

The connection commits on normal exit of the with block and rolls back on any exception. The pool is initialized lazily on first call with default settings. Server-side autocommit=0 is re-asserted on every checkout to defend against pool-reset drift.

Yields:

Type	Description
Connection	A mariadb.Connection borrowed from the pool. Returned to
Connection	the pool (not destroyed) when the with block exits.

Source code in src/noxdb/connection.py

@contextmanager
def get_connection() -> Iterator[mariadb.Connection]:
    """Yield a pooled connection.

    The connection commits on normal exit of the ``with`` block and
    rolls back on any exception. The pool is initialized lazily on
    first call with default settings. Server-side ``autocommit=0`` is
    re-asserted on every checkout to defend against pool-reset drift.

    Yields:
        A ``mariadb.Connection`` borrowed from the pool. Returned to
        the pool (not destroyed) when the ``with`` block exits.
    """
    conn = _get_pool().get_connection()
    # Pool reset between uses can revert server-side autocommit to the
    # server default (1 on MariaDB), which would silently break our rollback
    # path. The Python-side attribute is unreliable here: when the pool
    # config already declared autocommit=False, the driver's setter sees a
    # matching python flag and skips sending `SET autocommit=0`, leaving
    # the server in whatever state the previous checkout left it in.
    # Force server-side OFF explicitly with a SET so each transaction
    # starts from a known state.
    conn.autocommit = False
    _force_server_autocommit_off(conn)
    try:
        yield conn
        conn.commit()
    except BaseException:
        # BaseException, not Exception: KeyboardInterrupt / SystemExit must
        # also roll back, otherwise an interrupted transaction returns to the
        # pool still holding its locks and stalls the next checkout.
        try:
            conn.rollback()
        except Exception:
            # Don't mask the original exception with a rollback failure.
            logging.getLogger(__name__).exception("rollback failed")
        raise
    finally:
        # close() returns the connection to the pool, it does not destroy it.
        try:
            conn.close()
        except Exception:
            logging.getLogger(__name__).exception("connection close failed")

transaction

transaction() -> Iterator[_LoggingCursor]

Yield an audit-logging cursor. All statements share one transaction.

Commit and rollback are inherited from get_connection: if the with block exits normally everything commits; if any statement raises, everything rolls back atomically.

Yields:

Type	Description
_LoggingCursor	A _LoggingCursor that audits write statements
_LoggingCursor	(INSERT/UPDATE/DELETE/REPLACE) to ~/.noxdb/audit.log
_LoggingCursor	(override via NOXDB_AUDIT_LOG).

Source code in src/noxdb/connection.py

@contextmanager
def transaction() -> Iterator[_LoggingCursor]:
    """Yield an audit-logging cursor. All statements share one transaction.

    Commit and rollback are inherited from
    [`get_connection`][noxdb.connection.get_connection]: if the
    ``with`` block exits normally everything commits; if any statement
    raises, everything rolls back atomically.

    Yields:
        A ``_LoggingCursor`` that audits write statements
        (INSERT/UPDATE/DELETE/REPLACE) to ``~/.noxdb/audit.log``
        (override via ``NOXDB_AUDIT_LOG``).
    """
    with get_connection() as conn:
        cursor = _LoggingCursor(conn.cursor())
        try:
            yield cursor
        finally:
            cursor.close()

execute

execute(query: str, params: Any = None) -> list[dict[str, Any]]

Run one query and return rows as a list of dicts.

Each call uses its own pooled connection and its own transaction; for multi-statement atomicity use transaction instead.

Parameters:

Name	Type	Description	Default
query	str	SQL statement, optionally with ? placeholders.	required
params	Any	Bind parameters. None is normalized to ().	None

Returns:

Type	Description
list[dict[str, Any]]	For SELECT, one dict[str, Any] per row (column name →
list[dict[str, Any]]	value). For INSERT/UPDATE/DELETE/REPLACE, an empty list (and
list[dict[str, Any]]	the statement is audit-logged).

Source code in src/noxdb/connection.py

def execute(query: str, params: Any = None) -> list[dict[str, Any]]:
    """Run one query and return rows as a list of dicts.

    Each call uses its own pooled connection and its own transaction;
    for multi-statement atomicity use
    [`transaction`][noxdb.connection.transaction] instead.

    Args:
        query: SQL statement, optionally with ``?`` placeholders.
        params: Bind parameters. ``None`` is normalized to ``()``.

    Returns:
        For SELECT, one ``dict[str, Any]`` per row (column name →
        value). For INSERT/UPDATE/DELETE/REPLACE, an empty list (and
        the statement is audit-logged).
    """
    with get_connection() as conn:
        cur = conn.cursor()
        try:
            cur.execute(query, params if params is not None else ())
            _log_if_write(query, params, cur.rowcount)
            if cur.description is None:
                return []
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]
        finally:
            cur.close()