table.create(..., replace=True / ignore = True) closes #568

sqlite_utils/cli.py

@@ -1543,13 +1543,7 @@ def create_table(
         coltypes[name] = ctype.upper()
     # Does table already exist?
     if table in db.table_names():
-        if ignore:
-            return
-        elif replace:
-            db[table].drop()
-        elif transform:
-            pass
-        else:
+        if (not ignore and not replace and not transform):
             raise click.ClickException(
                 'Table "{}" already exists. Use --replace to delete and replace it.'.format(
                     table
@@ -1561,6 +1555,8 @@ def create_table(
         not_null=not_null,
         defaults=dict(default),
         foreign_keys=fk,
+        ignore=ignore,
+        replace=replace,
         transform=transform,
     )
 

sqlite_utils/db.py

@@ -932,6 +932,8 @@ class Database:
         hash_id_columns: Optional[Iterable[str]] = None,
         extracts: Optional[Union[Dict[str, str], List[str]]] = None,
         if_not_exists: bool = False,
+        replace: bool = False,
+        ignore: bool = False,
         transform: bool = False,
     ) -> "Table":
         """
@@ -950,9 +952,16 @@ class Database:
         :param hash_id_columns: List of columns to be used when calculating the hash ID for a row
         :param extracts: List or dictionary of columns to be extracted during inserts, see :ref:`python_api_extracts`
         :param if_not_exists: Use ``CREATE TABLE IF NOT EXISTS``
+        :param replace: Drop and replace table if it already exists
+        :param ignore: Silently do nothing if table already exists
         :param transform: If table already exists transform it to fit the specified schema
         """
         # Transform table to match the new definition if table already exists:
+        if self[name].exists():
+            if ignore:
+                return self[name]
+            elif replace:
+                self[name].drop()
         if transform and self[name].exists():
             table = cast(Table, self[name])
             should_transform = False
@@ -1610,6 +1619,8 @@ class Table(Queryable):
         hash_id_columns: Optional[Iterable[str]] = None,
         extracts: Optional[Union[Dict[str, str], List[str]]] = None,
         if_not_exists: bool = False,
+        replace: bool = False,
+        ignore: bool = False,
         transform: bool = False,
     ) -> "Table":
         """
@@ -1627,6 +1638,9 @@ class Table(Queryable):
         :param hash_id_columns: List of columns to be used when calculating the hash ID for a row
         :param extracts: List or dictionary of columns to be extracted during inserts, see :ref:`python_api_extracts`
         :param if_not_exists: Use ``CREATE TABLE IF NOT EXISTS``
+        :param replace: Drop and replace table if it already exists
+        :param ignore: Silently do nothing if table already exists
+        :param transform: If table already exists transform it to fit the specified schema
         """
         columns = {name: value for (name, value) in columns.items()}
         with self.db.conn:
@@ -1642,6 +1656,8 @@ class Table(Queryable):
                 hash_id_columns=hash_id_columns,
                 extracts=extracts,
                 if_not_exists=if_not_exists,
+                replace=replace,
+                ignore=ignore,
                 transform=transform,
             )
         return self

tests/test_create.py

@@ -1186,6 +1186,25 @@ def test_create_if_no_columns(fresh_db):
     assert error.value.args[0] == "Tables must have at least one column"
 
 
+def test_create_ignore(fresh_db):
+    fresh_db["t"].create({"id": int})
+    # This should error
+    with pytest.raises(sqlite3.OperationalError):
+        fresh_db["t"].create({"id": int})
+    # This should not
+    fresh_db["t"].create({"id": int}, ignore=True)
+
+
+def test_create_replace(fresh_db):
+    fresh_db["t"].create({"id": int})
+    # This should error
+    with pytest.raises(sqlite3.OperationalError):
+        fresh_db["t"].create({"id": int})
+    # This should not
+    fresh_db["t"].create({"name": str}, replace=True)
+    assert fresh_db["t"].schema == ("CREATE TABLE [t] (\n" "   [name] TEXT\n" ")")
+
+
 @pytest.mark.parametrize(
     "cols,kwargs,expected_schema,should_transform",
     (