db.iterdump() method uses sqlite-dump if available

This commit is contained in:
Simon Willison 2023-06-25 15:29:00 -07:00
commit 31bde41baf
8 changed files with 55 additions and 8 deletions

View file

@ -37,7 +37,7 @@ jobs:
run: sudo apt-get install libsqlite3-mod-spatialite
- name: On macOS with Python 3.10 test with sqlean.py
if: matrix.os == 'macos-latest' && matrix.python-version == '3.10'
run: pip install sqlean.py
run: pip install sqlean.py sqlite-dump
- name: Build extension for --load-extension test
if: matrix.os == 'ubuntu-latest'
run: |-

View file

@ -60,6 +60,12 @@ To install ``sqlean.py`` (which has compiled binary wheels available for all maj
sqlite-utils install sqlean.py
``pysqlite3`` and ``sqlean.py`` do not provide implementations of the ``.iterdump()`` method. To use that method (see :ref:`python_api_itedump`) or the ``sqlite-utils dump`` command you should also install the ``sqlite-dump`` package:
.. code-block:: bash
sqlite-utils install sqlite-dump
.. _installation_completion:
Setting up shell completion

View file

@ -1779,6 +1779,25 @@ The ``db.sqlite_version`` property returns a tuple of integers representing the
>>> db.sqlite_version
(3, 36, 0)
.. _python_api_itedump:
Dumping the database to SQL
===========================
The ``db.iterdump()`` method returns a sequence of SQL strings representing a complete dump of the database. Use it like this:
.. code-block:: python
full_sql = "".join(db.iterdump())
This uses the `sqlite3.Connection.iterdump() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.iterdump>`__ method.
If you are using ``pysqlite3`` or ``sqlean.py`` the underlying method may be missing. If you install the `sqlite-dump <https://pypi.org/project/sqlite-dump/>`__ package then the ``db.iterdump()`` method will use that implementation instead:
.. code-block:: bash
pip install sqlite-dump
.. _python_api_introspection:
Introspecting tables and views

View file

@ -1,4 +1,4 @@
[mypy]
[mypy-pysqlite3,sqlean]
[mypy-pysqlite3,sqlean,sqlite_dump]
ignore_missing_imports = True

View file

@ -395,7 +395,7 @@ def dump(path, load_extension):
"""
db = sqlite_utils.Database(path)
_load_extensions(db, load_extension)
for line in db.conn.iterdump():
for line in db.iterdump():
click.echo(line)
@ -1893,7 +1893,7 @@ def memory(
return
if dump:
for line in db.conn.iterdump():
for line in db.iterdump():
click.echo(line)
return
@ -1903,7 +1903,7 @@ def memory(
if save:
db2 = sqlite_utils.Database(save)
for line in db.conn.iterdump():
for line in db.iterdump():
db2.execute(line)
return

View file

@ -38,6 +38,12 @@ from typing import (
)
import uuid
try:
from sqlite_dump import iterdump
except ImportError:
iterdump = None
SQLITE_MAX_VARS = 999
_quote_fts_re = re.compile(r'\s+|(".*?")')
@ -1149,6 +1155,18 @@ class Database:
sql += " [{}]".format(name)
self.execute(sql)
def iterdump(self) -> Generator[str, None, None]:
"A sequence of strings representing a SQL dump of the database"
if iterdump:
yield from iterdump(self.conn)
else:
try:
yield from self.conn.iterdump()
except AttributeError:
raise AttributeError(
"conn.iterdump() not found - try pip install sqlite-dump"
)
def init_spatialite(self, path: Optional[str] = None) -> bool:
"""
The ``init_spatialite`` method will load and initialize the SpatiaLite extension.

View file

@ -135,7 +135,8 @@ def test_analyze_column(db_to_analyze, column, extra_kwargs, expected):
def db_to_analyze_path(db_to_analyze, tmpdir):
path = str(tmpdir / "test.db")
db = sqlite3.connect(path)
db.executescript("\n".join(db_to_analyze.conn.iterdump()))
sql = "\n".join(db_to_analyze.iterdump())
db.executescript(sql)
return path

View file

@ -164,9 +164,9 @@ def test_memory_dump(extra_args):
input="id,name\n1,Cleo\n2,Bants",
)
assert result.exit_code == 0
assert result.output.strip() == (
expected = (
"BEGIN TRANSACTION;\n"
'CREATE TABLE "stdin" (\n'
'CREATE TABLE IF NOT EXISTS "stdin" (\n'
" [id] INTEGER,\n"
" [name] TEXT\n"
");\n"
@ -176,6 +176,9 @@ def test_memory_dump(extra_args):
"CREATE VIEW t AS select * from [stdin];\n"
"COMMIT;"
)
# Using sqlite-dump it won't have IF NOT EXISTS
expected_alternative = expected.replace("IF NOT EXISTS ", "")
assert result.output.strip() in (expected, expected_alternative)
@pytest.mark.parametrize("extra_args", ([], ["select 1"]))