Add complete CRUD procedure for Blog Posts and Category.

Author: drmacsika

blog/api_crud.py

@@ -1,4 +1,5 @@
 from typing import Generic, List, Optional, Type, TypeVar
+from unicodedata import category
 
 from core.base_crud import SLUGTYPE, BaseCRUD
 from core.utils import unique_slug_generator
@@ -16,114 +17,110 @@ class PostCRUD(BaseCRUD[Post, CreatePost, UpdatePost, SLUGTYPE]):
     CRUD Operation for blog posts
     """
 
-#     >>> voted_choices = Choice.objects.filter(votes__gt=0)
-# >>> voted_choices
-# <QuerySet [<Choice: The sky>]>
-# >>> prefetch = Prefetch('choice_set', queryset=voted_choices)
-# >>> Question.objects.prefetch_related(prefetch).get().choice_set.all()
-# <QuerySet [<Choice: The sky>]>
-
-
-# prefetch = Prefetch('choice_set', queryset=voted_choices, to_attr='voted_choices')
-# >>> Question.objects.prefetch_related(prefetch).get().voted_choices
-# [<Choice: The sky>]
-# >>> Question.objects.prefetch_related(prefetch).get().choice_set.all()
-    
-    
-    
     def get(self, slug: SLUGTYPE) -> Optional[Post]:
         """
         Get single blog post.
         """
-        user_query = Post.objects.select_related("user").filter(slug=slug)
-        prefetch = Prefetch("category", queryset=user_query)
-        query = Post.objects.prefetch_related(prefetch).all()
-        return query
+        try:
+            query = Post.objects.select_related("user", "category").get(slug=slug)
+            return query
+        except ObjectDoesNotExist:
+            raise HTTPException(status_code=404, detail="This post does not exists.")
 
+    def get_multiple(self, limit:int = 100, offset: int = 0) -> List[Post]:
+        """
+        Get multiple posts using a query limit and offset flag.
+        """
+        query = Post.objects.select_related("user", "category").all()[offset:offset+limit]
+        if not query:
+            raise HTTPException(status_code=404, detail="There are no posts.")
+        return list(query)
 
-    def get_multiple(self, limit=100, offset=0) -> List[Post]:
+    def get_posts_by_category(self, slug: SLUGTYPE) -> List[Post]:
         """
-        get multiple items using a query limiting flag.
+        Get all posts belonging to a particular category.
         """
-        user_query = Post.objects.select_related("user")
-        prefetch = Prefetch("category", queryset=user_query)
-        query = Post.objects.prefetch_related(prefetch).all()
+        query_category = Category.objects.filter(slug=slug)
+        if not query_category:
+            raise HTTPException(status_code=404, detail="This category does not exist.")
+        query = Post.objects.filter(category__slug=slug).select_related("user").all()
         return list(query)
 
     def create(self, obj_in: CreatePost) -> Post:
         """
-        Create an item.
+        Create an post.
         """
         slug = unique_slug_generator(obj_in.title)
-        post = self.get(slug=slug)
-        if post:
+        post = Post.objects.filter(slug=slug)
+        if not post:
             slug = unique_slug_generator(obj_in.title, new_slug=True)
         obj_in = jsonable_encoder(obj_in)
         query = Post.objects.create(**obj_in)
         return query
 
-    # def update(self, obj_in: UpdateSchema, slug: SLUGTYPE) -> ModelType:
-    #     """
-    #     Update an item.
-    #     """
-    #     if not isinstance(obj_in, list):
-    #         obj_in = jsonable_encoder(obj_in)
-    #     return self.model.objects.filter(slug=slug).update(**obj_in)
+    def update(self, obj_in: UpdatePost, slug: SLUGTYPE) -> Post:
+        """
+        Update an item.
+        """
+        self.get(slug=slug)
+        if not isinstance(obj_in, list):
+            obj_in = jsonable_encoder(obj_in)
+        return Post.objects.filter(slug=slug).update(**obj_in)
 
-    # def delete(self, slug: SLUGTYPE) -> ModelType:
-    #     """Delete an item."""
-    #     self.model.objects.filter(slug=slug).delete()
-    #     return {"detail": "Successfully deleted!"}
-
-
+    def delete(self, slug: SLUGTYPE) -> Post:
+        """Delete an item."""
+        self.model.objects.filter(slug=slug).delete()
+        return {"detail": "Successfully deleted!"}
 
 
 class CategoryCRUD(BaseCRUD[Category, CreateCategory, UpdateCategory, SLUGTYPE]):
     """
     CRUD Operation for blog categories.
     """
-
-
+    
     def get(self, slug: SLUGTYPE) -> Optional[Category]:
         """
-        Get single blogcCategory.
-        """
-        query = Category.objects.filter(slug=slug).all()
-        return query
+        Get a single category.
+        """    
+        try:
+            query = Category.objects.get(slug=slug)
+            return query
+        except ObjectDoesNotExist:
+            raise HTTPException(status_code=404, detail="This post does not exists.")
 
-    
-    def get_multiple(self, limit=100, offset=0) -> List[Category]:
+    def get_multiple(self, limit:int = 100, offset: int = 0) -> List[Category]:
         """
-        get multiple items using a query limiting flag.
+        Get multiple categories using a query limiting flag.
         """
-        posts = super().get_multiple(limit=limit, offset=offset)
-        return list(posts)
-    
-    
+        query = Category.objects.all()[offset:offset+limit]
+        if not query:
+            raise HTTPException(status_code=404, detail="There are no posts.")
+        return list(query)
+
     def create(self, obj_in: CreateCategory) -> Category:
         """
-        Create an item.
+        Create a category.
         """
         slug = unique_slug_generator(obj_in.title)
-        category = self.get(slug=slug)
+        category = Category.objects.filter(slug=slug)
         if category:
             raise HTTPException(status_code=404, detail="Category exists already.")
         obj_in = jsonable_encoder(obj_in)
         query = Category.objects.create(**obj_in)
         return query
 
-    # def update(self, obj_in: UpdateSchema, slug: SLUGTYPE) -> ModelType:
-    #     """
-    #     Update an item.
-    #     """
-    #     if not isinstance(obj_in, list):
-    #         obj_in = jsonable_encoder(obj_in)
-    #     return self.model.objects.filter(slug=slug).update(**obj_in)
+    def update(self, obj_in: UpdateCategory, slug: SLUGTYPE) -> Category:
+        """
+        Update a category.
+        """
+        if not isinstance(obj_in, list):
+            obj_in = jsonable_encoder(obj_in)
+        return self.model.objects.filter(slug=slug).update(**obj_in)
 
-    # def delete(self, slug: SLUGTYPE) -> ModelType:
-    #     """Delete an item."""
-    #     self.model.objects.filter(slug=slug).delete()
-    #     return {"detail": "Successfully deleted!"}
+    def delete(self, slug: SLUGTYPE) -> Post:
+        """Delete a category."""
+        Post.objects.filter(slug=slug).delete()
+        return {"detail": "Successfully deleted!"}
 
 
 post = PostCRUD(Post)