CM-71: added support for applying queryset based on custom filters provided by frontend

social_protection/custom_filters.py

@@ -1,6 +1,8 @@
 import logging
+import re
 
 from collections import namedtuple
+from django.db.models.query import QuerySet
 from typing import List
 
 from core.custom_filters import CustomFilterWizardInterface
@@ -30,8 +32,8 @@ def load_definition(self, tuple_type: type, **kwargs) -> List[namedtuple]:
         This method retrieves the definition of how to create filters and returns it as a list of named tuples.
         Each named tuple is built with the provided `tuple_type` and has the fields `field`, `filter`, and `value`.
 
-        Example named tuple: <Type>(field=<str>, filter=<str>, value=<str>)
-        Example usage: BenefitPlan(field='income', filter='lt, gte, icontains, exact', value='')
+        Example named tuple: <Type>(field=<str>, filter=<str>, type=<str>)
+        Example usage: BenefitPlan(field='income', filter='lt, gte, icontains, exact', type='integer')
 
         :param tuple_type: The type of the named tuple.
         :type tuple_type: type
@@ -44,6 +46,25 @@ def load_definition(self, tuple_type: type, **kwargs) -> List[namedtuple]:
         list_of_tuple_with_definitions = self.__process_schema_and_build_tuple(benefit_plan, tuple_type)
         return list_of_tuple_with_definitions
 
+    def apply_filter_to_queryset(self, custom_filters: List[namedtuple], query: QuerySet) -> QuerySet:
+        """
+        Apply custom filters to a queryset.
+
+        :param custom_filters: Structure of custom filter tuple: <Type>(field=<str>, filter=<str>, type=<str>).
+        Example usage of filter tuple: BenefitPlan(field='income', filter='lt, gte, icontains, exact', type='integer')
+
+        :param query: The original queryset with filters for example: Queryset[Beneficiary].
+
+        :return: The updated queryset with additional filters applied for example: Queryset[Beneficiary].
+        """
+        for filter_part in custom_filters:
+            field, value = filter_part.split('=')
+            field, value_type = field.rsplit('__', 1)
+            value = self.__cast_value(value, value_type)
+            filter_kwargs = {f"json_ext__{field}": value}
+            query = query.filter(**filter_kwargs)
+        return query
+
     def __process_schema_and_build_tuple(
         self,
         benefit_plan: BenefitPlan,
@@ -65,3 +86,33 @@ def __process_schema_and_build_tuple(
                            'on the provided schema due to either empty schema '
                            'or missing properties in schema file')
         return tuples_with_definitions
+
+    def __cast_value(self, value: str, value_type: str):
+        if value_type == 'integer':
+            return int(value)
+        elif value_type == 'string':
+            return str(value[1:-1])
+        elif value_type == 'numeric':
+            return float(value)
+        elif value_type == 'boolean':
+            cleaned_value = self.__remove_unexpected_chars(value)
+            if cleaned_value.lower() == 'true':
+                return True
+            elif cleaned_value.lower() == 'false':
+                return False
+        elif value_type == 'date':
+            # Perform date parsing logic here
+            # Assuming you have a specific date format, you can use datetime.strptime
+            # Example: return datetime.strptime(value, '%Y-%m-%d').date()
+            pass
+
+        # Return None if the value type is not recognized
+        return None
+
+    def __remove_unexpected_chars(self, string: str):
+        pattern = r'[^\w\s]'  # Remove any character that is not alphanumeric or whitespace
+
+        # Use re.sub() to remove the unwanted characters
+        cleaned_string = re.sub(pattern, '', string)
+
+        return cleaned_string

social_protection/schema.py

@@ -1,5 +1,7 @@
 import graphene
 import pandas as pd
+import re
+
 from django.contrib.auth.models import AnonymousUser
 from django.db.models import Q
 from django.core.exceptions import PermissionDenied
@@ -11,6 +13,7 @@
 from core.schema import OrderedDjangoFilterConnectionField
 from core.utils import append_validity_filter
 from social_protection.apps import SocialProtectionConfig
+from social_protection.custom_filters import BenefitPlanCustomFilterWizard
 from social_protection.gql_mutations import (
     CreateBenefitPlanMutation,
     UpdateBenefitPlanMutation,
@@ -40,6 +43,7 @@ def patch_details(beneficiary_df: pd.DataFrame):
     df_final = df_final.drop('json_ext', axis=1)
     return df_final
 
+
 class Query(ExportableQueryMixin, graphene.ObjectType):
     export_patches = {
         'beneficiary': [
@@ -50,6 +54,7 @@ class Query(ExportableQueryMixin, graphene.ObjectType):
         ]
     }
     exportable_fields = ['beneficiary', 'group_beneficiary']
+    type_of_custom_filter_wizard = BenefitPlanCustomFilterWizard
 
     benefit_plan = OrderedDjangoFilterConnectionField(
         BenefitPlanGQLType,
@@ -67,7 +72,8 @@ class Query(ExportableQueryMixin, graphene.ObjectType):
         dateValidFrom__Gte=graphene.DateTime(),
         dateValidTo__Lte=graphene.DateTime(),
         applyDefaultValidityFilter=graphene.Boolean(),
-        client_mutation_id=graphene.String()
+        client_mutation_id=graphene.String(),
+        customFilters=graphene.List(of_type=graphene.String),
     )
     group_beneficiary = OrderedDjangoFilterConnectionField(
         GroupBeneficiaryGQLType,
@@ -154,6 +160,10 @@ def resolve_beneficiary(self, info, **kwargs):
             SocialProtectionConfig.gql_beneficiary_search_perms
         )
         query = Beneficiary.objects.filter(*filters)
+
+        custom_filters = kwargs.get("customFilters", None)
+        if custom_filters:
+            query = BenefitPlanCustomFilterWizard().apply_filter_to_queryset(custom_filters, query)
         return gql_optimizer.query(query, info)
 
     def resolve_group_beneficiary(self, info, **kwargs):