lib.db

class lib.db.Database(name, dbapi, connect, formatting='named')

Bases: object

A database abstraction layer based on DB-API2 specification.

It provides basic functionality to access databases using Python driver implementations based on the DB-API2 specification (PEP 249).

The following methods are provided: ‚__init__()‘ - create a new database object ‚connect()‘ - establish the connection to the database ‚close()‘ - close the connection to the database ‚setup()‘ - check/update/upgrade database structure ‚execute()‘ - execute statement (no result returned) ‚fetchone()‘ - execute statement and return first row from result ‚fetchall()‘ - execute statement and reeturn all rows from result ‚cursor()‘ - create a cursor object to execute multiple statements ‚commit()‘ - commit a transaction (if the selcted database supports it) ‚rollback()‘ - rollback a transaction (if the selcted database supports it) ‚lock()‘ - acquire the database lock (prevent simultaneous reads/writes) ‚release()‘ - release the database lock ‚transaction()‘ - context manager: run several statements as one commit/rollback unit, using lock()/release() internally ‚verify()‘ - check database connection and reconnect if required ‚connected()‘ - check if database is connected

The SQL statements executed may have placeholders and parameters which are passed to the execution methods listed above. The following DB-API driver implementations are supported: - qmark: Specify placeholders as „?“ and parameters as list - format: Specify placeholders as „%s“ and parameters as list - numeric: Specify placeholders as „:1“ and parameters as list - named: Specify placeholders as „:name“ and parameters as dict - pyformat: Specify placeholders as „%(arg)s“ and parameters as dict

Further you can choose a different formatting style in your code when using this class. Specify one of the formatting listed above or use the default - which is named.

In case the driver implementation uses a different formatting it will be converted transparently!

close()

Closes the database connection

commit()

Commit the current transaction

connect()

Connects to the database

connected()

Return the connected status

cursor()

Create a new cursor for executing statements

execute(stmt, params=(), formatting=None, cur=<object object>, quiet=False)

Execute the given statement

This will execute the statement specified in the ‚stmt‘ parameter which may contain parameter placeholders (depending on selected formatting style given in constructor).

The parameters can be specified in ‚params‘ parameter as list or dict depending on selected formatting style.

To overwrite the global formatting style given in constructor, the parameter ‚formatting‘ can be used to change the style for the given statement.

If already aqcuired a cursor you can use this cursor by using the ‚cur‘ parameter. If omitted a new cursor will be aqcuire for this statement and released afterwards. Passing ‚cur‘ explicitly as None is treated as a caller error (TypeError) rather than „omitted“ - see the NO_CURSOR module comment.

Set ‚quiet‘ to True to suppress the error-level log entry for an expected failure (e.g. a first-run „table does not exist yet“ probe). The exception is always raised regardless of ‚quiet‘ - only the log entry is conditional.

fetchall(stmt, params=(), formatting=None, cur=<object object>, quiet=False)

Execute given statement and fetch all rows from result

This method can be used to fetch all rows from the result. It accepts the same arguments as mentioned in the ‚execute()‘ method.

fetchone(stmt, params=(), formatting=None, cur=<object object>, quiet=False)

Execute given statement and fetch one row from result

This method can be used in case you only want to fetch one row from the result. It accepts the same arguments as mentioned in the ‚execute()‘ method.

is_connection_error(exc)

True if exc is a PEP 249 connection-trouble class for this driver.

Uses the raw classes tuple, not _reconnect_exceptions‘ (Exception,) fallback - an unclassifiable driver must never downgrade an unrelated bug’s log level just because it can’t be classified.

lock(timeout=-1)

Acquire a database lock

Raises RuntimeError immediately - never waits out timeout - if the calling thread already holds this lock. self._fdb_lock is a plain, non-reentrant threading.Lock: without this check a same-thread re-entry would just block until timeout, indistinguishable from genuine cross-thread contention. This is what makes transaction()‘s non-reentrancy hazard (see its docstring for what triggers it) fail loud and immediately instead of silently hanging.

Tracks threading.current_thread() (the Thread object), not threading.get_ident() (a small OS-recyclable int) - an owner marker holding a live reference to the actual Thread object can’t collide with an unrelated later thread the way a recycled ident could once the original owner had already exited without releasing.

A stale read of the owner marker (benign: plain attribute read under the GIL, no separate synchronization) can only cause a false negative here - falling through to the real self._fdb_lock.acquire() call, which then behaves exactly as it did before this check existed. It can never falsely raise for a thread that doesn’t actually hold the lock.

lock_holder_description()

‚ (lock currently held by thread <name>)‘, or ‚‘ if unheld.

For a log message reporting a failed lock acquisition - names the cause instead of just the symptom, e.g. distinguishing „someone else is mid-compaction“ from a genuine connection failure. Safe to call from a different thread than the one that failed to acquire: same benign, unsynchronized attribute read as lock()‘s own reentrancy check (see its docstring) - at worst one step stale, never wrong about a thread that doesn’t actually hold the lock.

release()

Release the database lock

rollback()

Rollback the current transaction

setup(queries)

Setup or update the database structure.

This method can be used to setup the database structure by providing the SQL statements to this method. Additionally it will check if the structure is already up to date by checking the data of the version table (which will also be created by this method if it does not exist already).

To setup the database you need to specify the required SQL statements (e.g. ‚CREATE TABLE‘, ‚CREATE INDEX‘ etc.) in the ‚queries‘ parameter. This will be a dictionary where the keys are simple version numbers and values are a two-item list for a rollout and rollback statement.

E.g.::

db.setup({1: [‚CREATE TABLE xyz (…)‘, ‚DROP TABLE xyz‘], 2: […]})

For an extended example take a look into the ‚database‘ plugin.

Each version’s rollout statement and its version-row bookkeeping commit as their own transaction(), not one commit for the whole migration - MySQL/MariaDB DDL (CREATE TABLE/ALTER TABLE/…) commits implicitly as a side effect regardless of any wrapping transaction, so a single end-of-loop commit couldn’t make a multi-step migration atomic there anyway; on a crash between one step’s DDL and its own version-row commit, only that one step needs to be figured out by hand on restart (the DDL re-running and failing with „already exists“), not the whole remaining migration. On sqlite, Python’s sqlite3 module autocommits DDL anyway under its default (legacy) isolation handling - an implicit BEGIN is only issued before DML - so per-step commits match what actually happens there too; neither backend gets multi-step atomicity.

transaction(timeout=None)

Run a block of statements as one transaction.

Acquires self._fdb_lock, yields a cursor for the caller to run multiple statements against, commits on clean exit, rolls back (best-effort) on any exception and re-raises it, always releases the lock. Use this instead of hand-rolling lock()/cursor()/commit()/rollback()/release() at each call site - a hand-rolled block that omits cleanup on failure leaves a corrupted connection for the next caller to inherit.

Rolling back unconditionally on any exception is always safe here - no DB-API2 exception-class distinction needed. If the connection is genuinely dead, the rollback attempt itself fails, which triggers the existing self-healing reset in rollback() and re-raises - so this gets „reset on real connection failure“ for free, without a fragile classification heuristic.

Usage:

with self._db.transaction() as cur:
    self._log_store.insert(item_id, entry, item_type, now_ms, cur=cur)

IMPORTANT - non-reentrancy: self._fdb_lock is a plain, non-reentrant threading.Lock, held for the entire block. Two hazards follow from this, both enforced by lock() itself raising RuntimeError immediately on a same-thread re-entry (see lock()) rather than deadlocking/timing out:

• transaction() cannot be nested.

• Any call from inside the block that acquires the lock internally - a cur=None execute()/fetchone()/fetchall() call, or connect()/close()/verify()/setup() - hits the same wall. Always pass the yielded cur through explicitly to statements run inside the block.

Parameter:

timeout – Seconds to wait for the lock; defaults to the configured db_query_timeout.

verify(retry=5, delay=5, probe_timeout=5)

Verifies the connection status and reconnets if required

The connected status of the connection will be checked by executing a simple SQL statement. If this fails or the connection is not established already a new connection will be opened.

In case the reconnect fails you can specify how many times a reconnect will be executed until it will give up. This can be specified by the ‚retry‘ parameter.

To specify the delay between retries use the delay parameter, which defaults to 5 seconds.

Cost note: each attempt can cost up to probe_timeout (see below) even against a completely unresponsive server, and retry multiplies that. A caller whose own failure path already gets retried on its own cadence (e.g. a scheduled cycle) should still pass a low retry here rather than relying on this loop alone.

probe_timeout (pymysql only, default 5s): a „SELECT 1“ probe doesn’t need the full db_query_timeout (60s default) a real query gets. Overrides pymysql’s read_timeout/write_timeout for the duration of this call, then restores them (finally block below) - including on an already-open connection, since pymysql applies these fresh on every read/write rather than caching them at connect time. hasattr-guarded: read_timeout is not part of pymysql’s public API and could be renamed in a future version, in which case this silently no-ops instead of raising. Scoped to the ‚pymysql‘ driver name, not the wider _pymysql_driver_names set - MySQLdb/mysql.connector may name this attribute differently.

version()

Best-effort database engine/server version string, or None on failure.

Cached after a successful lookup - the engine version can only change via a server restart, which always tears down and reconnects this object’s connection (see _reset_connection_locked()), so that’s the only point the cache needs to be invalidated.