1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
#
"""Ensure SQLite usage prevents common mistakes, and get best performance."""
from __future__ import annotations
from typing import Callable
import apsw
import apsw.ext
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>`__.
"""
connection.pragma("journal_mode", "wal")
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)
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")
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)
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,
library_logging,
)
"All of them"
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)
|
