-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsqlite_conn.py
More file actions
285 lines (219 loc) · 9.74 KB
/
sqlite_conn.py
File metadata and controls
285 lines (219 loc) · 9.74 KB
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
import sqlite3
from typing import Any
import logging
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)
class SQLiteConnectionContextManager:
def __init__(self, db_path: str):
"""
Initialize a SQLite connection using Context Manager Pattern.
This automatically handles connection lifecycle within a 'with' statement.
Pros:
- Automatic resource cleanup guaranteed by Python's context manager protocol
- Clear scope of database operations within the 'with' block
- Exception-safe - connection will be closed even if an exception occurs
- No need to remember to call close() explicitly
Cons:
- Connection is short-lived - limited to the 'with' block scope
- Need to create new context for each database operation session
- May not be suitable for long-running database sessions
Best for: Well-defined database operation sessions where you want guaranteed
cleanup and clear resource management boundaries.
Usage:
with SQLiteConnection("database.db") as db:
db.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
"""
self.db_path = db_path
self.conn: sqlite3.Connection | None = None
def __enter__(self):
self.conn = sqlite3.connect(self.db_path)
return self
# -- These params don't have to be used, but are required for context manager --
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any):
if self.conn:
self.conn.close()
self.conn = None
def execute(self, query: str, params: tuple[str, ...] = ()):
"""
Executes a SQL query with the given parameters.
See more about using params/placeholders to bind values in SQL queries here:
https://docs.python.org/3/library/sqlite3.html#how-to-use-placeholders-to-bind-values-in-sql-queries
"""
if not self.conn:
raise RuntimeError("Connection not established.")
cursor = self.conn.cursor()
cursor.execute(query, params)
self.conn.commit()
# --- OTHER CONNECTION PATTERNS ---
class SQLiteConnectionEager:
def __init__(self, db_path: str):
"""
Initialize a SQLite connection upon initialization (Eager Connection Pattern).
This pattern establishes the database connection immediately upon object creation.
Pros:
- Simple and straightforward
- Connection is ready to use immediately
- No separate connect() method
Cons:
- May waste resources if the connection isn't used immediately
- No control over when the connection is established
- You have to create a new instance to reconnect
- Harder to handle connection failures gracefully
- Can lead to "too many connections" errors in multi-threaded applications
Best for: Simple scripts or applications with immediate database usage.
"""
self.db_path = db_path
# -- Immediately establish connection --
self.conn: sqlite3.Connection | None = sqlite3.connect(self.db_path)
def close(self):
if self.conn:
self.conn.close()
self.conn = None
def execute(self, query: str, params: tuple[str, ...]):
if not self.conn:
raise RuntimeError("Connection not established.")
cursor = self.conn.cursor()
cursor.execute(query, params)
self.conn.commit()
return cursor
class SQLiteConnectionLazy:
"""Truly Lazy Pattern - connects only when execute() is called.
Pros:
- Connection is established only when needed, helping reduce connection overhead
Cons:
- Need to check/establish connection in class methods
Best for: Applications with infrequent database access or where connection overhead is a concern.
"""
def __init__(self, db_path: str):
self.db_path = db_path
self._conn = None
def _ensure_connection(self):
"""Quietly ensure the connection is established."""
if not self._conn:
self._conn = sqlite3.connect(self.db_path)
def _get_cursor(self) -> sqlite3.Cursor:
if not self._conn:
raise RuntimeError("Database connection is not established.")
return self._conn.cursor()
def close(self):
if self._conn:
self._conn.close()
self._conn = None
def execute(self, query: str, params: tuple[str, ...]):
self._ensure_connection() # -- Connects only when needed (when 'execute' is called) --
cursor = self._get_cursor()
cursor.execute(query, params)
assert self._conn is not None
self._conn.commit()
return cursor
class SQLiteConnectionSingleton:
"""
Singleton Pattern - ensures only one connection instance exists and has lazy initialization.
Pros:
- Guarantees single connection across application
- Prevents overhead from multiple connections
- Simple global access
Cons:
- Can create bottlenecks in multi-threaded applications
- Harder to test and mock
- Global state can make debugging difficult
- Not suitable for applications needing multiple databases
Best for: Simple applications with a single database and minimal concurrency.
Important notes particular to python:
- Python's module system loads each module only once, which can itself act as a singleton.
If all you need is shared state, a module-level variable or function may be sufficient.
- You could use a private class (`_PrivateClass`) at the module level, but you'd have to
hard-code or otherwise inject the db_path early.
"""
# -- Class variables to hold the singleton instance and connection --
_instance: "SQLiteConnectionSingleton | None" = None
_conn: "sqlite3.Connection | None" = None
def __new__(cls, db_path: str):
"""__new__ is used to control/return an instance of the class (instantiation) and is called before __init__.
Below is a common pattern for singletons in Python.
The first call to __new__ will create the instance as normal, but subsequent calls will return the
same instance since _instance is a class variable that holds the singleton instance.
"""
if cls._instance is None:
# -- Create a new instance of the class (super is the object class in this case) --
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, db_path: str):
"""Initialize the singleton instance with the database path.
Even though __new__ ensures a singleton and is run before this, __init__ still runs on
every instantiation, even if the singleton already exists.
We use a class variable guard (_initialized) to avoid re-initializing attributes (i.e. db_path).
"""
# -- Check if an instance has already been initialized to avoid re-initialization. --
if not getattr(self, "_initialized", False):
self.db_path = db_path
# -- _initialized is set on the instance and not the class because setting it on the
# -- class would affect all instances, which would conflict with resetting the singleton
# -- during tests. This way it goes away when _instance is reset.
self._initialized = True
else:
if self.db_path != db_path:
raise ValueError(
f"Singleton already initialized with db_path='{self.db_path}'."
f" Cannot reinitialize with '{db_path}'"
)
logging.debug(
f"SQLiteConnectionSingleton already initialized with db_path='{self.db_path}'."
" Using existing instance."
)
def _ensure_connection(self):
"""Quietly ensure the connection is established."""
if not self._conn:
self._conn = sqlite3.connect(self.db_path)
def close(self):
if self._conn:
self._conn.close()
self._conn = None
def connect(self):
if self._conn is None:
self._conn = sqlite3.connect(self.db_path)
def execute(self, query: str, params: tuple[str, ...] = ()) -> sqlite3.Cursor:
self._ensure_connection()
assert self._conn is not None, "Connection not established."
cursor = self._conn.cursor()
cursor.execute(query, params)
self._conn.commit()
return cursor
@classmethod
def reset(cls):
cls._instance = None
cls._conn = None
class SQLiteConnectionMultition:
"""Multition Pattern - allows multiple instances with different configurations.
Pros:
- Multiple connections with different configurations
- Useful for applications needing multiple databases
- Each instance can have its own connection settings
Cons:
- More complex than Singleton
- Need to manage multiple instances
- Can lead to resource exhaustion if not managed properly
Best for: Applications needing multiple database connections with different settings.
"""
pass
class SQLiteConnectionPool:
"""
Connection Pool Pattern - manages multiple connections for concurrent access.
Pros:
- Better performance for multi-threaded applications
- Reuses connections instead of creating new ones
- Can limit the number of concurrent connections
- Handles connection lifecycle automatically
Cons:
- More complex implementation
- Additional overhead for simple applications
- Need to handle thread safety
Best for: Multi-threaded applications with frequent database access.
"""
pass
class SQLiteConnectionFactory:
pass
class SQLiteConnectionDependencyInjection:
pass
class SQLiteConnectionDecorator:
pass