RFC: Modelschema rework

ninja/orm/factory.py

@@ -1,9 +1,21 @@
 import itertools
-from typing import Any, Dict, Iterator, List, Optional, Set, Tuple, Type, Union, cast
+from typing import (
+    Any,
+    Dict,
+    Iterator,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Type,
+    Union,
+    cast,
+)
 
 from django.db.models import Field as DjangoField
 from django.db.models import ManyToManyRel, ManyToOneRel, Model
 from pydantic import create_model as create_pydantic_model
+from typing_extensions import Literal
 
 from ninja.errors import ConfigError
 from ninja.orm.fields import get_schema_field
@@ -38,36 +50,31 @@ def create_schema(
         *,
         name: str = "",
         depth: int = 0,
-        fields: Optional[List[str]] = None,
+        fields: Optional[Union[List[str], Literal["__all__"]]] = "__all__",
         exclude: Optional[List[str]] = None,
         optional_fields: Optional[List[str]] = None,
         custom_fields: Optional[List[Tuple[str, Any, Any]]] = None,
         base_class: Type[Schema] = Schema,
+        primary_key_optional: bool = True,
     ) -> Type[Schema]:
         name = name or model.__name__
 
-        if fields and exclude:
-            raise ConfigError("Only one of 'fields' or 'exclude' should be set.")
-
         key = self.get_key(
             model, name, depth, fields, exclude, optional_fields, custom_fields
         )
-        if key in self.schemas:
-            return self.schemas[key]
 
-        model_fields_list = list(self._selected_model_fields(model, fields, exclude))
-        if optional_fields:
-            if optional_fields == "__all__":
-                optional_fields = [f.name for f in model_fields_list]
+        schema = self.get_schema(key)
+        if schema is not None:
+            return schema
 
-        definitions = {}
-        for fld in model_fields_list:
-            python_type, field_info = get_schema_field(
-                fld,
-                depth=depth,
-                optional=optional_fields and (fld.name in optional_fields),
-            )
-            definitions[fld.name] = (python_type, field_info)
+        definitions = self.convert_django_fields(
+            model,
+            depth=depth,
+            fields=fields,
+            exclude=exclude,
+            optional_fields=optional_fields,
+            primary_key_optional=primary_key_optional,
+        )
 
         if custom_fields:
             for fld_name, python_type, field_info in custom_fields:
@@ -78,32 +85,59 @@ def create_schema(
         if name in self.schema_names:
             name = self._get_unique_name(name)
 
-        schema: Type[Schema] = create_pydantic_model(
+        schema = create_pydantic_model(
             name,
             __config__=None,
             __base__=base_class,
             __module__=base_class.__module__,
             __validators__={},
             **definitions,
         )  # type: ignore
-        # __model_name: str,
-        # *,
-        # __config__: ConfigDict | None = None,
-        # __base__: None = None,
-        # __module__: str = __name__,
-        # __validators__: dict[str, AnyClassMethod] | None = None,
-        # __cls_kwargs__: dict[str, Any] | None = None,
-        # **field_definitions: Any,
+
         self.schemas[key] = schema
         self.schema_names.add(name)
         return schema
 
+    def get_schema(self, key: SchemaKey) -> Union[Type[Schema], None]:
+        if key in self.schemas:
+            return self.schemas[key]
+        return None
+
+    def convert_django_fields(
+        self,
+        model: Type[Model],
+        *,
+        depth: int = 0,
+        fields: Optional[Union[List[str], Literal["__all__"]]] = None,
+        exclude: Optional[List[str]] = None,
+        optional_fields: Optional[List[str]] = None,
+        primary_key_optional: bool = True,
+    ) -> Dict[str, Tuple[Any, Any]]:
+        if (fields and fields != "__all__") and exclude:
+            raise ConfigError("Only one of 'fields' or 'exclude' should be set.")
+
+        model_fields_list = list(self._selected_model_fields(model, fields, exclude))
+        if optional_fields and optional_fields == "__all__":
+            optional_fields = [f.name for f in model_fields_list]
+
+        definitions = {}
+        for fld in model_fields_list:
+            python_type, field_info = get_schema_field(
+                fld,
+                depth=depth,
+                optional=optional_fields and (fld.name in optional_fields),
+                primary_key_optional=primary_key_optional,
+            )
+            definitions[fld.name] = (python_type, field_info)
+
+        return definitions
+
     def get_key(
         self,
         model: Type[Model],
         name: str,
         depth: int,
-        fields: Union[str, List[str], None],
+        fields: Optional[Union[List[str], Literal["__all__"]]],
         exclude: Optional[List[str]],
         optional_fields: Optional[Union[List[str], str]],
         custom_fields: Optional[List[Tuple[str, str, Any]]],
@@ -131,15 +165,19 @@ def _get_unique_name(self, name: str) -> str:
     def _selected_model_fields(
         self,
         model: Type[Model],
-        fields: Optional[List[str]] = None,
+        fields: Optional[Union[List[str], Literal["__all__"]]] = None,
         exclude: Optional[List[str]] = None,
     ) -> Iterator[DjangoField]:
         "Returns iterator for model fields based on `exclude` or `fields` arguments"
         all_fields = {f.name: f for f in self._model_fields(model)}
 
+        if fields == "__all__":
+            fields = None
+
         if not fields and not exclude:
             for f in all_fields.values():
                 yield f
+            return
 
         invalid_fields = (set(fields or []) | set(exclude or [])) - all_fields.keys()
         if invalid_fields:

ninja/orm/fields.py

@@ -110,7 +110,11 @@ def _validate(cls, v: Any, _):
 
 @no_type_check
 def get_schema_field(
-    field: DjangoField, *, depth: int = 0, optional: bool = False
+    field: DjangoField,
+    *,
+    depth: int = 0,
+    optional: bool = False,
+    primary_key_optional: bool = True,
 ) -> Tuple:
     "Returns pydantic field from django's model field"
     alias = None
@@ -150,7 +154,7 @@ def get_schema_field(
         internal_type = field.get_internal_type()
         python_type = TYPES[internal_type]
 
-        if field.primary_key or blank or null or optional:
+        if (field.primary_key and primary_key_optional) or blank or null or optional:
             default = None
             nullable = True
 

ninja/orm/metaclass.py

@@ -1,68 +1,63 @@
 import warnings
-from typing import Any, List, Optional, Union, no_type_check
+from inspect import getmembers
+from typing import List, Optional, Type, Union, no_type_check
 
 from django.db.models import Model as DjangoModel
-from pydantic.dataclasses import dataclass
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
+from typing_extensions import Literal, Self
 
 from ninja.errors import ConfigError
-from ninja.orm.factory import create_schema
+from ninja.orm.factory import factory
 from ninja.schema import ResolverMetaclass, Schema
 
-_is_modelschema_class_defined = False
-
-
-@dataclass
-class MetaConf:
-    model: Any
-    fields: Optional[List[str]] = None
-    exclude: Union[List[str], str, None] = None
-    fields_optional: Union[List[str], str, None] = None
-
-    @staticmethod
-    def from_schema_class(name: str, namespace: dict) -> "MetaConf":
-        if "Meta" in namespace:
-            meta = namespace["Meta"]
-            model = meta.model
-            fields = getattr(meta, "fields", None)
-            exclude = getattr(meta, "exclude", None)
-            optional_fields = getattr(meta, "fields_optional", None)
-
-        elif "Config" in namespace:
-            config = namespace["Config"]
-            model = config.model
-            fields = getattr(config, "model_fields", None)
-            exclude = getattr(config, "model_exclude", None)
-            optional_fields = getattr(config, "model_fields_optional", None)
 
+class MetaConf(BaseModel):
+    """
+    Mirrors the relevant arguments for create_schema
+
+    model: Django model being used to create the Schema
+    fields: List of field names in the model to use. Defaults to '__all__' which includes all fields
+    exclude: List of field names to exclude
+    optional_fields: List of field names which will be optional, can also take '__all__'
+    depth: If > 0 schema will also be created for the nested ForeignKeys and Many2Many (with the provided depth of lookup)
+    primary_key_optional: Defaults to True, controls if django's primary_key=True field in the provided model is required
+
+    fields_optional: same as optional_fields, deprecated in order to match `create_schema()` API
+    """
+
+    model: Optional[Type[DjangoModel]] = None
+    # aliased for Config
+    fields: Union[List[str], Literal["__all__"], None] = Field(
+        None, validation_alias=AliasChoices("fields", "model_fields")
+    )
+    exclude: Optional[List[str]] = None
+    optional_fields: Union[List[str], Literal["__all__"], None] = None
+    depth: int = 0
+    primary_key_optional: Optional[bool] = None
+    # deprecated
+    fields_optional: Union[List[str], Literal["__all__"], None] = Field(
+        default=None, exclude=True
+    )
+
+    model_config = ConfigDict(extra="forbid")
+
+    @model_validator(mode="after")
+    def check_fields(self) -> Self:
+        if self.model and (self.exclude and self.fields):
+            raise ValueError("Specify either `exclude` or `fields`")
+
+        if self.fields_optional:
+            if self.optional_fields is not None:
+                raise ValueError(
+                    "Use only `optional_fields`, `fields_optional` is deprecated."
+                )
             warnings.warn(
-                "The use of `Config` class is deprecated for ModelSchema, use 'Meta' instead",
+                "The use of `fields_optional` is deprecated. Use `optional_fields` instead to match `create_schema()` API",
                 DeprecationWarning,
                 stacklevel=2,
             )
-
-        else:
-            raise ConfigError(
-                f"ModelSchema class '{name}' requires a 'Meta' (or a 'Config') subclass"
-            )
-
-        assert issubclass(model, DjangoModel)
-
-        if not fields and not exclude:
-            raise ConfigError(
-                "Creating a ModelSchema without either the 'fields' attribute"
-                " or the 'exclude' attribute is prohibited"
-            )
-
-        if fields == "__all__":
-            fields = None
-            # ^ when None is passed to create_schema - all fields are selected
-
-        return MetaConf(
-            model=model,
-            fields=fields,
-            exclude=exclude,
-            fields_optional=optional_fields,
-        )
+            self.optional_fields = self.fields_optional
+        return self
 
 
 class ModelSchemaMetaclass(ResolverMetaclass):
@@ -74,52 +69,55 @@ def __new__(
         namespace: dict,
         **kwargs,
     ):
+        conf_class = None
+        meta_conf = None
+
+        if "Meta" in namespace:
+            conf_class = namespace["Meta"]
+        elif "Config" in namespace:
+            conf_class = namespace["Config"]
+            warnings.warn(
+                "The use of `Config` class is deprecated for ModelSchema, use 'Meta' instead",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        if conf_class:
+            conf_dict = {
+                k: v for k, v in getmembers(conf_class) if not k.startswith("__")
+            }
+            meta_conf = MetaConf.model_validate(conf_dict)
+
+        if meta_conf and meta_conf.model:
+            base_model_fields = {}
+            for base in bases:
+                base_model_fields.update(**base.model_fields)
+
+            meta_conf = meta_conf.model_dump(exclude_none=True)
+
+            fields = factory.convert_django_fields(**meta_conf)
+            namespace.setdefault("__annotations__", {})
+            for field, val in fields.items():
+                # don't overwrite custom fields in annotations or previous django fields in base_model_fields
+                if not namespace["__annotations__"].get(field, None) and not base_model_fields.get(field, None):
+                    # set type
+                    namespace["__annotations__"][field] = val[0]
+                    # and default value
+                    namespace[field] = val[1]
+
         cls = super().__new__(
             mcs,
             name,
             bases,
             namespace,
             **kwargs,
         )
-        for base in reversed(bases):
-            if (
-                _is_modelschema_class_defined
-                and issubclass(base, ModelSchema)
-                and base == ModelSchema
-            ):
-                meta_conf = MetaConf.from_schema_class(name, namespace)
-
-                custom_fields = []
-                annotations = namespace.get("__annotations__", {})
-                for attr_name, type in annotations.items():
-                    if attr_name.startswith("_"):
-                        continue
-                    default = namespace.get(attr_name, ...)
-                    custom_fields.append((attr_name, type, default))
-
-                # # cls.__doc__ = namespace.get("__doc__", config.model.__doc__)
-                # cls.__fields__ = {}  # forcing pydantic recreate
-                # # assert False, "!! cls.model_fields"
-
-                # print(config.model, name, fields, exclude, "!!")
-
-                model_schema = create_schema(
-                    meta_conf.model,
-                    name=name,
-                    fields=meta_conf.fields,
-                    exclude=meta_conf.exclude,
-                    optional_fields=meta_conf.fields_optional,
-                    custom_fields=custom_fields,
-                    base_class=cls,
-                )
-                model_schema.__doc__ = cls.__doc__
-                return model_schema
-
         return cls
 
 
 class ModelSchema(Schema, metaclass=ModelSchemaMetaclass):
-    pass
-
-
-_is_modelschema_class_defined = True
+    @no_type_check
+    def __new__(cls, *args, **kwargs):
+        if not getattr(getattr(cls, "Meta", {}), "model", None):
+            raise ConfigError(f"No model set for class '{cls.__name__}'")
+        return super().__new__(cls)