Tag Lists by Hierarchy

main.py

@@ -7,6 +7,7 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from scripts.db.psql.create_default_tables import create_default_psql_dependencies
 from scripts.services import route
+from scripts.services.tag_list_services import tag_service_router
 from scripts.utils.security_utils.decorators import CookieAuthentication
 
 auth = CookieAuthentication()
@@ -43,6 +44,7 @@ if os.environ.get("ENABLE_CORS") in (True, "true", "True") and os.environ.get(
     print(os.environ.get("CORS_URLS").split(","))
 
 app.include_router(route)
+app.include_router(tag_service_router)
 
 
 @app.on_event("startup")

scripts/constants/__init__.py

@@ -19,6 +19,9 @@ class Endpoints:
     api_hierarchy = "/hierarchy"
     hierarchy_api = 'ilens_config/get_site_level_hierarchy'
 
+    #tag_lists
+    api_tags_lists = "/tag_lists"
+    api_tags = "/tags"
 
 class StatusCodes:
     SUCCESS = [200, 201, 204]

scripts/core/handlers/tag_handler.py

+from scripts.db.mongo.ilens_configuration.aggregations.tag_hierarchy import TagHierarchyAggregate
+from scripts.db.mongo.ilens_configuration.collections.tag_hierarchy import TagHierarchy
+from scripts.db.mongo.schema.tag_hierarchy import GetTagsLists
+from scripts.constants.db_connections import mongo_client
+
+
+class TagHierarchyHandler:
+    def __init__(self, project_id=None):
+        self.tag_hierarchy_conn = TagHierarchy(mongo_client=mongo_client, project_id=project_id)
+
+    def get_tags_list_by_hierarchy(self, input_data: GetTagsLists):
+        try:
+            aggregate_query = TagHierarchyAggregate.tag_aggregate(project_id=input_data.project_id,
+                                                                  hierarchy_id=input_data.hierarchy_id)
+            tag_lists = self.tag_hierarchy_conn.get_tag_hierarchy_by_aggregate(query=aggregate_query)
+            tag_lists = tag_lists[0] if tag_lists else {}
+            return tag_lists
+        except Exception:
+            raise

scripts/db/mongo/ilens_configuration/aggregations/tag_hierarchy.py

+class TagHierarchyAggregate:
+    @staticmethod
+    def tag_aggregate(project_id, hierarchy_id):
+        query = [
+            {
+                '$match': {
+                    'project_id': project_id,
+                    'id': {
+                        '$regex': f'{hierarchy_id}\\$tag'
+                    }
+                }
+            }, {
+                '$lookup': {
+                    'from': 'tag_category',
+                    'localField': 'tag_category_id',
+                    'foreignField': 'tag_category_id',
+                    'as': 'tag_category'
+                }
+            }, {
+                '$unwind': {
+                    'path': '$tag_category'
+                }
+            }, {
+                '$group': {
+                    '_id': None,
+                    'data': {
+                        '$push': {
+                            'k': '$tag_category.tag_category_name',
+                            'v': '$id'
+                        }
+                    }
+                }
+            }, {
+                '$replaceRoot': {
+                    'newRoot': {
+                        '$arrayToObject': '$data'
+                    }
+                }
+            }
+        ]
+        return query

scripts/db/mongo/schema/tag_hierarchy.py

+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class GetTagsLists(BaseModel):
+    project_id : str
+    hierarchy_id : str

scripts/services/tag_list_services.py

+from fastapi import APIRouter
+
+from scripts.constants import Endpoints, ResponseCodes
+from scripts.core.handlers.tag_handler import TagHierarchyHandler
+from scripts.db.mongo.schema.tag_hierarchy import GetTagsLists
+from scripts.schemas.response_models import DefaultFailureResponse, DefaultResponse
+
+tag_service_router = APIRouter(prefix=Endpoints.api_tags,tags=["Tags Services"])
+
+
+@tag_service_router.post(Endpoints.api_tags_lists)
+def get_tags_list(request_data: GetTagsLists):
+    try:
+        tags_list = TagHierarchyHandler()
+        data = tags_list.get_tags_list_by_hierarchy(request_data)
+        return DefaultResponse(
+            data=data,
+            status=ResponseCodes.SUCCESS,
+            message="Tags Categories fetched successfully",
+        )
+    except Exception as e:
+        return DefaultFailureResponse(error=e)