Merge pull request fastapi#4 from watkinsm/dynamic-model-creation

feat: add dynamic model creation

Author: leynier

sqlmodel/__init__.py

@@ -137,3 +137,4 @@
 from .main import SQLModel as SQLModel
 from .main import Field as Field
 from .main import Relationship as Relationship
+from .main import create_model as create_model

sqlmodel/main.py

@@ -644,3 +644,46 @@ def _calculate_keys(
     @declared_attr  # type: ignore
     def __tablename__(cls) -> str:
         return cls.__name__.lower()
+
+
+def create_model(
+    model_name: str,
+    field_definitions: Dict[str, Tuple[Any, Any]],
+    *,
+    __module__: str = __name__,
+    **kwargs,
+) -> Type[SQLModelMetaclass]:
+    """
+    Dynamically create a model, similar to the Pydantic `create_model()` method
+
+    :param model_name: name of the created model
+    :param field_definitions: data fields of the create model
+    :param __module__: module of the created model
+    :param **kwargs: Other keyword arguments to pass to the metaclass constructor, e.g. table=True
+    """
+    fields = {}
+    annotations = {}
+
+    for f_name, f_def in field_definitions.items():
+        if f_name.startswith("_"):
+            raise ValueError("Field names may not start with an underscore")
+        try:
+            if isinstance(f_def, tuple) and len(f_def) > 1:
+                f_annotation, f_value = f_def
+            elif isinstance(f_def, tuple):
+                f_annotation, f_value = f_def[0], Field(nullable=False)
+            else:
+                f_annotation, f_value = f_def, Field(nullable=False)
+        except ValueError as e:
+            raise ConfigError(
+                "field_definitions values must be either a tuple of (<type_annotation>, <default_value>)"
+                "or just a type annotation [or a 1-tuple of (<type_annotation>,)]"
+            ) from e
+
+        if f_annotation:
+            annotations[f_name] = f_annotation
+        fields[f_name] = f_value
+
+    namespace = {"__annotations__": annotations, "__module__": __module__, **fields}
+
+    return type(model_name, (SQLModel,), namespace, **kwargs)

tests/test_create_model.py

@@ -0,0 +1,46 @@
+from typing import Optional
+
+from sqlmodel import Field, Session, SQLModel, create_engine, create_model
+
+
+def test_create_model(clear_sqlmodel):
+    """
+    Test dynamic model creation, query, and deletion
+    """
+
+    hero = create_model(
+        "Hero",
+        {
+            "id": (Optional[int], Field(default=None, primary_key=True)),
+            "name": str,
+            "secret_name": (str,),  # test 1-tuple
+            "age": (Optional[int], None),
+        },
+        table=True,
+    )
+
+    hero_1 = 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:
+        query_hero = session.query(hero).first()
+        assert query_hero
+        assert query_hero.id == hero_1.id
+        assert query_hero.name == hero_1.name
+        assert query_hero.secret_name == hero_1.secret_name
+        assert query_hero.age == hero_1.age
+
+    with Session(engine) as session:
+        session.delete(hero_1)
+        session.commit()
+
+    with Session(engine) as session:
+        query_hero = session.query(hero).first()
+        assert not query_hero