Merge pull request #2 from honglei/main

get_column_from_field support functional sa_column

Author: mbsantiago

sqlmodel/main.py

@@ -55,10 +55,7 @@
 else:
     from typing_extensions import get_args, get_origin
 
-if sys.version_info >= (3, 9):
-    from typing import Annotated
-else:
-    from typing_extensions import Annotated
+from typing_extensions import Annotated, _AnnotatedAlias
 
 _T = TypeVar("_T")
 NoArgAnyCallable = Callable[[], Any]
@@ -167,7 +164,7 @@ def Field(
     unique: bool = False,
     nullable: Union[bool, PydanticUndefinedType] = PydanticUndefined,
     index: Union[bool, PydanticUndefinedType] = PydanticUndefined,
-    sa_column: Union[Column, PydanticUndefinedType, types.FunctionType] = PydanticUndefined,  # type: ignore
+    sa_column: Union[Column, PydanticUndefinedType, Callable[[], Column]] = PydanticUndefined,  # type: ignore
     sa_column_args: Union[Sequence[Any], PydanticUndefinedType] = PydanticUndefined,
     sa_column_kwargs: Union[
         Mapping[str, Any], PydanticUndefinedType
@@ -440,17 +437,19 @@ def __init__(
             ModelMetaclass.__init__(cls, classname, bases, dict_, **kw)
 
 
+def _is_optional_or_union(type_: Optional[type]) -> bool:
+    if sys.version_info >= (3, 10):
+        return get_origin(type_) in (types.UnionType, Union)
+    else:
+        return get_origin(type_) is Union
+
+
 def get_sqlalchemy_type(field: FieldInfo) -> Any:
-    type_: Optional[type] = field.annotation
+    type_: Optional[type] | _AnnotatedAlias = field.annotation
 
     # Resolve Optional/Union fields
-    def is_optional_or_union(type_: Optional[type]) -> bool:
-        if sys.version_info >= (3, 10):
-            return get_origin(type_) in (types.UnionType, Union)
-        else:
-            return get_origin(type_) is Union
 
-    if type_ is not None and is_optional_or_union(type_):
+    if type_ is not None and _is_optional_or_union(type_):
         bases = get_args(type_)
         if len(bases) > 2:
             raise RuntimeError(
@@ -462,14 +461,20 @@ def is_optional_or_union(type_: Optional[type]) -> bool:
     # UrlConstraints(max_length=512,
     # allowed_schemes=['smb', 'ftp', 'file']) ]
     if type_ is pydantic.AnyUrl:
-        meta = field.metadata[0]
-        return AutoString(length=meta.max_length)
+        if field.metadata:
+            meta = field.metadata[0]
+            return AutoString(length=meta.max_length)
+        else:
+            return AutoString
 
-    if get_origin(type_) is Annotated:
+    org_type = get_origin(type_)
+    if org_type is Annotated:
         type2 = get_args(type_)[0]
         if type2 is pydantic.AnyUrl:
             meta = get_args(type_)[1]
             return AutoString(length=meta.max_length)
+    elif org_type is pydantic.AnyUrl and type(type_) is _AnnotatedAlias:
+        return AutoString(type_.__metadata__[0].max_length)
 
     # The 3rd is PydanticGeneralMetadata
     metadata = _get_field_metadata(field)
@@ -519,11 +524,18 @@ def is_optional_or_union(type_: Optional[type]) -> bool:
 
 
 def get_column_from_field(field: FieldInfo) -> Column:  # type: ignore
+    """
+    sa_column > field attributes > annotation info
+    """
     sa_column = getattr(field, "sa_column", PydanticUndefined)
     if isinstance(sa_column, Column):
         return sa_column
     if isinstance(sa_column, MappedColumn):
         return sa_column.column
+    if isinstance(sa_column, types.FunctionType):
+        col = sa_column()
+        assert isinstance(col, Column)
+        return col
     sa_type = get_sqlalchemy_type(field)
     primary_key = getattr(field, "primary_key", False)
     index = getattr(field, "index", PydanticUndefined)
@@ -587,6 +599,10 @@ def __new__(cls, *args: Any, **kwargs: Any) -> Any:
         # in the Pydantic model so that when SQLAlchemy sets attributes that are
         # added (e.g. when querying from DB) to the __fields_set__, this already exists
         object.__setattr__(new_object, "__pydantic_fields_set__", set())
+        if not hasattr(new_object, "__pydantic_extra__"):
+            object.__setattr__(new_object, "__pydantic_extra__", None)
+        if not hasattr(new_object, "__pydantic_private__"):
+            object.__setattr__(new_object, "__pydantic_private__", None)
         return new_object
 
     def __init__(__pydantic_self__, **data: Any) -> None:
@@ -636,7 +652,10 @@ def model_validate(
         # remove defaults so they don't get validated
         data = {}
         for key, value in validated:
-            field = cls.model_fields[key]
+            field = cls.model_fields.get(key)
+
+            if field is None:
+                continue
 
             if (
                 hasattr(field, "default")
@@ -661,10 +680,11 @@ def _is_field_noneable(field: FieldInfo) -> bool:
             return False
         if field.annotation is None or field.annotation is NoneType:
             return True
-        if get_origin(field.annotation) is Union:
+        if _is_optional_or_union(field.annotation):
             for base in get_args(field.annotation):
                 if base is NoneType:
                     return True
+
         return False
     return False
 

sqlmodel/sql/sqltypes.py

@@ -8,7 +8,6 @@
 
 
 class AutoString(types.TypeDecorator):  # type: ignore
-
     impl = types.String
     cache_ok = True
     mysql_default_length = 255

tests/test_class_hierarchy.py

@@ -0,0 +1,78 @@
+import datetime
+import sys
+
+import pytest
+from pydantic import AnyUrl, UrlConstraints
+from sqlmodel import (
+    BigInteger,
+    Column,
+    DateTime,
+    Field,
+    Integer,
+    SQLModel,
+    String,
+    create_engine,
+)
+from typing_extensions import Annotated
+
+MoveSharedUrl = Annotated[
+    AnyUrl, UrlConstraints(max_length=512, allowed_schemes=["smb", "ftp", "file"])
+]
+
+
+@pytest.mark.skipif(sys.version_info < (3, 10), reason="requires python3.10 or higher")
+def test_field_resuse():
+    class BasicFileLog(SQLModel):
+        resourceID: int = Field(
+            sa_column=lambda: Column(Integer, index=True), description="""   """
+        )
+        transportID: Annotated[int | None, Field(description=" for ")] = None
+        fileName: str = Field(
+            sa_column=lambda: Column(String, index=True), description=""" """
+        )
+        fileSize: int | None = Field(
+            sa_column=lambda: Column(BigInteger), ge=0, description=""" """
+        )
+        beginTime: datetime.datetime | None = Field(
+            sa_column=lambda: Column(
+                DateTime(timezone=True),
+                index=True,
+            ),
+            description="",
+        )
+
+    class SendFileLog(BasicFileLog, table=True):
+        id: int | None = Field(
+            sa_column=Column(Integer, primary_key=True, autoincrement=True),
+            description="""   """,
+        )
+        sendUser: str
+        dstUrl: MoveSharedUrl | None
+
+    class RecvFileLog(BasicFileLog, table=True):
+        id: int | None = Field(
+            sa_column=Column(Integer, primary_key=True, autoincrement=True),
+            description="""   """,
+        )
+        recvUser: str
+
+    sqlite_file_name = "database.db"
+    sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+    engine = create_engine(sqlite_url, echo=True)
+    SQLModel.metadata.drop_all(engine)
+    SQLModel.metadata.create_all(engine)
+    SendFileLog(
+        sendUser="j",
+        resourceID=1,
+        fileName="a.txt",
+        fileSize=3234,
+        beginTime=datetime.datetime.now(),
+    )
+    RecvFileLog(
+        sendUser="j",
+        resourceID=1,
+        fileName="a.txt",
+        fileSize=3234,
+        beginTime=datetime.datetime.now(),
+    )

tests/test_model_copy.py

@@ -0,0 +1,50 @@
+from typing import Optional
+
+from sqlmodel import Field, Session, SQLModel, create_engine
+
+
+def test_model_copy(clear_sqlmodel):
+    """Test validation of implicit and explict None values.
+
+    # For consistency with pydantic, validators are not to be called on
+    # arguments that are not explicitly provided.
+
+    https://github.com/tiangolo/sqlmodel/issues/230
+    https://github.com/samuelcolvin/pydantic/issues/1223
+
+    """
+
+    class Hero(SQLModel, table=True):
+        id: Optional[int] = Field(default=None, primary_key=True)
+        name: str
+        secret_name: str
+        age: Optional[int] = None
+
+    hero = Hero(name="Deadpond", secret_name="Dive Wilson", age=25)
+
+    engine = create_engine("sqlite://")
+
+    SQLModel.metadata.create_all(engine)
+
+    with Session(engine) as session:
+        session.add(hero)
+        session.commit()
+        session.refresh(hero)
+
+    model_copy = hero.model_copy(update={"name": "Deadpond Copy"})
+
+    assert (
+        model_copy.name == "Deadpond Copy"
+        and model_copy.secret_name == "Dive Wilson"
+        and model_copy.age == 25
+    )
+
+    db_hero = session.get(Hero, hero.id)
+
+    db_copy = db_hero.model_copy(update={"name": "Deadpond Copy"})
+
+    assert (
+        db_copy.name == "Deadpond Copy"
+        and db_copy.secret_name == "Dive Wilson"
+        and db_copy.age == 25
+    )

tests/test_nullable.py

@@ -1,8 +1,14 @@
 from typing import Optional
 
 import pytest
+from pydantic import AnyUrl, UrlConstraints
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import Field, Session, SQLModel, create_engine
+from typing_extensions import Annotated
+
+MoveSharedUrl = Annotated[
+    AnyUrl, UrlConstraints(max_length=512, allowed_schemes=["smb", "ftp", "file"])
+]
 
 
 def test_nullable_fields(clear_sqlmodel, caplog):
@@ -13,6 +19,8 @@ class Hero(SQLModel, table=True):
         )
         required_value: str
         optional_default_ellipsis: Optional[str] = Field(default=...)
+        optional_no_field: Optional[str]
+        optional_no_field_default: Optional[str] = Field(description="no default")
         optional_default_none: Optional[str] = Field(default=None)
         optional_non_nullable: Optional[str] = Field(
             nullable=False,
@@ -49,6 +57,13 @@ class Hero(SQLModel, table=True):
         str_default_str_nullable: str = Field(default="default", nullable=True)
         str_default_ellipsis_non_nullable: str = Field(default=..., nullable=False)
         str_default_ellipsis_nullable: str = Field(default=..., nullable=True)
+        base_url: AnyUrl
+        optional_url: Optional[MoveSharedUrl] = Field(default=None, description="")
+        url: MoveSharedUrl
+        annotated_url: Annotated[MoveSharedUrl, Field(description="")]
+        annotated_optional_url: Annotated[
+            Optional[MoveSharedUrl], Field(description="")
+        ] = None
 
     engine = create_engine("sqlite://", echo=True)
     SQLModel.metadata.create_all(engine)
@@ -59,6 +74,8 @@ class Hero(SQLModel, table=True):
     assert "primary_key INTEGER NOT NULL," in create_table_log
     assert "required_value VARCHAR NOT NULL," in create_table_log
     assert "optional_default_ellipsis VARCHAR NOT NULL," in create_table_log
+    assert "optional_no_field VARCHAR," in create_table_log
+    assert "optional_no_field_default VARCHAR NOT NULL," in create_table_log
     assert "optional_default_none VARCHAR," in create_table_log
     assert "optional_non_nullable VARCHAR NOT NULL," in create_table_log
     assert "optional_nullable VARCHAR," in create_table_log
@@ -77,6 +94,11 @@ class Hero(SQLModel, table=True):
     assert "str_default_str_nullable VARCHAR," in create_table_log
     assert "str_default_ellipsis_non_nullable VARCHAR NOT NULL," in create_table_log
     assert "str_default_ellipsis_nullable VARCHAR," in create_table_log
+    assert "base_url VARCHAR NOT NULL," in create_table_log
+    assert "optional_url VARCHAR(512), " in create_table_log
+    assert "url VARCHAR(512) NOT NULL," in create_table_log
+    assert "annotated_url VARCHAR(512) NOT NULL," in create_table_log
+    assert "annotated_optional_url VARCHAR(512)," in create_table_log
 
 
 # Test for regression in https://github.com/tiangolo/sqlmodel/issues/420