lint

Author: freemindcore

easy/controller/meta.py

@@ -13,7 +13,7 @@
 from ninja_extra.pagination import paginate
 
 from easy.controller.meta_conf import MODEL_FIELDS_ATTR_DEFAULT, ModelOptions
-from easy.domain.orm import CrudModel
+from easy.domain.meta import CrudModel
 from easy.response import BaseApiResponse
 from easy.services import BaseService
 from easy.utils import copy_func

easy/domain/__init__.py

@@ -1,3 +1,7 @@
 from .base import BaseDomain
+from .meta import CrudModel
 
-__all__ = ["BaseDomain"]
+__all__ = [
+    "BaseDomain",
+    "CrudModel",
+]

easy/domain/meta.py

@@ -0,0 +1,37 @@
+from abc import abstractmethod
+from typing import Any, Dict, Optional
+
+
+class CrudModel(object):
+    Meta: Dict = {}
+
+    def __init__(self, model: Any):
+        self.model = model
+
+    @abstractmethod
+    def crud_add_obj(self, **payload: Dict) -> Any:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_del_obj(self, pk: int) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_update_obj(self, pk: int, payload: Dict) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_get_obj(self, pk: int) -> Any:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_get_objs_all(self, maximum: Optional[int] = None, **filters: Any) -> Any:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_filter(self, **kwargs: Any) -> Any:
+        raise NotImplementedError
+
+    @abstractmethod
+    def crud_filter_exclude(self, **kwargs: Any) -> Any:
+        raise NotImplementedError

easy/domain/orm.py

@@ -1,52 +1,16 @@
 import logging
-from abc import abstractmethod
 from typing import Any, Dict, List, Optional, Tuple, Type
 
 from django.db import models, transaction
-from django.db.models.query import QuerySet
 from ninja_extra.shortcuts import get_object_or_none
 
 from easy.controller.meta_conf import ModelMetaConfig
+from easy.domain.meta import CrudModel
 from easy.exception import BaseAPIException
 
 logger = logging.getLogger(__name__)
 
 
-class CrudModel(object):
-    Meta: Dict = {}
-
-    def __init__(self, model: Any):
-        self.model = model
-
-    @abstractmethod
-    def crud_add_obj(self, **payload: Dict) -> Any:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_del_obj(self, pk: int) -> bool:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_update_obj(self, pk: int, payload: Dict) -> bool:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_get_obj(self, pk: int) -> Any:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_get_objs_all(self, maximum: Optional[int] = None, **filters: Any) -> Any:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_filter(self, **kwargs: Any) -> QuerySet:
-        raise NotImplementedError
-
-    @abstractmethod
-    def crud_filter_exclude(self, **kwargs: Any) -> QuerySet:
-        raise NotImplementedError
-
-
 class DjangoOrmModel(CrudModel):
     def __init__(self, model: Type[models.Model]):
         self.model = model
@@ -164,3 +128,128 @@ def crud_filter(self, **kwargs: Any) -> Any:
 
     def crud_filter_exclude(self, **kwargs: Any) -> Any:
         return self.model.objects.all().exclude(**kwargs)
+
+
+class DjangoSerializer(ModelMetaConfig):
+    @staticmethod
+    def is_model_instance(data: Any) -> bool:
+        return isinstance(data, models.Model)
+
+    @staticmethod
+    def is_queryset(data: Any) -> bool:
+        return isinstance(data, models.query.QuerySet)
+
+    @staticmethod
+    def is_one_relationship(data: Any) -> bool:
+        return isinstance(data, models.ForeignKey) or isinstance(
+            data, models.OneToOneRel
+        )
+
+    @staticmethod
+    def is_many_relationship(data: Any) -> bool:
+        return (
+            isinstance(data, models.ManyToManyRel)
+            or isinstance(data, models.ManyToManyField)
+            or isinstance(data, models.ManyToOneRel)
+        )
+
+    @staticmethod
+    def is_paginated(data: Any) -> bool:
+        return isinstance(data, dict) and isinstance(
+            data.get("items", None), models.query.QuerySet
+        )
+
+    def serialize_model_instance(
+        self, obj: models.Model, referrers: Any = tuple()
+    ) -> Dict[Any, Any]:
+        """Serializes Django model instance to dictionary"""
+        out = {}
+        for field in obj._meta.get_fields():
+            if self.show_field(obj, field.name):
+                if self.is_one_relationship(field):
+                    out.update(
+                        self.serialize_foreign_key(obj, field, referrers + (obj,))
+                    )
+
+                elif self.is_many_relationship(field):
+                    out.update(
+                        self.serialize_many_relationship(obj, referrers + (obj,))
+                    )
+
+                else:
+                    out.update(self.serialize_value_field(obj, field))
+        return out
+
+    def serialize_queryset(
+        self, data: models.query.QuerySet, referrers: Tuple[Any, ...] = tuple()
+    ) -> List[Dict[Any, Any]]:
+        """Serializes Django Queryset to dictionary"""
+        return [self.serialize_model_instance(obj, referrers) for obj in data]
+
+    def serialize_foreign_key(
+        self, obj: models.Model, field: Any, referrers: Any = tuple()
+    ) -> Dict[Any, Any]:
+        """Serializes foreign key field of Django model instance"""
+        try:
+            if not hasattr(obj, field.name):
+                return {field.name: None}  # pragma: no cover
+            related_instance = getattr(obj, field.name)
+            if related_instance is None:
+                return {field.name: None}
+            if related_instance in referrers:
+                return {}  # pragma: no cover
+            field_value = getattr(related_instance, "pk")
+        except Exception as exc:  # pragma: no cover
+            logger.error(f"serialize_foreign_key error - {obj}", exc_info=exc)
+            return {field.name: None}
+
+        if self.get_model_recursive(obj):
+            return {
+                field.name: self.serialize_model_instance(related_instance, referrers)
+            }
+        return {field.name: field_value}
+
+    def serialize_many_relationship(
+        self, obj: models.Model, referrers: Any = tuple()
+    ) -> Dict[Any, Any]:
+        """
+        Serializes many relationship (ManyToMany, ManyToOne) of Django model instance
+        """
+        if not hasattr(obj, "_prefetched_objects_cache"):
+            return {}
+        out = {}
+        try:
+            for k, v in obj._prefetched_objects_cache.items():  # type: ignore
+                field_name = k if hasattr(obj, k) else k + "_set"
+                if v:
+                    if self.get_model_join(obj):
+                        out[field_name] = self.serialize_queryset(v, referrers + (obj,))
+                    else:
+                        out[field_name] = [o.pk for o in v]
+                else:
+                    out[field_name] = []
+        except Exception as exc:  # pragma: no cover
+            logger.error(f"serialize_many_relationship error - {obj}", exc_info=exc)
+        return out
+
+    def serialize_value_field(self, obj: models.Model, field: Any) -> Dict[Any, Any]:
+        """
+        Serializes regular 'jsonable' field (Char, Int, etc.) of Django model instance
+        """
+        return {field.name: getattr(obj, field.name)}
+
+    def serialize_data(self, data: Any) -> Any:
+        out = data
+        # Queryset
+        if self.is_queryset(data):
+            out = self.serialize_queryset(data)
+        # Model
+        elif self.is_model_instance(data):
+            out = self.serialize_model_instance(data)
+        # Add limit_off pagination support
+        elif self.is_paginated(data):
+            out = self.serialize_queryset(data.get("items"))
+        return out
+
+
+django_serializer = DjangoSerializer()

easy/domain/serializers.py

@@ -5,14 +5,14 @@
 from django.conf import settings
 from django.http import HttpRequest, HttpResponse
 from django.utils.module_loading import module_has_submodule
+from domain.orm import django_serializer
 from ninja.constants import NOT_SET, NOT_SET_TYPE
 from ninja.parser import Parser
 from ninja.renderers import BaseRenderer
 from ninja.types import TCallable
 from ninja_extra import NinjaExtraAPI
 
 from easy.controller.auto_api import create_admin_controller
-from easy.domain.serializers import django_serializer
 from easy.renderer.json import EasyJSONRenderer
 from easy.response import BaseApiResponse