#
"""Ensure SQLite usage prevents common mistakes, and get best performance."""
from __future__ import annotations
from typing import Callable
import apsw
import apsw.ext
[docs]
def connection_wal(connection: apsw.Connection) -> None:
"""Turns on write ahead logging
Reduces contention and improves write performance. WAL is
`described here <https://www.sqlite.org/wal.html>`__.
"""
try:
if not connection.readonly("main"):
connection.pragma("journal_mode", "wal")
except apsw.ReadOnlyError:
# journal/wal etc could still be readonly
pass
[docs]
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)
[docs]
def connection_enable_foreign_keys(connection: apsw.Connection) -> None:
"""Enables foreign key constraints
`Foreign keys <https://www.sqlite.org/foreignkeys.html>`__ need to
`be enabled <https://www.sqlite.org/foreignkeys.html#fk_enable>`__
to have an effect.
"""
connection.pragma("foreign_keys", "ON")
[docs]
def connection_dqs(connection: apsw.Connection) -> None:
"""Double quotes are for identifiers only, not strings
Turns off `allowing double quoted strings
<https://www.sqlite.org/quirks.html#dblquote>`__ if they don't
match any identifier (column/table names etc), making it an error
to use double quotes around strings. SQL strings use single
quotes.
"""
connection.config(apsw.SQLITE_DBCONFIG_DQS_DML, 0)
connection.config(apsw.SQLITE_DBCONFIG_DQS_DDL, 0)
[docs]
def connection_optimize(connection: apsw.Connection) -> None:
"""Enables query planner optimization
It enables the query planner to record cases when it would benefit
from having accurate statistics about tables and indexes for the
queries you make.
You can later run :code:`connection.pragma("optimize")` to have
those statistics updated, such as when closing a database or
periodically when the database is open for long periods of time.
The statistics are recorded in the database and help with future
queries during this connection, and all future connections.
There is more detail in the `documentation
<https://sqlite.org/lang_analyze.html>`__.
"""
try:
if not connection.readonly("main"):
connection.pragma("optimize", 0x10002)
except apsw.ReadOnlyError:
# journal/wal etc could still be readonly
pass
[docs]
def connection_recursive_triggers(connection: apsw.Connection) -> None:
"""Recursive triggers are off for historical backwards compatibility
This `enables them
<https://www.sqlite.org/pragma.html#pragma_recursive_triggers>`__.
"""
connection.pragma("recursive_triggers", "ON")
[docs]
def library_logging() -> None:
"""Forwards SQLite logging to Python logging module
See :meth:`apsw.ext.log_sqlite`
"""
apsw.ext.log_sqlite()
recommended: tuple[Callable, ...] = (
connection_wal,
connection_busy_timeout,
connection_enable_foreign_keys,
connection_dqs,
connection_optimize,
connection_recursive_triggers,
library_logging,
)
"All of them"
[docs]
def apply(which: tuple[Callable, ...]) -> None:
"Applies library immediately and connection to new connections"
hooks: list[Callable] = []
for func in which:
if func.__name__.startswith("connection_"):
hooks.append(func)
else:
func()
def best_practise_connection_apply(connection: apsw.Connection) -> None:
for func in hooks:
func(connection)
apsw.connection_hooks.append(best_practise_connection_apply)