Implement SqliteBackupConnector

Author: whileman133

dbbackup/db/sqlite.py

@@ -1,7 +1,8 @@
+import sqlite3
 import warnings
 from io import BytesIO
 from shutil import copyfileobj
-from tempfile import SpooledTemporaryFile
+from tempfile import SpooledTemporaryFile, NamedTemporaryFile
 
 from django.db import IntegrityError, OperationalError
 
@@ -108,3 +109,38 @@ def restore_dump(self, dump):
         path = self.connection.settings_dict["NAME"]
         with open(path, "wb") as db_file:
             copyfileobj(dump, db_file)
+
+
+class SqliteBackupConnector(BaseDBConnector):
+    """
+    Create a dump using the SQLite backup command,
+    which is safe to execute when the database is
+    in use (unlike simply copying the database file).
+    Restore by copying the backup file over the
+    database file.
+    """
+    extension = 'sqlite3'
+
+    def _write_dump(self, fileobj):
+        pass
+
+    def create_dump(self):
+        if not self.connection.is_usable():
+            self.connection.connect()
+        # Important: ensure the connection to the DB
+        # has been established.
+        self.connection.ensure_connection()
+        src_db_connection = self.connection.connection
+
+        bkp_db_file = NamedTemporaryFile()
+        bkp_path = bkp_db_file.name
+        with sqlite3.connect(bkp_path) as bkp_db_connection:
+            src_db_connection.backup(bkp_db_connection)
+
+        bkp_db_file.seek(0)
+        return bkp_db_file
+
+    def restore_dump(self, dump):
+        path = self.connection.settings_dict["NAME"]
+        with open(path, "wb") as db_file:
+            copyfileobj(dump, db_file)

dbbackup/tests/test_connectors/test_sqlite.py

@@ -4,7 +4,7 @@
 from django.db import connection
 from django.test import TestCase
 
-from dbbackup.db.sqlite import SqliteConnector, SqliteCPConnector
+from dbbackup.db.sqlite import SqliteConnector, SqliteCPConnector, SqliteBackupConnector
 from dbbackup.tests.testapp.models import CharModel, TextModel
 
 
@@ -65,3 +65,17 @@ def test_restore_dump(self):
         connector = SqliteCPConnector()
         dump = connector.create_dump()
         connector.restore_dump(dump)
+
+
+class SqliteBackupConnectorTest(TestCase):
+    def test_create_dump(self):
+        connector = SqliteBackupConnector()
+        dump = connector.create_dump()
+        dump_content = dump.read()
+        self.assertTrue(dump_content)
+        self.assertTrue(dump_content.startswith(b"SQLite format 3"))
+
+    def test_restore_dump(self):
+        connector = SqliteBackupConnector()
+        dump = connector.create_dump()
+        connector.restore_dump(dump)