Possibility to change default Pydantic to SQLAlchemy mappings

docs/advanced/change-default-type-mapping.md

@@ -0,0 +1,70 @@
+# Change default Pydantic to SQLAlchemy mappings
+
+In most cases, you do not need to know how SQLAlchemy transforms the Python types to the type suitable for storing data
+in the database, and you can use the default mapping.
+
+But in some cases you may need to have possibility to change default mapping provided by SQLmodel. For example to use
+mssql dialect with some UTF-8 data you should use NVARCHAR field (sa.Unicode)
+
+Now changing default mapping is simple to use - see example bellow:
+
+```python
+import sqlmodel.main
+import sqlalchemy as sa
+from sqlmodel import Field, SQLModel
+from typing import Optional
+
+sqlmodel.main.sa_types_map[str] = lambda type_, meta, annotation: sa.Unicode(
+    length=getattr(meta, "max_length", None)
+)
+
+
+class Hero(SQLModel, table=True):
+    id: int = Field(default=None, primary_key=True)
+    name: str = Field(max_length=255)
+    history: Optional[str]
+
+
+assert isinstance(Hero.name.type, sa.Unicode)
+```
+
+# Some details
+
+Let's get little deeper to process of mapping. At the `sqlmodel.main` module defined the `sa_types_map` dictionary,
+which uses the Python types as keys, and the sqlalchemy type or callable that takes the input of 3 parameters and
+returns the sqlalchemy type as values.
+
+Callable format present bellow:
+
+```python
+def map_python_type_to_sa_type(type_: "PythonType", meta: "PydanticMeta", annotation: "FieldAnnotatedType"):
+    return sqlalchemyType(length=getattr(meta, "max_length", None))
+```
+
+* `type_` - used to pass python type, provided by pydantic annotation, cleared from Union/Optional and other wrappers.
+  Can be passed to sa.Enum type to properly store enumerated data.
+* `meta` - pydantic metadata used to store field params e.g. length of str field or precision of decimal field
+* `annotation` - original annotation given by pydantic. Used to provide `type` parameter for PydanticJSONType
+
+## Current mapping
+
+| Python type           | SqlAlchemy type                                                                                                                    |
+|-----------------------|------------------------------------------------------------------------------------------------------------------------------------|
+| Enum                  | `lambda type_, meta, annotation: sa_Enum(type_)`                                                                                   |
+| str                   | `lambda type_, meta, annotation: AutoString(length=getattr(meta, "max_length", None))`                                             |
+| float                 | Float                                                                                                                              |
+| bool                  | Boolean                                                                                                                            |
+| int                   | Integer                                                                                                                            |
+| datetime              | DateTime                                                                                                                           |
+| date                  | Date                                                                                                                               |
+| timedelta             | Interval                                                                                                                           |
+| time                  | Time                                                                                                                               |
+| bytes                 | LargeBinary                                                                                                                        |
+| Decimal               | `lambda type_, meta, annotation: Numeric(precision=getattr(meta, "max_digits", None),scale=getattr(meta, "decimal_places", None))` |
+| ipaddress.IPv4Address | AutoString                                                                                                                         |
+| ipaddress.IPv4Network | AutoString                                                                                                                         |
+| ipaddress.IPv6Address | AutoString                                                                                                                         |
+| ipaddress.IPv6Network | AutoString                                                                                                                         |
+| Path                  | AutoString                                                                                                                         |
+| uuid.UUID             | Uuid                                                                                                                               |
+| BaseModel             | `lambda type_, meta, annotation: PydanticJSONType(type=annotation)`                                                                |

docs/advanced/index.md

@@ -4,7 +4,7 @@ The **Advanced User Guide** is gradually growing, you can already read about som
 
 At some point it will include:
 
-* How to use `async` and `await` with the async session.
+    * How to use `async` and `await` with the async session.
 * How to run migrations.
 * How to combine **SQLModel** models with SQLAlchemy.
 * ...and more. 🤓

docs/advanced/pydantic-json-type.md

@@ -0,0 +1,60 @@
+# Storing Pydantic models at database
+
+In some cases you might need to be able to store Pydantic models as a JSON data instead of create new table for them. You can do it now with new SqlAlchemy type `PydanticJSonType` mapped to BaseModel inside SqlModel.
+
+For example let's add some stats to our heroes and save them at database as JSON data.
+
+At first, we need to create new class `Stats` inherited from pydantic `BaseModel` or even `SqlModel`:
+
+```{.python .annotate  }
+{!./docs_src/advanced/pydantic_json_type/tutorial001.py[ln:8-14]!}
+```
+
+Then create new field `stats` to `Hero` model
+
+```{.python .annotate  hl_lines="6" }
+{!./docs_src/advanced/pydantic_json_type/tutorial001.py[ln:17-22]!}
+```
+And... that's all of you need to do to store pydantic data as JSON at database.
+
+/// details | 👀 Full tutorial preview
+```Python
+{!./docs_src/advanced/pydantic_json_type/tutorial001.py!}
+```
+
+///
+
+Here we define new Pydantic model `Stats` contains statistics of our hero and map this model to SqlModel class `Hero`.
+
+Then we create new instances of Hero model with random generated stats and save it at database.
+
+
+# How to watch for mapped model changes at runtime
+
+In previous example we have one *non bug but feature* - `Stats` model isn't mutable and if we try to load our Hero form database and  then change some stats and call `session.commit()` there no changes will be saved.
+
+Let's see how to avoid it.
+
+At first, we need to inherit our Stats model from `sqlalchemy.ext.mutable.Mutable`:
+```{.python .annotate hl_lines="1" }
+{!./docs_src/advanced/pydantic_json_type/tutorial002.py[ln:10-19]!}
+```
+
+Then map Stats to Hero as shown bellow:
+```{.python .annotate hl_lines="1-4" }
+{!./docs_src/advanced/pydantic_json_type/tutorial002.py[ln:36-39]!}
+```
+
+After all of these actions we can change mutated model, and it will be saved to database after we call `session.commit()`
+
+```{.python .annotate hl_lines="4" }
+{!./docs_src/advanced/pydantic_json_type/tutorial002.py[ln:76-94]!}
+```
+
+/// details | 👀 Full tutorial preview
+
+```Python
+{!./docs_src/advanced/pydantic_json_type/tutorial002.py!}
+```
+
+///

docs_src/advanced/pydantic_json_type/tutorial001.py

@@ -0,0 +1,85 @@
+import random
+from typing import Optional
+
+from pydantic import BaseModel
+from sqlmodel import Field, Session, SQLModel, create_engine, select
+
+
+class Stats(BaseModel):
+    strength: int
+    dexterity: int
+    constitution: int
+    intelligence: int
+    wisdom: int
+    charisma: int
+
+
+class Hero(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    secret_name: str
+    age: Optional[int]
+    stats: Optional[Stats]
+
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+engine = create_engine(sqlite_url, echo=True)
+
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+
+def random_stat():
+    random.seed()
+
+    return Stats(
+        strength=random.randrange(1, 20, 2),
+        dexterity=random.randrange(1, 20, 2),
+        constitution=random.randrange(1, 20, 2),
+        intelligence=random.randrange(1, 20, 2),
+        wisdom=random.randrange(1, 20, 2),
+        charisma=random.randrange(1, 20, 2),
+    )
+
+
+def create_heroes():
+    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", stats=random_stat())
+    hero_2 = Hero(
+        name="Spider-Boy", secret_name="Pedro Parqueador", stats=random_stat()
+    )
+    hero_3 = Hero(
+        name="Rusty-Man", secret_name="Tommy Sharp", age=48, stats=random_stat()
+    )
+
+    with Session(engine) as session:
+        session.add(hero_1)
+        session.add(hero_2)
+        session.add(hero_3)
+
+        session.commit()
+
+
+def select_heroes():
+    with Session(engine) as session:
+        statement = select(Hero).where(Hero.name == "Deadpond")
+        results = session.exec(statement)
+        hero_1 = results.one()
+        print("Hero 1:", hero_1)
+
+        statement = select(Hero).where(Hero.name == "Rusty-Man")
+        results = session.exec(statement)
+        hero_2 = results.one()
+        print("Hero 2:", hero_2)
+
+
+def main():
+    create_db_and_tables()
+    create_heroes()
+    select_heroes()
+
+
+if __name__ == "__main__":
+    main()

docs_src/advanced/pydantic_json_type/tutorial002.py

@@ -0,0 +1,102 @@
+import random
+from typing import Any, Optional
+
+from pydantic import BaseModel
+from sqlalchemy import Column
+from sqlalchemy.ext.mutable import Mutable
+from sqlmodel import Field, Session, SQLModel, create_engine, select
+from sqlmodel.sql.sqltypes import PydanticJSONType
+
+
+class Stats(BaseModel, Mutable):
+    strength: int
+    dexterity: int
+    constitution: int
+    intelligence: int
+    wisdom: int
+    charisma: int
+
+    @classmethod
+    def coerce(cls, key: str, value: Any) -> Optional[Any]:
+        return value
+
+    def __setattr__(self, key, value):
+        # set the attribute
+        object.__setattr__(self, key, value)
+
+        # alert all parents to the change
+        self.changed()
+
+
+class Hero(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    name: str = Field(index=True)
+    secret_name: str
+    age: Optional[int]
+    stats: Stats = Field(
+        default_factory=None,
+        sa_column=Column(Stats.as_mutable(PydanticJSONType(type=Stats))),
+    )
+
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+engine = create_engine(sqlite_url, echo=True)
+
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+
+def random_stat():
+    random.seed()
+
+    return Stats(
+        strength=random.randrange(1, 20, 2),
+        dexterity=random.randrange(1, 20, 2),
+        constitution=random.randrange(1, 20, 2),
+        intelligence=random.randrange(1, 20, 2),
+        wisdom=random.randrange(1, 20, 2),
+        charisma=random.randrange(1, 20, 2),
+    )
+
+
+def create_hero():
+    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", stats=random_stat())
+
+    with Session(engine) as session:
+        session.add(hero_1)
+
+        session.commit()
+
+
+def mutate_hero():
+    with Session(engine) as session:
+        statement = select(Hero).where(Hero.name == "Deadpond")
+        results = session.exec(statement)
+        hero_1 = results.one()
+
+        print("Hero 1:", hero_1.stats)
+
+        hero_1.stats.strength = 100500
+        session.commit()
+
+    with Session(engine) as session:
+        statement = select(Hero).where(Hero.name == "Deadpond")
+        results = session.exec(statement)
+        hero_1 = results.one()
+
+        print("Hero 1 strength:", hero_1.stats.strength)
+
+        print("Hero 1:", hero_1)
+
+
+def main():
+    create_db_and_tables()
+    create_hero()
+    mutate_hero()
+
+
+if __name__ == "__main__":
+    main()

mkdocs.yml

@@ -98,6 +98,8 @@ nav:
   - Advanced User Guide:
     - advanced/index.md
     - advanced/decimal.md
+    - advanced/change-default-type-mapping.md
+    - advanced/pydantic-json-type.md
   - alternatives.md
   - help.md
   - contributing.md

sqlmodel/_compat.py

@@ -17,6 +17,7 @@
     Union,
 )
 
+from annotated_types import BaseMetadata
 from pydantic import VERSION as PYDANTIC_VERSION
 from pydantic.fields import FieldInfo
 from typing_extensions import get_args, get_origin
@@ -190,8 +191,9 @@ def get_type_from_field(field: Any) -> Any:
 
     def get_field_metadata(field: Any) -> Any:
         for meta in field.metadata:
-            if isinstance(meta, PydanticMetadata):
+            if isinstance(meta, (PydanticMetadata, BaseMetadata)):
                 return meta
+
         return FakeMetadata()
 
     def post_init_field_info(field_info: FieldInfo) -> None: