Flexible bestpractice controls

[infogulch]

apsw's bestpractice system is nice, since they provide a way to centrally manage how all new connections are set up.

But I don't see how the current design allows for customization of the parameters:

apsw/apsw/bestpractice.py

Lines 27 to 37 in aa2c46f

	def connection_busy_timeout(connection: apsw.Connection, duration_ms: int = 100) -> None:
	"""Sets a short busy timeout
	:param duration_ms: How many thousandths of a second to wait
	When another thread or process has locked the database, SQLite
	immediately raises :exc:`apsw.BusyError`. Changing the `busy
	timeout <https://www.sqlite.org/c3ref/busy_timeout.html>`__ gives
	a grace period during which SQLite retries.
	"""
	connection.set_busy_timeout(duration_ms)

apsw.bestpractice.apply(
    (
        # how to use apply to set busy timeout 1000ms?
        apsw.bestpractice.connection_busy_timeout,
    )
)

So I came up with a pattern to define connection_* funcs that enables inline customization, and I thought I'd share it. Maybe you find it interesting:

def connection_load_extension(
    path: str, entrypoint: str | None = None
) -> Callable[[apsw.Connection], None]:
    def connection_load_extension_(conn: apsw.Connection) -> None:
        conn.enableloadextension(True)
        conn.loadextension(path, entrypoint=entrypoint)

    return connection_load_extension_


def connection_busy_timeout(
    duration_ms: int = 100,
) -> Callable[[apsw.Connection], None]:
    def connection_busy_timeout_(conn: apsw.Connection) -> None:
        conn.set_busy_timeout(duration_ms)

    return connection_busy_timeout_


def connection_syncronus(mode: str = "NORMAL") -> Callable[[apsw.Connection], None]:
    def connection_syncronus_(conn: apsw.Connection) -> None:
        conn.pragma("synchronous", mode)

    return connection_syncronus_


def connection_journal_mode(mode: str = "WAL") -> Callable[[apsw.Connection], None]:
    def connection_journal_mode_(conn: apsw.Connection) -> None:
        try:
            if not conn.readonly("main"):
                conn.pragma("journal_mode", mode)
        except apsw.ReadOnlyError:
            pass

    return connection_journal_mode_


def connection_cache_size(size: int = 2000) -> Callable[[apsw.Connection], None]:
    def connection_cache_size_(conn: apsw.Connection) -> None:
        conn.pragma("cache_size", size)

    return connection_cache_size_


def connection_trusted_schema(
    trusted: bool = True,
) -> Callable[[apsw.Connection], None]:
    def connection_trusted_schema_(conn: apsw.Connection) -> None:
        conn.pragma("trusted_schema", int(trusted))

    return connection_trusted_schema_


apsw.bestpractice.apply(
    (
        connection_load_extension(HONKER_EXTENSION_PATH, "sqlite3_honkerext_init"),
        connection_busy_timeout(1000),
        connection_syncronus("NORMAL"),
        connection_journal_mode("WAL"),
        connection_cache_size(2000),
        connection_trusted_schema(False),
        apsw.bestpractice.connection_dqs,
        apsw.bestpractice.connection_enable_foreign_keys,
        apsw.bestpractice.connection_optimize,
        apsw.bestpractice.connection_recursive_triggers,
        apsw.bestpractice.library_logging,
    )
)