Implement primitive model and API handlers for Category

This adds new feature that gets together models that has same attribute
(feature, purpose and so on).

Author: userlocalhost

CHANGELOG.md

@@ -6,6 +6,10 @@
 * Enable to get items considering with Alias name (by simple solution)
   Contributed by @userlocalhost
 
+* Added new feature that gets together models that has same attribute
+  (feature, purpose and so on).
+  Contributed by @userlocalhost
+
 ### Changed
 
 ### Fixed

airone/lib/test.py

@@ -2,13 +2,15 @@
 import inspect
 import os
 import sys
+from typing import List
 
 from django.conf import settings
 from django.test import Client, TestCase, override_settings
 from pytz import timezone
 
 from airone.lib.acl import ACLType
 from airone.lib.types import AttrType
+from category.models import Category
 from entity.models import Entity, EntityAttr
 from entry.models import Entry
 from user.models import User
@@ -148,6 +150,17 @@ def add_entry(self, user: User, name: str, schema: Entity, values={}, is_public=
 
         return entry
 
+    def create_category(self, user: User, name: str, note: str = "", models: List[Entity] = []):
+        # create target Category instance
+        category = Category.objects.create(name=name, note=note, created_user=user)
+
+        # attach created category to each specified Models
+        for model in models:
+            if model.is_active:
+                model.categories.add(category)
+
+        return category
+
     def _do_login(self, uname, is_superuser=False) -> User:
         # create test user to authenticate
         user = User(username=uname, is_superuser=is_superuser)

airone/settings_common.py

@@ -64,6 +64,7 @@ class Common(Configuration):
         "simple_history",
         "storages",
         "trigger",
+        "category",
     ]
 
     if os.path.exists(BASE_DIR + "/custom_view"):

airone/urls.py

@@ -42,6 +42,7 @@
     re_path(r"^webhook/", include(("webhook.urls", "webhook"))),
     re_path(r"^role/", include(("role.urls", "role"))),
     re_path(r"^trigger/", include(("trigger.urls", "trigger"))),
+    re_path(r"^category/", include(("category.urls", "category"))),
 ]
 
 if settings.DEBUG:

category/__init__.py

@@ -0,0 +1 @@
+# Register your models here.

category/admin.py

@@ -0,0 +1,62 @@
+from rest_framework import serializers
+
+from category.models import Category
+from entity.api_v2.serializers import EntitySerializer
+from entity.models import Entity
+
+
+class CategoryListSerializer(serializers.ModelSerializer):
+    models = EntitySerializer(many=True)
+
+    class Meta:
+        model = Category
+        fields = ["id", "name", "note", "models"]
+
+
+class CategoryCreateSerializer(serializers.ModelSerializer):
+    id = serializers.IntegerField(required=False, read_only=True)
+    models = serializers.ListField(write_only=True, required=False, default=[])
+
+    class Meta:
+        model = Category
+        fields = ["id", "name", "note", "models"]
+
+    def create(self, validated_data):
+        # craete Category instance
+        category = Category.objects.create(
+            created_user=self.context["request"].user,
+            **{k: v for (k, v) in validated_data.items() if k != "models"},
+        )
+
+        # make relations created Category with specified Models
+        for model in Entity.objects.filter(id__in=validated_data.get("models", []), is_active=True):
+            model.categories.add(category)
+
+        return category
+
+
+class CategoryUpdateSerializer(serializers.ModelSerializer):
+    models = serializers.ListField(write_only=True, required=False, default=[])
+
+    class Meta:
+        model = Category
+        fields = ["name", "note", "models"]
+
+    def update(self, category: Category, validated_data):
+        curr_model_ids = [x.id for x in category.models.filter(is_active=True)]
+
+        # remove models that are unselected from current ones
+        removing_model_ids = list(
+            set(curr_model_ids) - set(validated_data.get("models", curr_model_ids))
+        )
+        for model in Entity.objects.filter(id__in=removing_model_ids, is_active=True):
+            model.categories.remove(category)
+
+        # add new models that are added to current ones
+        adding_model_ids = list(
+            set(validated_data.get("models", curr_model_ids)) - set(curr_model_ids)
+        )
+        for model in Entity.objects.filter(id__in=adding_model_ids, is_active=True):
+            model.categories.add(category)
+
+        return category

category/api_v2/__init__.py

@@ -0,0 +1,25 @@
+from django.urls import path
+
+from . import views
+
+urlpatterns = [
+    path(
+        "",
+        views.CategoryAPI.as_view(
+            {
+                "get": "list",
+                "post": "create",
+            }
+        ),
+    ),
+    path(
+        "<int:pk>/",
+        views.CategoryAPI.as_view(
+            {
+                "get": "retrieve",
+                "put": "update",
+                "delete": "destroy",
+            }
+        ),
+    ),
+]

category/api_v2/serializers.py

@@ -0,0 +1,43 @@
+from django_filters.rest_framework import DjangoFilterBackend
+from rest_framework import filters, status, viewsets
+from rest_framework.pagination import LimitOffsetPagination
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from airone.lib.drf import ObjectNotExistsError
+from category.api_v2.serializers import (
+    CategoryCreateSerializer,
+    CategoryListSerializer,
+    CategoryUpdateSerializer,
+)
+from category.models import Category
+from entity.api_v2.views import EntityPermission
+
+
+class CategoryAPI(viewsets.ModelViewSet):
+    pagination_class = LimitOffsetPagination
+    permission_classes = [IsAuthenticated & EntityPermission]
+    filter_backends = [DjangoFilterBackend, filters.OrderingFilter, filters.SearchFilter]
+    search_fields = ["name"]
+    ordering = ["name"]
+
+    def get_serializer_class(self):
+        serializer = {
+            "create": CategoryCreateSerializer,
+            "update": CategoryUpdateSerializer,
+        }
+        return serializer.get(self.action, CategoryListSerializer)
+
+    def get_queryset(self):
+        return Category.objects.filter(is_active=True)
+
+    def destroy(self, request: Request, *args, **kwargs) -> Response:
+        category: Category = self.get_object()
+        if not category.is_active:
+            raise ObjectNotExistsError("specified entry has already been deleted")
+
+        # delete specified category
+        category.delete()
+
+        return Response(status=status.HTTP_204_NO_CONTENT)

category/api_v2/urls.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class CategoryConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "category"

category/api_v2/views.py

@@ -0,0 +1,7 @@
+from django.db import models
+
+from acl.models import ACLBase
+
+
+class Category(ACLBase):
+    note = models.CharField(max_length=500, blank=True, default="")

category/apps.py

@@ -0,0 +1,122 @@
+import json
+from typing import List
+
+from airone.lib.test import AironeViewTest
+from category.models import Category
+from entity.models import Entity
+from user.models import User
+
+
+class ViewTest(AironeViewTest):
+    def setUp(self):
+        super(ViewTest, self).setUp()
+
+        self.user: User = self.guest_login()
+
+    def test_list(self):
+        # initialize Model and Categories for testing API processing
+        model: Entity = self.create_entity(self.user, "Model")
+        categories: List[Category] = [
+            self.create_category(self.user, "Category-%d" % n, "Note-%d" % n, [model])
+            for n in range(3)
+        ]
+
+        # get all Categories
+        resp = self.client.get("/category/api/v2/")
+        self.assertEqual(resp.status_code, 200)
+
+        # check returned data has expected values
+        self.assertEqual(resp.json()["count"], 3)
+        self.assertEqual(
+            resp.json()["results"],
+            [
+                {
+                    "id": x.id,
+                    "name": x.name,
+                    "note": x.note,
+                    "models": [{"id": model.id, "name": model.name, "is_public": True}],
+                }
+                for x in categories
+            ],
+        )
+
+    def test_list_include_deleted_category(self):
+        pass
+
+    def test_get(self):
+        # initialize Models and Categories for testing API processing
+        models: List[Entity] = [self.create_entity(self.user, "Model-%d" % n) for n in range(3)]
+        category: Category = self.create_category(self.user, "Category", "note", models)
+
+        # get specified Category
+        resp = self.client.get("/category/api/v2/%d/" % category.id)
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(
+            resp.json(),
+            {
+                "id": category.id,
+                "name": category.name,
+                "note": category.note,
+                "models": [{"id": x.id, "name": x.name, "is_public": True} for x in models],
+            },
+        )
+
+    def test_get_include_deleted_model(self):
+        pass
+
+    def test_create(self):
+        # initialize Models for testing API processing
+        models: List[Entity] = [self.create_entity(self.user, "Model-%d" % n) for n in range(3)]
+
+        # send request to create a Category
+        params = {
+            "name": "New Category",
+            "note": "Hoge",
+            "models": [x.id for x in models],
+        }
+        resp = self.client.post("/category/api/v2/", json.dumps(params), "application/json")
+        self.assertEqual(resp.status_code, 201)
+
+        # check Category item is created and has expected attribute values
+        category = Category.objects.last()
+        self.assertEqual(category.id, resp.json()["id"])
+        self.assertEqual(category.name, resp.json()["name"])
+        self.assertEqual(category.note, resp.json()["note"])
+        self.assertEqual(category.models.count(), 3)
+        self.assertEqual(list(category.models.all()), models)
+
+    def test_delete(self):
+        # create Category instance that will be deleted in this test
+        category: Category = self.create_category(self.user, "Category", "note")
+
+        # send request to delete target Category
+        resp = self.client.delete("/category/api/v2/%d/" % category.id)
+        self.assertEqual(resp.status_code, 204)
+
+        # check target category instance is existed but inactivated
+        self.assertTrue(Category.objects.filter(id=category.id).exists())
+        self.assertFalse(Category.objects.filter(id=category.id, is_active=True).exists())
+
+    def test_update(self):
+        """
+        Initially, there are 3 models (M0, M1, M2) and Category has M0 and M1
+        Then, this test sends a request to change its beloning category to M1 and M2.
+        """
+        models: List[Entity] = [self.create_entity(self.user, "M%d" % n) for n in range(3)]
+        category: Category = self.create_category(
+            self.user, "Category", "note", [models[0], models[1]]
+        )
+        self.assertEqual(list(category.models.all()), [models[0], models[1]])
+
+        params = {
+            "name": "Updated Category",
+            "note": "Updated Note",
+            "models": [models[1].id, models[2].id],
+        }
+        resp = self.client.put(
+            "/category/api/v2/%s/" % category.id, json.dumps(params), "application/json"
+        )
+        self.assertEqual(resp.status_code, 200)
+
+        # check all data is updated expectedly
+        self.assertEqual(list(category.models.all()), [models[1], models[2]])

category/migrations/__init__.py

@@ -0,0 +1,5 @@
+from django.urls import include, re_path
+
+urlpatterns = [
+    re_path(r"^api/v2/", include(("category.api_v2.urls", "category.api_v2"))),
+]

category/models.py

@@ -0,0 +1 @@
+# Create your views here.

category/tests/__init__.py

@@ -6,6 +6,7 @@
 
 from acl.models import ACLBase
 from airone.lib.acl import ACLObjType
+from category.models import Category
 from webhook.models import Webhook
 
 
@@ -100,6 +101,9 @@ class Entity(ACLBase):
 
     history = HistoricalRecords(excluded_fields=["status", "updated_time"])
 
+    # The Category that groups Models according to their purpose (which is defined by User)
+    categories = models.ManyToManyField(Category, default=[], related_name="models")
+
     def __init__(self, *args, **kwargs):
         super(Entity, self).__init__(*args, **kwargs)
         self.objtype = ACLObjType.Entity

category/tests/test_api_v2.py

@@ -4,6 +4,7 @@
 from django.test import TestCase
 
 from airone.lib.types import AttrType
+from category.models import Category
 from entity.admin import EntityAttrResource, EntityResource
 from entity.models import Entity, EntityAttr
 from user.models import User
@@ -401,3 +402,27 @@ def test_max_attributes_per_entity(self):
             created_user=self._test_user,
             parent_entity=entity,
         )
+
+    def test_category(self):
+        models = [
+            Entity.objects.create(name="Model-%d" % n, created_user=self._test_user)
+            for n in range(3)
+        ]
+        categories = [
+            Category.objects.create(name="Category-%d" % n, created_user=self._test_user)
+            for n in range(3)
+        ]
+
+        # set relation from single Model to multiple Category
+        for n in range(3):
+            models[0].categories.add(categories[n])
+
+        # check each categories are set properly
+        self.assertEqual(list(models[0].categories.all()), categories)
+
+        # set relation from single Category to multiple Model
+        for n in range(3):
+            models[n].categories.add(categories[0])
+
+        # check each models are set properly
+        self.assertEqual(list(categories[0].models.all()), models)