Add duplicate table feature

sqlite_utils/db.py

@@ -305,11 +305,7 @@ class Database:
         ), "Either specify a filename_or_conn or pass memory=True"
         if memory_name:
             uri = "file:{}?mode=memory&cache=shared".format(memory_name)
-            self.conn = sqlite3.connect(
-                uri,
-                uri=True,
-                check_same_thread=False,
-            )
+            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)):
@@ -1104,9 +1100,7 @@ class Queryable:
         self.name = name
 
     def count_where(
-        self,
-        where: str = None,
-        where_args: Optional[Union[Iterable, dict]] = None,
+        self, where: str = None, where_args: Optional[Union[Iterable, dict]] = None,
     ) -> int:
         """
         Executes ``SELECT count(*) FROM table WHERE ...`` and returns a count.
@@ -1476,6 +1470,20 @@ class Table(Queryable):
             )
         return self
 
+    def duplicate(self, name_new: str) -> "Table":
+        """
+        Duplicate this table in this database.
+
+        :param name_new: Name of new table.
+        """
+        assert self.exists()
+        with self.db.conn:
+            sql = "CREATE TABLE [{new_table}] AS SELECT * FROM [{table}];".format(
+                new_table=name_new, table=self.name,
+            )
+            self.db.execute(sql)
+        return self.db[name_new]
+
     def transform(
         self,
         *,
@@ -1711,13 +1719,7 @@ class Table(Queryable):
                 )
         else:
             lookup_table.create(
-                {
-                    **{
-                        "id": int,
-                    },
-                    **lookup_columns_definition,
-                },
-                pk="id",
+                {**{"id": int,}, **lookup_columns_definition,}, pk="id",
             )
         lookup_columns = [(rename.get(col) or col) for col in columns]
         lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True)

tests/test_duplicate.py

@@ -0,0 +1,36 @@
+import datetime
+
+
+def test_duplicate(fresh_db):
+    # Create table using native Sqlite statement:
+    fresh_db.execute(
+        """CREATE TABLE [table1] (
+    [text_col] TEXT,
+    [real_col] REAL,
+    [int_col] INTEGER,
+    [bool_col] INTEGER,
+    [datetime_col] TEXT)"""
+    )
+    # Insert one row of mock data:
+    dt = datetime.datetime.now()
+    data = {
+        "text_col": "Cleo",
+        "real_col": 3.14,
+        "int_col": -255,
+        "bool_col": True,
+        "datetime_col": str(dt),
+    }
+    table1 = fresh_db["table1"]
+    row_id = table1.insert(data).last_rowid
+    # Duplicate table:
+    table2 = table1.duplicate("table2")
+    # Ensure data integrity:
+    assert data == table2.get(row_id)
+    # Ensure schema integrity:
+    assert [
+        {"name": "text_col", "type": "TEXT"},
+        {"name": "real_col", "type": "REAL"},
+        {"name": "int_col", "type": "INT"},
+        {"name": "bool_col", "type": "INT"},
+        {"name": "datetime_col", "type": "TEXT"},
+    ] == [{"name": col.name, "type": col.type} for col in table2.columns]