Feat: Custom Pages Routes (#1185)

* refactor: Add CustomPage and related models for custom pages functionality

* refactor: Update CustomPage model to use "content" instead of "elements"

* refactor: Update import statement in models.py

* refactor: Update CustomPage model to use "is_published" instead of "status"

* refactor: Update default value for CommunityCustomPage sharing_type

* chore: Add models and migrations for custom pages functionality

* refactor: Update CustomPage model to use "content" instead of "elements"

* chore: generated new migration fle

* refactor: Implement create_unique_slug function for generating unique slugs

* refactor: Update create_unique_slug function to include timestamp for uniqueness

* test: Add unit tests for create_unique_slug function

* feat: Implement CustomPagesService and CustomPagesHandler for community custom page management

* refactor: regenerated migration files for the new model changes

* feat: Enhance parse_list function to handle lists of dictionaries

* feat: Add content field to CustomPage model and update versioning format

* refactor: Simplify community custom page methods and remove unused block methods

* feat: Add CustomPagesHandler to API routes and simplify slug creation logic

* refactor: Simplify slug creation logic and remove unused migration file

* feat: Add integration tests for creating community custom pages

* feat: Restrict community custom page actions to admins and improve error handling

* feat: Add helper function to create community custom pages in tests

* refactor: Optimize parse_list function for better performance and error logging

src/_main_/utils/common.py

@@ -89,20 +89,20 @@ def get_request_contents(request, **kwargs):
 
 def parse_list(d):
     try:
-        tmp = []
         if isinstance(d, str):
             return d.strip().split(",") if d else []
-        elif isinstance(d, dict):
-            tmp = list(d.values())
 
-        res = []
-        for i in tmp:
-            if i.isnumeric():
-                res.append(i)
-        return res
+        if isinstance(d, dict):
+            return [value for value in d.values() if str(value).isnumeric()]
 
+        if isinstance(d, list):
+            if d and isinstance(d[0], dict):
+                return d
+            return [item for item in d if str(item).isnumeric()]
+
+        return []
     except Exception as e:
-        log.exception(e)
+        log.exception("Error in parse_list: %s", e)
         return []
 
 def parse_dict(d: object) -> object:

src/api/handlers/custom_pages.py

@@ -0,0 +1,143 @@
+from _main_.utils.context import Context
+from _main_.utils.massenergize_response import MassenergizeResponse
+from _main_.utils.route_handler import RouteHandler
+from api.decorators import admins_only
+from api.services.custom_pages import CustomPagesService
+
+
+class CustomPagesHandler(RouteHandler):
+
+  def __init__(self):
+    super().__init__()
+    self.service = CustomPagesService()
+    self.registerRoutes()
+
+  def registerRoutes(self):
+    self.add("/community.custom.pages.create", self.create_community_custom_page) 
+    self.add("/community.custom.pages.update", self.update_community_custom_page)
+    self.add("/community.custom.pages.delete", self.delete_community_custom_page)
+    self.add("/community.custom.pages.share", self.share_community_custom_page)
+    self.add("/community.custom.pages.info", self.community_custom_page_info)
+    self.add("/community.custom.pages.list", self.list_community_custom_pages)
+    self.add("/custom.page.publish", self.publish_custom_page)
+
+  @admins_only
+  def create_community_custom_page(self, request): 
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("title", str, is_required=True)
+    self.validator.expect("community_id", str, is_required=True)
+    self.validator.expect("content", list, is_required=False)
+    self.validator.expect("audience", "str_list", is_required=False)
+    self.validator.expect("sharing_type", str, is_required=False)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.create_community_custom_page(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+  @admins_only
+  def update_community_custom_page(self, request): 
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("id", str, is_required=True)
+    self.validator.expect("title", str, is_required=False)
+    self.validator.expect("content", list, is_required=False)
+    self.validator.expect("audience", "str_list", is_required=False)
+    self.validator.expect("sharing_type", str, is_required=False)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.update_community_custom_page(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+  @admins_only
+  def delete_community_custom_page(self, request):
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("id", str, is_required=True)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.delete_community_custom_page(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+  @admins_only
+  def share_community_custom_page(self, request): 
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("community_page_id", str, is_required=True)
+    self.validator.expect("community_ids", "str_list", is_required=True)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.share_community_custom_page(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+
+  def community_custom_page_info(self, request): 
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("id", str, is_required=True)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.community_custom_page_info(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+  @admins_only
+  def list_community_custom_pages(self, request): 
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("community_id", "id", is_required=True)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.list_community_custom_pages(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)
+
+  @admins_only
+  def publish_custom_page(self, request):
+    context: Context = request.context
+    args: dict = context.args
+
+    self.validator.expect("id", str, is_required=True)
+
+    args, err = self.validator.verify(args, strict=True)
+    if err:
+      return err
+
+    page, err = self.service.publish_custom_page(context, args)
+    if err:
+      return err
+    return MassenergizeResponse(data=page)

src/api/services/custom_pages.py

@@ -0,0 +1,89 @@
+from typing import Tuple
+from _main_.utils.common import serialize, serialize_all
+from _main_.utils.context import Context
+from _main_.utils.massenergize_errors import MassEnergizeAPIError
+from api.store.custom_pages import CustomPagesStore
+
+
+class CustomPagesService:
+  """
+  Service Layer for all the custom pages
+  """
+
+
+  def __init__(self):
+    self.store =  CustomPagesStore()
+
+  def create_community_custom_page(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+      page, err = self.store.create_community_custom_page(context, args)
+      if err:
+        return None, err
+
+      return serialize(page), None
+    except Exception as e:
+      return None, MassEnergizeAPIError(str(e))
+
+  def update_community_custom_page(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]: 
+    try:
+      page, err = self.store.update_community_custom_page(context, args)
+      if err:
+        return None, err
+      return serialize(page), None
+    except Exception as e:
+      return None, MassEnergizeAPIError(str(e))
+
+  def delete_community_custom_page(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+
+      page, err = self.store.delete_community_custom_page(context, args)
+
+      if err:
+        return None, err
+      return serialize(page), None
+    except Exception as e:
+      return None, MassEnergizeAPIError(str(e))
+
+  def list_community_custom_pages(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+      pages, err = self.store.list_community_custom_pages(context, args)
+      if err:
+        return None, err
+      return serialize_all(pages), None
+    except Exception as e:
+      return None, MassEnergizeAPIError(str(e))
+
+
+  def community_custom_page_info(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+
+        page, err = self.store.community_custom_page_info(context, args)
+        if err:
+          return None, err
+        return serialize(page), None
+    except Exception as e:
+        return None, MassEnergizeAPIError(str(e))
+
+
+  def share_community_custom_page(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+        page, err = self.store.share_community_custom_page(context, args)
+        if err:
+          return None, err
+        return serialize(page), None
+    except Exception as e:
+        return None, MassEnergizeAPIError(str(e))
+
+
+  def publish_custom_page(self, context: Context, args) -> Tuple[dict, MassEnergizeAPIError]:
+    try:
+        page, err = self.store.publish_custom_page(context, args)
+        if err:
+          return None, err
+        return serialize(page), None
+    except Exception as e:
+        return None, MassEnergizeAPIError(str(e))
+
+
+
+