memory_name= feature, closes #405

This commit is contained in:
Simon Willison 2022-02-15 17:21:25 -08:00
commit 4bc06a2437
3 changed files with 25 additions and 3 deletions

View file

@ -98,6 +98,12 @@ If you want to create an in-memory database, you can do so like this:
db = Database(memory=True)
You can also create a named in-memory database. Unlike regular memory databases these can be accessed by multiple threads, provided at least one reference to the database still exists. `del db` will clear the database from memory.
.. code-block:: python
db = Database(memory_name="my_shared_database")
Connections use ``PRAGMA recursive_triggers=on`` by default. If you don't want to use `recursive triggers <https://www.sqlite.org/pragma.html#pragma_recursive_triggers>`__ you can turn them off using:
.. code-block:: python

View file

@ -278,6 +278,7 @@ class Database:
- ``filename_or_conn`` - String path to a file, or a ``pathlib.Path`` object, or a
``sqlite3`` connection
- ``memory`` - set to ``True`` to create an in-memory database
- ``memory_name`` - creates a named in-memory database that can be shared across multiple connections.
- ``recreate`` - set to ``True`` to delete and recreate a file database (**dangerous**)
- ``recursive_triggers`` - defaults to ``True``, which sets ``PRAGMA recursive_triggers=on;`` -
set to ``False`` to avoid setting this pragma
@ -294,15 +295,23 @@ class Database:
self,
filename_or_conn: Union[str, pathlib.Path, sqlite3.Connection] = None,
memory: bool = False,
memory_name: str = None,
recreate: bool = False,
recursive_triggers: bool = True,
tracer: Callable = None,
use_counts_table: bool = False,
):
assert (filename_or_conn is not None and not memory) or (
filename_or_conn is None and memory
assert (filename_or_conn is not None and (not memory and not memory_name)) or (
filename_or_conn is None and (memory or memory_name)
), "Either specify a filename_or_conn or pass memory=True"
if memory or filename_or_conn == ":memory:":
if memory_name:
uri = "file:{}?mode=memory&cache=shared".format(memory_name)
self.conn = sqlite3.connect(
uri,
uri=True,
check_same_thread=False,
)
elif memory or filename_or_conn == ":memory:":
self.conn = sqlite3.connect(":memory:")
elif isinstance(filename_or_conn, (str, pathlib.Path)):
if recreate and os.path.exists(filename_or_conn):

View file

@ -9,3 +9,10 @@ def test_recursive_triggers():
def test_recursive_triggers_off():
db = Database(memory=True, recursive_triggers=False)
assert not db.execute("PRAGMA recursive_triggers").fetchone()[0]
def test_memory_name():
db1 = Database(memory_name="shared")
db2 = Database(memory_name="shared")
db1["dogs"].insert({"name": "Cleo"})
assert list(db2["dogs"].rows) == [{"name": "Cleo"}]