feat: add unique constraint param to Field function

Author: Raphael Gibson

sqlmodel/main.py

@@ -70,6 +70,7 @@ def __init__(self, default: Any = Undefined, **kwargs: Any) -> None:
         primary_key = kwargs.pop("primary_key", False)
         nullable = kwargs.pop("nullable", Undefined)
         foreign_key = kwargs.pop("foreign_key", Undefined)
+        unique = kwargs.pop("unique", False)
         index = kwargs.pop("index", Undefined)
         sa_column = kwargs.pop("sa_column", Undefined)
         sa_column_args = kwargs.pop("sa_column_args", Undefined)
@@ -89,6 +90,7 @@ def __init__(self, default: Any = Undefined, **kwargs: Any) -> None:
         self.primary_key = primary_key
         self.nullable = nullable
         self.foreign_key = foreign_key
+        self.unique = unique
         self.index = index
         self.sa_column = sa_column
         self.sa_column_args = sa_column_args
@@ -150,6 +152,7 @@ def Field(
     regex: str = None,
     primary_key: bool = False,
     foreign_key: Optional[Any] = None,
+    unique: bool = False,
     nullable: Union[bool, UndefinedType] = Undefined,
     index: Union[bool, UndefinedType] = Undefined,
     sa_column: Union[Column, UndefinedType] = Undefined,
@@ -180,6 +183,7 @@ def Field(
         regex=regex,
         primary_key=primary_key,
         foreign_key=foreign_key,
+        unique=unique,
         nullable=nullable,
         index=index,
         sa_column=sa_column,
@@ -424,12 +428,14 @@ def get_column_from_field(field: ModelField) -> Column:
             nullable = field_nullable
     args = []
     foreign_key = getattr(field.field_info, "foreign_key", None)
+    unique = getattr(field.field_info, "unique", False)
     if foreign_key:
         args.append(ForeignKey(foreign_key))
     kwargs = {
         "primary_key": primary_key,
         "nullable": nullable,
         "index": index,
+        "unique": unique
     }
     sa_default = Undefined
     if field.field_info.default_factory:

tests/test_main.py

@@ -0,0 +1,91 @@
+import pytest
+from typing import Optional
+
+from sqlmodel import Field, Session, SQLModel, create_engine
+from sqlalchemy.exc import IntegrityError
+
+
+def test_should_allow_duplicate_row_if_unique_constraint_is_not_passed(clear_sqlmodel):
+    class Hero(SQLModel, table=True):
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        secret_name: str
+        age: Optional[int] = None
+
+    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
+    hero_2 = Hero(name="Deadpond", secret_name="Dive Wilson")
+
+    engine = create_engine("sqlite://")
+
+    SQLModel.metadata.create_all(engine)
+
+    with Session(engine) as session:
+        session.add(hero_1)
+        session.commit()
+        session.refresh(hero_1)
+
+    with Session(engine) as session:
+        session.add(hero_2)
+        session.commit()
+        session.refresh(hero_2)
+
+    with Session(engine) as session:
+        heroes = session.query(Hero).all()
+        assert len(heroes) == 2
+        assert heroes[0].name == heroes[1].name
+
+
+def test_should_allow_duplicate_row_if_unique_constraint_is_false(clear_sqlmodel):
+    class Hero(SQLModel, table=True):
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        secret_name: str = Field(unique=False)
+        age: Optional[int] = None
+
+    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
+    hero_2 = Hero(name="Deadpond", secret_name="Dive Wilson")
+
+    engine = create_engine("sqlite://")
+
+    SQLModel.metadata.create_all(engine)
+
+    with Session(engine) as session:
+        session.add(hero_1)
+        session.commit()
+        session.refresh(hero_1)
+
+    with Session(engine) as session:
+        session.add(hero_2)
+        session.commit()
+        session.refresh(hero_2)
+
+    with Session(engine) as session:
+        heroes = session.query(Hero).all()
+        assert len(heroes) == 2
+        assert heroes[0].name == heroes[1].name
+
+
+def test_should_raise_exception_when_try_to_duplicate_row_if_unique_constraint_is_true(clear_sqlmodel):
+    class Hero(SQLModel, table=True):
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        secret_name: str = Field(unique=True)
+        age: Optional[int] = None
+
+    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson")
+    hero_2 = Hero(name="Deadpond", secret_name="Dive Wilson")
+
+    engine = create_engine("sqlite://")
+
+    SQLModel.metadata.create_all(engine)
+
+    with Session(engine) as session:
+        session.add(hero_1)
+        session.commit()
+        session.refresh(hero_1)
+
+    with pytest.raises(IntegrityError):
+        with Session(engine) as session:
+            session.add(hero_2)
+            session.commit()
+            session.refresh(hero_2)