From aafb16fbe8278b6fe1ebaabca6b8b524adca0715 Mon Sep 17 00:00:00 2001 From: seger Date: Wed, 7 Oct 2026 16:19:50 +0000 Subject: [PATCH] Ability to define unique columns when creating a table Adds a new unique= argument to Database.create_table(), Table.create(), Table.transform() and the table-creating insert/upsert methods (insert, insert_all, upsert, upsert_all). A string names a single column and a tuple names a compound constraint spanning several columns, so unique=["email"] and unique=[("account_id", "email")] are both supported. The argument is also available as a table default via db.table(..., unique=[...]). Also adds a table.uniques introspection property that lists the UNIQUE constraints on a table, mirroring table.checks. Refs #558 --- docs/python-api.rst | 60 +++++++++++++++++- sqlite_utils/db.py | 129 +++++++++++++++++++++++++++++++++++--- tests/test_create.py | 143 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 320 insertions(+), 12 deletions(-) diff --git a/docs/python-api.rst b/docs/python-api.rst index d515642a5..44fe0497e 100644 --- a/docs/python-api.rst +++ b/docs/python-api.rst @@ -785,7 +785,7 @@ You can directly create a new table without inserting any data into it using the The first argument here is a dictionary specifying the columns you would like to create. Each column is paired with a Python type indicating the type of column. See :ref:`python_api_add_column` for full details on how these types work. -This method takes optional arguments ``pk=``, ``column_order=``, ``foreign_keys=``, ``not_null=set()`` and ``defaults=dict()`` - explained below. +This method takes optional arguments ``pk=``, ``column_order=``, ``foreign_keys=``, ``not_null=set()``, ``unique=[]`` and ``defaults=dict()`` - explained below. A ``sqlite_utils.utils.sqlite3.OperationalError`` will be raised if a table of that name already exists. @@ -816,6 +816,7 @@ The ``transform=True`` option will update the table schema if any of the followi - The specified primary key - The order of the columns, defined using ``column_order=`` - The ``not_null=`` or ``defaults=`` arguments +- The ``unique=`` argument Changes to ``foreign_keys=`` are not currently detected and applied by ``transform=True``. @@ -844,6 +845,61 @@ An ``ANY`` column can store integers, floating point values, text, binary data o .. note:: In the CLI: :ref:`sqlite-utils create-table ` +.. _python_api_unique_constraints: + +Unique constraints +------------------ + +Pass ``unique=`` to add ``UNIQUE`` constraints to a table created with ``.create()``. Each item in the list describes one constraint. A string names a single column, while a tuple of strings names a compound constraint spanning several columns: + +.. code-block:: python + + db.table("users").create({ + "id": int, + "email": str, + "username": str, + }, pk="id", unique=["email"]) + +This produces: + +.. code-block:: sql + + CREATE TABLE "users" ( + "id" INTEGER PRIMARY KEY, + "email" TEXT, + "username" TEXT, + UNIQUE ("email") + ) + +Pass multiple strings to add a separate constraint for each column, or a tuple for a compound constraint. These can be combined: + +.. code-block:: python + + db.table("memberships").create({ + "account_id": int, + "email": str, + "role": str, + }, unique=["role", ("account_id", "email")]) + +The ``unique=`` argument is also accepted by ``.transform()``, so you can add a unique constraint to an existing table. Repeating a constraint that is already present is a no-op: + +.. code-block:: python + + db.table("users").transform(unique=["email"]) + +The methods that create a table on demand - ``.insert()``, ``.insert_all()``, ``.upsert()`` and ``.upsert_all()`` - accept ``unique=`` as well, applied when they create the table: + +.. code-block:: python + + db["users"].insert({"id": 1, "email": "one@example.com"}, pk="id", unique=["email"]) + +The ``.uniques`` property returns the ``UNIQUE`` constraints on a table: + +.. code-block:: python + + for constraint in db.table("users").uniques: + print([column.name for column in constraint.columns]) + .. _python_api_compound_primary_keys: Compound primary keys @@ -979,7 +1035,7 @@ You can set default values for these methods by accessing the table through the # Now you can call .insert() like so: table.insert({"id": 1, "name": "Tracy", "score": 5}) -The configuration options that can be specified in this way are ``pk``, ``foreign_keys``, ``column_order``, ``not_null``, ``defaults``, ``batch_size``, ``hash_id``, ``hash_id_columns``, ``alter``, ``ignore``, ``replace``, ``extracts``, ``conversions``, ``columns``, ``strict``. These are all documented below. +The configuration options that can be specified in this way are ``pk``, ``foreign_keys``, ``column_order``, ``not_null``, ``unique``, ``defaults``, ``batch_size``, ``hash_id``, ``hash_id_columns``, ``alter``, ``ignore``, ``replace``, ``extracts``, ``conversions``, ``columns``, ``strict``. These are all documented below. .. _python_api_defaults_not_null: diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index c011d9bb0..0963d1dff 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -1434,6 +1434,7 @@ def create_table_sql( foreign_keys: ForeignKeysType | None = None, column_order: list[str] | None = None, not_null: Iterable[str] | None = None, + unique: Iterable[str] | Iterable[Iterable[str]] | None = None, defaults: dict[str, Any] | None = None, hash_id: str | None = None, hash_id_columns: Iterable[str] | None = None, @@ -1454,6 +1455,7 @@ def create_table_sql( :param foreign_keys: List of foreign key definitions for this table :param column_order: List specifying which columns should come first :param not_null: List of columns that should be created as ``NOT NULL`` + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Dictionary specifying default values for columns :param hash_id: Name of column to be used as a primary key containing a hash of the other columns :param hash_id_columns: List of columns to be used when calculating the hash ID for a row @@ -1508,7 +1510,12 @@ def create_table_sql( table_checks.append(check) uniques_by_column: dict[str, list[Unique]] = {} table_uniques: list[Unique] = [] - for unique in _uniques or (): + all_uniques: list[Unique] = list(_uniques or ()) + all_uniques.extend( + Unique(tuple(UniqueColumn(column) for column in columns_tuple)) + for columns_tuple in resolve_uniques(unique) + ) + for unique_constraint in all_uniques: resolved_unique = Unique( tuple( UniqueColumn( @@ -1516,13 +1523,15 @@ def create_table_sql( collation=column.collation, order=column.order, ) - for column in unique.columns + for column in unique_constraint.columns ), - name=unique.name, + name=unique_constraint.name, column=( - resolve_casing(unique.column, columns) if unique.column else "" + resolve_casing(unique_constraint.column, columns) + if unique_constraint.column + else "" ), - conflict=unique.conflict, + conflict=unique_constraint.conflict, ) missing = [ column.name @@ -1705,6 +1714,7 @@ def create_table( foreign_keys: ForeignKeysType | None = None, column_order: list[str] | None = None, not_null: Iterable[str] | None = None, + unique: Iterable[str] | Iterable[Iterable[str]] | None = None, defaults: dict[str, Any] | None = None, hash_id: str | None = None, hash_id_columns: Iterable[str] | None = None, @@ -1726,6 +1736,7 @@ def create_table( :param foreign_keys: List of foreign key definitions for this table :param column_order: List specifying which columns should come first :param not_null: List of columns that should be created as ``NOT NULL`` + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Dictionary specifying default values for columns :param hash_id: Name of column to be used as a primary key containing a hash of the other columns :param hash_id_columns: List of columns to be used when calculating the hash ID for a row @@ -1789,6 +1800,24 @@ def create_table( ) if current_not_null != desired_not_null: should_transform = True + # Any unique changes? + if unique: + existing_unique_columns = { + tuple( + resolve_casing(column.name, existing_columns) + for column in unique_constraint.columns + ) + for unique_constraint in table.uniques + } + for columns_tuple in resolve_uniques(unique): + if ( + tuple( + resolve_casing(c, existing_columns) for c in columns_tuple + ) + not in existing_unique_columns + ): + should_transform = True + break # How about defaults? if ( defaults @@ -1805,6 +1834,7 @@ def create_table( drop=columns_to_drop, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, pk=pk, ) @@ -1816,6 +1846,7 @@ def create_table( foreign_keys=foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, hash_id=hash_id, hash_id_columns=hash_id_columns, @@ -1830,6 +1861,7 @@ def create_table( foreign_keys=foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, hash_id=hash_id, hash_id_columns=hash_id_columns, @@ -2290,6 +2322,7 @@ def __init__( foreign_keys: ForeignKeysType | None = None, column_order: list[str] | None = None, not_null: Iterable[str] | None = None, + unique: Iterable[str] | Iterable[Iterable[str]] | None = None, defaults: dict[str, Any] | None = None, batch_size: int = 100, hash_id: str | None = None, @@ -2308,6 +2341,7 @@ def __init__( "foreign_keys": foreign_keys, "column_order": column_order, "not_null": not_null, + "unique": unique, "defaults": defaults, "batch_size": batch_size, "hash_id": hash_id, @@ -2387,6 +2421,13 @@ def table_checks(self) -> list[Check]: "Table-level CHECK constraints on this table." return [check for check in self.checks if not check.column] + @property + def uniques(self) -> list[Unique]: + "List of column-level and table-level UNIQUE constraints on this table." + if not self.exists() or self.virtual_table_using is not None: + return [] + return parse_uniques(self.schema) + def get(self, pk_values: list | tuple | str | int) -> dict: """ Return row (as dictionary) for the specified primary key. @@ -2544,6 +2585,7 @@ def create( foreign_keys: ForeignKeysType | None | Default = DEFAULT, column_order: list[str] | None | Default = DEFAULT, not_null: Iterable[str] | None | Default = DEFAULT, + unique: Iterable[str] | Iterable[Iterable[str]] | None | Default = DEFAULT, defaults: dict[str, Any] | None | Default = DEFAULT, hash_id: str | None | Default = DEFAULT, hash_id_columns: Iterable[str] | None | Default = DEFAULT, @@ -2564,6 +2606,7 @@ def create( :param foreign_keys: List of foreign key definitions for this table :param column_order: List specifying which columns should come first :param not_null: List of columns that should be created as ``NOT NULL`` + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Dictionary specifying default values for columns :param hash_id: Name of column to be used as a primary key containing a hash of the other columns :param hash_id_columns: List of columns to be used when calculating the hash ID for a row @@ -2579,6 +2622,7 @@ def create( foreign_keys = self.value_or_default("foreign_keys", foreign_keys) column_order = self.value_or_default("column_order", column_order) not_null = self.value_or_default("not_null", not_null) + unique = self.value_or_default("unique", unique) defaults = self.value_or_default("defaults", defaults) hash_id = self.value_or_default("hash_id", hash_id) hash_id_columns = self.value_or_default("hash_id_columns", hash_id_columns) @@ -2595,6 +2639,8 @@ def create( self._defaults["column_order"] = column_order if not_null is not None: self._defaults["not_null"] = not_null + if unique is not None: + self._defaults["unique"] = unique if defaults is not None: self._defaults["defaults"] = defaults if hash_id is not None: @@ -2615,6 +2661,7 @@ def create( foreign_keys=foreign_keys, # type: ignore[arg-type] column_order=column_order, # type: ignore[arg-type] not_null=not_null, # type: ignore[arg-type] + unique=unique, # type: ignore[arg-type] defaults=defaults, # type: ignore[arg-type] hash_id=hash_id, # type: ignore[arg-type] hash_id_columns=hash_id_columns, # type: ignore[arg-type] @@ -2648,6 +2695,7 @@ def transform( drop: Iterable | None = None, pk: Any | None = DEFAULT, not_null: Iterable[str] | None = None, + unique: Iterable[str] | Iterable[Iterable[str]] | None = None, defaults: dict[str, Any] | None = None, drop_foreign_keys: Iterable[str] | None = None, add_foreign_keys: ForeignKeysType | None = None, @@ -2672,6 +2720,7 @@ def transform( :param drop: Columns to drop :param pk: New primary key for the table :param not_null: Columns to set as ``NOT NULL`` + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Default values for columns :param drop_foreign_keys: Foreign key constraints to remove - a column name drops any foreign key that column participates in, a tuple of column names @@ -2693,6 +2742,7 @@ def transform( drop=drop, pk=pk, not_null=not_null, + unique=unique, defaults=defaults, drop_foreign_keys=drop_foreign_keys, add_foreign_keys=add_foreign_keys, @@ -2778,6 +2828,7 @@ def transform_sql( drop: Iterable | None = None, pk: Any | None = DEFAULT, not_null: Iterable[str] | None = None, + unique: Iterable[str] | Iterable[Iterable[str]] | None = None, defaults: dict[str, Any] | None = None, drop_foreign_keys: Iterable | None = None, add_foreign_keys: ForeignKeysType | None = None, @@ -2795,6 +2846,7 @@ def transform_sql( :param drop: Columns to drop :param pk: New primary key for the table :param not_null: Columns to set as ``NOT NULL`` + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Default values for columns :param drop_foreign_keys: Foreign key constraints to remove - a column name drops any foreign key that column participates in, a tuple of column names @@ -2833,6 +2885,11 @@ def transform_sql( } elif isinstance(not_null, set): not_null = {resolve_casing(c, existing_columns) for c in not_null} + if unique is not None: + unique = [ + tuple(resolve_casing(c, existing_columns) for c in columns_tuple) + for columns_tuple in resolve_uniques(unique) + ] if defaults is not None: defaults = { resolve_casing(c, existing_columns): v for c, v in defaults.items() @@ -2872,19 +2929,21 @@ def transform_sql( ) create_table_uniques: list[Unique] = [] - for unique in existing_uniques: + for unique_constraint in existing_uniques: columns = tuple( UniqueColumn( resolve_casing(column.name, existing_columns), collation=column.collation, order=column.order, ) - for column in unique.columns + for column in unique_constraint.columns ) if any(column.name in drop for column in columns): continue owner = ( - resolve_casing(unique.column, existing_columns) if unique.column else "" + resolve_casing(unique_constraint.column, existing_columns) + if unique_constraint.column + else "" ) create_table_uniques.append( Unique( @@ -2896,11 +2955,26 @@ def transform_sql( ) for column in columns ), - name=unique.name, + name=unique_constraint.name, column=rename.get(owner) or owner, - conflict=unique.conflict, + conflict=unique_constraint.conflict, ) ) + if unique is not None: + existing_unique_columns = { + tuple(column.name for column in unique_constraint.columns) + for unique_constraint in create_table_uniques + } + for columns_tuple in unique: + renamed_columns = tuple( + rename.get(column) or column for column in columns_tuple + ) + if renamed_columns in existing_unique_columns: + continue + create_table_uniques.append( + Unique(tuple(UniqueColumn(column) for column in renamed_columns)) + ) + existing_unique_columns.add(renamed_columns) create_table_column_comments: dict[str, ColumnComments] = {} for column, comments in existing_column_comments.items(): @@ -4608,6 +4682,7 @@ def insert( foreign_keys: ForeignKeysType | Default | None = DEFAULT, column_order: list[str] | Default | None = DEFAULT, not_null: Iterable[str] | Default | None = DEFAULT, + unique: Iterable[str] | Iterable[Iterable[str]] | Default | None = DEFAULT, defaults: dict[str, Any] | Default | None = DEFAULT, hash_id: str | Default | None = DEFAULT, hash_id_columns: Iterable[str] | Default | None = DEFAULT, @@ -4637,6 +4712,7 @@ def insert( :param column_order: List of strings specifying a full or partial column order to use when creating the table. :param not_null: Set of strings specifying columns that should be ``NOT NULL``. + :param unique: Columns to constrain as ``UNIQUE``. Use a string for a single column or a tuple for a compound constraint :param defaults: Dictionary specifying default values for specific columns. :param hash_id: Name of a column to create and use as a primary key, where the value of that primary key will be derived as a SHA1 hash of the other column values @@ -4658,6 +4734,7 @@ def insert( foreign_keys=foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, hash_id=hash_id, hash_id_columns=hash_id_columns, @@ -4677,6 +4754,7 @@ def insert_all( foreign_keys: ForeignKeysType | Default | None = DEFAULT, column_order: list[str] | Default | None = DEFAULT, not_null: Iterable[str] | Default | None = DEFAULT, + unique: Iterable[str] | Iterable[Iterable[str]] | Default | None = DEFAULT, defaults: dict[str, Any] | Default | None = DEFAULT, batch_size: int | Default = DEFAULT, hash_id: str | Default | None = DEFAULT, @@ -4702,6 +4780,7 @@ def insert_all( foreign_keys = self.value_or_default("foreign_keys", foreign_keys) column_order = self.value_or_default("column_order", column_order) not_null = self.value_or_default("not_null", not_null) + unique = self.value_or_default("unique", unique) defaults = self.value_or_default("defaults", defaults) batch_size = self.value_or_default("batch_size", batch_size) hash_id = self.value_or_default("hash_id", hash_id) @@ -4850,6 +4929,7 @@ def insert_all( foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, hash_id=hash_id, hash_id_columns=hash_id_columns, @@ -5035,6 +5115,7 @@ def upsert( foreign_keys: ForeignKeysType | Default | None = DEFAULT, column_order: list[str] | Default | None = DEFAULT, not_null: Iterable[str] | Default | None = DEFAULT, + unique: Iterable[str] | Iterable[Iterable[str]] | Default | None = DEFAULT, defaults: dict[str, Any] | Default | None = DEFAULT, hash_id: str | Default | None = DEFAULT, hash_id_columns: Iterable[str] | Default | None = DEFAULT, @@ -5056,6 +5137,7 @@ def upsert( foreign_keys=foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, hash_id=hash_id, hash_id_columns=hash_id_columns, @@ -5073,6 +5155,7 @@ def upsert_all( foreign_keys: ForeignKeysType | Default | None = DEFAULT, column_order: list[str] | Default | None = DEFAULT, not_null: Iterable[str] | Default | None = DEFAULT, + unique: Iterable[str] | Iterable[Iterable[str]] | Default | None = DEFAULT, defaults: dict[str, Any] | Default | None = DEFAULT, batch_size: int | Default = DEFAULT, hash_id: str | Default | None = DEFAULT, @@ -5093,6 +5176,7 @@ def upsert_all( foreign_keys=foreign_keys, column_order=column_order, not_null=not_null, + unique=unique, defaults=defaults, batch_size=batch_size, hash_id=hash_id, @@ -5521,6 +5605,31 @@ def resolve_extracts( return extracts +def resolve_uniques( + unique: "Iterable[str] | Iterable[Iterable[str]] | None", +) -> list[tuple[str, ...]]: + """Normalize a ``unique=`` argument into a list of column tuples. + + Each item describes one ``UNIQUE`` constraint. A plain string names a + single column, while a list or tuple names a compound constraint across + several columns:: + + unique=["email"] # one single-column constraint + unique=["email", "username"] # two single-column constraints + unique=[("a", "b")] # one compound constraint + unique=["email", ("a", "b")] # one of each + """ + if unique is None: + return [] + resolved: list[tuple[str, ...]] = [] + for item in unique: + if isinstance(item, str): + resolved.append((item,)) + else: + resolved.append(tuple(item)) + return resolved + + def _decode_default_value(value: str) -> object: if value.startswith("'") and value.endswith("'"): # It's a string; unescape doubled single quotes diff --git a/tests/test_create.py b/tests/test_create.py index b738df952..5cb1e587e 100644 --- a/tests/test_create.py +++ b/tests/test_create.py @@ -1739,3 +1739,146 @@ def test_chained_create_sets_pks(fresh_db): {"id": int, "name": str, "color": str} ) assert table.pks == ["id"] + + +def test_create_table_with_unique(fresh_db): + table = fresh_db.create_table( + "users", {"id": int, "email": str}, pk="id", unique=["email"] + ) + assert ( + 'CREATE TABLE "users" (\n "id" INTEGER PRIMARY KEY,\n "email" TEXT,\n UNIQUE ("email")\n)' + == table.schema + ) + assert [("email",)] == [tuple(c.name for c in u.columns) for u in table.uniques] + + +def test_create_table_with_multiple_unique_columns(fresh_db): + table = fresh_db.create_table( + "users", + {"id": int, "email": str, "username": str}, + pk="id", + unique=["email", "username"], + ) + assert ( + 'CREATE TABLE "users" (\n "id" INTEGER PRIMARY KEY,\n "email" TEXT,\n' + ' "username" TEXT,\n UNIQUE ("email"),\n UNIQUE ("username")\n)' + == table.schema + ) + + +def test_create_table_with_compound_unique(fresh_db): + table = fresh_db.create_table( + "memberships", + {"account_id": int, "email": str}, + unique=[("account_id", "email")], + ) + assert ( + 'CREATE TABLE "memberships" (\n "account_id" INTEGER,\n "email" TEXT,\n' + ' UNIQUE ("account_id", "email")\n)' == table.schema + ) + assert [("account_id", "email")] == [ + tuple(c.name for c in u.columns) for u in table.uniques + ] + + +def test_create_table_with_mixed_unique(fresh_db): + table = fresh_db.create_table( + "t", {"a": str, "b": str, "c": int}, unique=["a", ("b", "c")] + ) + assert ( + 'CREATE TABLE "t" (\n "a" TEXT,\n "b" TEXT,\n "c" INTEGER,\n' + ' UNIQUE ("a"),\n UNIQUE ("b", "c")\n)' == table.schema + ) + + +def test_create_table_with_unique_bad_column(fresh_db): + with pytest.raises(AlterError): + fresh_db.create_table("users", {"id": int}, pk="id", unique=["nope"]) + + +def test_create_table_with_unique_enforced(fresh_db): + table = fresh_db.create_table( + "users", {"id": int, "email": str}, pk="id", unique=["email"] + ) + table.insert({"id": 1, "email": "one@example.com"}) + with pytest.raises(sqlite3.IntegrityError): + table.insert({"id": 2, "email": "one@example.com"}) + + +def test_create_table_with_compound_unique_enforced(fresh_db): + table = fresh_db.create_table( + "memberships", + {"account_id": int, "email": str}, + unique=[("account_id", "email")], + ) + table.insert({"account_id": 1, "email": "one@example.com"}) + # Different account_id, same email: allowed for a compound constraint + table.insert({"account_id": 2, "email": "one@example.com"}) + with pytest.raises(sqlite3.IntegrityError): + table.insert({"account_id": 1, "email": "one@example.com"}) + + +def test_unique_persists_after_create_655(fresh_db): + """When unique is passed to create(), it should be stored in _defaults.""" + table = fresh_db.table("users") + table.create({"id": int, "email": str}, pk="id", unique=["email"]) + assert table._defaults["unique"] == ["email"] + + +def test_transform_add_unique(fresh_db): + table = fresh_db.create_table("users", {"id": int, "email": str}, pk="id") + assert "UNIQUE" not in table.schema + table.transform(unique=["email"]) + assert 'UNIQUE ("email")' in table.schema + # Applying the same constraint again should not duplicate it + table.transform(unique=["email"]) + assert table.schema.count("UNIQUE") == 1 + + +def test_transform_add_compound_unique(fresh_db): + table = fresh_db.create_table("t", {"a": str, "b": str, "c": int}) + table.transform(unique=[("a", "b")]) + assert 'UNIQUE ("a", "b")' in table.schema + + +def test_transform_add_unique_renamed_column(fresh_db): + table = fresh_db.create_table("t", {"a": str, "b": str}) + table.transform(rename={"a": "x"}, unique=["x"]) + assert 'UNIQUE ("x")' in table.schema + + +def test_create_transform_true_applies_unique(fresh_db): + fresh_db.create_table("users", {"id": int, "email": str}, pk="id") + table = fresh_db.create_table( + "users", {"id": int, "email": str}, pk="id", unique=["email"], transform=True + ) + assert 'UNIQUE ("email")' in table.schema + + +def test_insert_with_unique(fresh_db): + table = fresh_db["users"].insert( + {"id": 1, "email": "one@example.com"}, pk="id", unique=["email"] + ) + assert 'UNIQUE ("email")' in table.schema + + +def test_insert_all_with_unique(fresh_db): + table = fresh_db["users"].insert_all( + [{"id": 1, "email": "one@example.com"}], pk="id", unique=["email"] + ) + assert 'UNIQUE ("email")' in table.schema + with pytest.raises(sqlite3.IntegrityError): + table.insert({"id": 2, "email": "one@example.com"}) + + +def test_upsert_with_unique(fresh_db): + table = fresh_db["users"].upsert( + {"id": 1, "email": "one@example.com"}, pk="id", unique=["email"] + ) + assert 'UNIQUE ("email")' in table.schema + + +def test_unique_via_table_defaults(fresh_db): + table = fresh_db.table("users", unique=["email"]) + table.insert({"id": 1, "email": "one@example.com"}, pk="id") + assert 'UNIQUE ("email")' in table.schema