diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index c835fbd..b18c436 100644 --- a/sqlite_utils/db.py +++ b/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) diff --git a/tests/test_duplicate.py b/tests/test_duplicate.py new file mode 100644 index 0000000..0324b00 --- /dev/null +++ b/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]