lint: django orm model

Author: freemindcore

easy/controller/admin_auto_api.py renamed to easy/controller/auto_api.py

@@ -2,7 +2,7 @@
 import logging
 import re
 import uuid
-from abc import ABCMeta
+from abc import ABC, ABCMeta
 from collections import ChainMap
 from typing import Any, Match, Optional, Tuple, Type
 
@@ -20,7 +20,7 @@
 logger = logging.getLogger(__name__)
 
 
-class CrudAPI(CrudModel):
+class CrudAPI(CrudModel, ABC):
     # Never add type note to service, it will cause injection error
     def __init__(self, service=None):  # type: ignore
         # Critical to set __Meta

easy/controller/meta.py

@@ -1,5 +1,5 @@
-from easy.domain.orm import BaseOrm
+from easy.domain.orm import DjangoOrmModel
 
 
-class BaseDomain(BaseOrm):
+class BaseDomain(DjangoOrmModel):
     pass

easy/domain/base.py

@@ -1,5 +1,6 @@
 import logging
-from typing import Any, Dict, List, Tuple, Type
+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
@@ -11,13 +12,47 @@
 
 
 class CrudModel(object):
+    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
         self.m2m_fields_list: List = list(
             _field
             for _field in self.model._meta.get_fields(include_hidden=True)
             if isinstance(_field, models.ManyToManyField)
         )
+        super().__init__(model)
 
     def _separate_payload(self, payload: Dict) -> Tuple[Dict, Dict]:
         m2m_fields = {}
@@ -45,7 +80,7 @@ def _crud_set_m2m_obj(obj: models.Model, m2m_fields: Dict) -> None:
 
     # Define BASE CRUD
     @transaction.atomic()
-    def _crud_add_obj(self, **payload: Dict) -> Any:
+    def crud_add_obj(self, **payload: Dict) -> Any:
         local_f_payload, m2m_f_payload = self._separate_payload(payload)
 
         try:
@@ -58,7 +93,7 @@ def _crud_add_obj(self, **payload: Dict) -> Any:
         if obj:
             return obj.id
 
-    def _crud_del_obj(self, pk: int) -> bool:
+    def crud_del_obj(self, pk: int) -> bool:
         obj = get_object_or_none(self.model, pk=pk)
         if obj:
             self.model.objects.filter(pk=pk).delete()
@@ -67,7 +102,7 @@ def _crud_del_obj(self, pk: int) -> bool:
             return False
 
     @transaction.atomic()
-    def _crud_update_obj(self, pk: int, payload: Dict) -> bool:
+    def crud_update_obj(self, pk: int, payload: Dict) -> bool:
         local_fields, m2m_fields = self._separate_payload(payload)
         if not self.model.objects.filter(pk=pk).exists():
             return False
@@ -78,7 +113,7 @@ def _crud_update_obj(self, pk: int, payload: Dict) -> bool:
             raise BaseAPIException(f"Update Error - {e}")
         return bool(obj)
 
-    def _crud_get_obj(self, pk: int) -> Any:
+    def crud_get_obj(self, pk: int) -> Any:
         if self.m2m_fields_list:
             qs = self.model.objects.filter(pk=pk).prefetch_related(
                 self.m2m_fields_list[0].name
@@ -90,10 +125,11 @@ def _crud_get_obj(self, pk: int) -> Any:
         if qs:
             return qs.first()
 
-    def _crud_get_objs_all(self, **filters: Any) -> Any:
+    def crud_get_objs_all(self, maximum: Optional[int] = None, **filters: Any) -> Any:
         """
         CRUD: get multiple objects, with django orm filters support
         Args:
+            maximum: {int}
             filters: {"field_name__lte", 1}
         Returns: qs
 
@@ -104,6 +140,8 @@ def _crud_get_objs_all(self, **filters: Any) -> Any:
                 qs = self.model.objects.filter(**filters)
             except Exception as e:  # pragma: no cover
                 logger.error(e)
+        elif maximum:
+            qs = self.model.objects.all()[:maximum]
         else:
             qs = self.model.objects.all()
         # If there are 2m2_fields
@@ -113,12 +151,8 @@ def _crud_get_objs_all(self, **filters: Any) -> Any:
                 qs = qs.prefetch_related(f.name)
         return qs
 
-    def _crud_filter(self, **kwargs: Any) -> QuerySet:
+    def crud_filter(self, **kwargs: Any) -> Any:
         return self.model.objects.filter(**kwargs)  # pragma: no cover
 
-    def _crud_filter_exclude(self, **kwargs: Any) -> QuerySet:
+    def crud_filter_exclude(self, **kwargs: Any) -> Any:
         return self.model.objects.all().exclude(**kwargs)
-
-
-class BaseOrm(CrudModel):
-    ...

easy/domain/orm.py

@@ -11,7 +11,7 @@
 from ninja.types import TCallable
 from ninja_extra import NinjaExtraAPI
 
-from easy.controller.admin_auto_api import create_admin_controller
+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

easy/main.py

@@ -4,36 +4,36 @@
 from asgiref.sync import sync_to_async
 from django.db import models
 
-from easy.domain.orm import CrudModel
+from easy.domain.orm import DjangoOrmModel
 
 logger = logging.getLogger(__name__)
 
 
-class CrudService(CrudModel):
+class CrudService(DjangoOrmModel):
     def __init__(self, model: Type[models.Model]):
         super().__init__(model)
         self.model = model
 
     async def get_obj(self, id: int) -> Any:
-        return await sync_to_async(self._crud_get_obj)(id)
+        return await sync_to_async(self.crud_get_obj)(id)
 
     async def get_objs(self, **filters: Any) -> Any:
-        return await sync_to_async(self._crud_get_objs_all)(**filters)
+        return await sync_to_async(self.crud_get_objs_all)(**filters)
 
     async def patch_obj(self, id: int, payload: Any) -> Any:
-        return await sync_to_async(self._crud_update_obj)(id, payload)
+        return await sync_to_async(self.crud_update_obj)(id, payload)
 
     async def del_obj(self, id: int) -> Any:
-        return await sync_to_async(self._crud_del_obj)(id)
+        return await sync_to_async(self.crud_del_obj)(id)
 
     async def add_obj(self, **payload: Any) -> Any:
-        return await sync_to_async(self._crud_add_obj)(**payload)
+        return await sync_to_async(self.crud_add_obj)(**payload)
 
     async def filter_objs(self, **payload: Any) -> Any:
-        return await sync_to_async(self._crud_filter)(**payload)  # pragma: no cover
+        return await sync_to_async(self.crud_filter)(**payload)  # pragma: no cover
 
     async def filter_exclude_objs(self, **payload: Any) -> Any:
-        return await sync_to_async(self._crud_filter_exclude)(**payload)
+        return await sync_to_async(self.crud_filter_exclude)(**payload)
 
     # async def bulk_create_objs(self):
     #     ...

easy/services/crud.py

@@ -132,7 +132,7 @@ async def get_objs_list_with_filter_exclude(self, request):
         "/qs/",
     )
     async def list_events(self):
-        qs = await sync_to_async(self.model.objects.all)()
+        qs = await sync_to_async(self.service.crud_get_objs_all)(maximum=10)
         await sync_to_async(list)(qs)
         if qs:
             return qs