Merge pull request #1106 from hkad98/jkd/bounded-filter

feat: add bounded filter

Author: hkad98

gooddata-pipelines/poetry.lock

@@ -253,6 +253,7 @@
     AllMetricValueFilter,
     AllTimeFilter,
     AttributeFilter,
+    BoundedFilter,
     Filter,
     InlineFilter,
     MetricValueFilter,

gooddata-sdk/gooddata_sdk/__init__.py

@@ -6,6 +6,7 @@
 from gooddata_sdk.compute.model.filter import (
     AbsoluteDateFilter,
     AllTimeFilter,
+    BoundedFilter,
     Filter,
     MetricValueFilter,
     NegativeAttributeFilter,
@@ -71,11 +72,22 @@ def convert_filter(filter_dict: dict[str, Any]) -> Filter:
             if ("from" not in f) or ("to" not in f):
                 return AllTimeFilter(ref_extract_obj_id(f["dataset"]))
 
+            # Extract bounded filter if present
+            bounded_filter = None
+            if "boundedFilter" in f:
+                bf = f["boundedFilter"]
+                bounded_filter = BoundedFilter(
+                    granularity=bf["granularity"],
+                    from_shift=bf.get("from"),
+                    to_shift=bf.get("to"),
+                )
+
             return RelativeDateFilter(
                 dataset=ref_extract_obj_id(f["dataset"]),
                 granularity=f["granularity"],
                 from_shift=f["from"],
                 to_shift=f["to"],
+                bounded_filter=bounded_filter,
             )
 
         if "absoluteDateFilter" in filter_dict:

gooddata-sdk/gooddata_sdk/compute/compute_to_sdk_converter.py

@@ -5,6 +5,7 @@
 from importlib.util import find_spec
 from typing import Any, Optional, Union
 
+import attrs
 from gooddata_api_client.model.inline_filter_definition_inline import InlineFilterDefinitionInline
 
 if find_spec("icu") is not None:
@@ -153,8 +154,35 @@ def description(self, labels: dict[str, str], format_locale: Optional[str] = Non
 }
 
 
+@attrs.define
+class BoundedFilter:
+    granularity: str
+    from_shift: Optional[int] = None
+    to_shift: Optional[int] = None
+
+    def __attrs_post_init__(self) -> None:
+        # Validate that exactly one of from_shift or to_shift is set
+        if (self.from_shift is None) == (self.to_shift is None):
+            raise ValueError("BoundedFilter requires exactly one of `from_shift` or `to_shift` to be set.")
+
+    def as_api_model(self) -> afm_models.BoundedFilter:
+        return afm_models.BoundedFilter(
+            granularity=self.granularity,
+            _from=self.from_shift,
+            to=self.to_shift,
+            _check_type=False,
+        )
+
+
 class RelativeDateFilter(Filter):
-    def __init__(self, dataset: ObjId, granularity: str, from_shift: int, to_shift: int) -> None:
+    def __init__(
+        self,
+        dataset: ObjId,
+        granularity: str,
+        from_shift: int,
+        to_shift: int,
+        bounded_filter: Optional[BoundedFilter] = None,
+    ) -> None:
         super().__init__()
 
         if granularity not in _GRANULARITY:
@@ -166,6 +194,7 @@ def __init__(self, dataset: ObjId, granularity: str, from_shift: int, to_shift:
         self._granularity = granularity
         self._from_shift = from_shift
         self._to_shift = to_shift
+        self._bounded_filter = bounded_filter
 
     @property
     def dataset(self) -> ObjId:
@@ -183,17 +212,26 @@ def from_shift(self) -> int:
     def to_shift(self) -> int:
         return self._to_shift
 
+    @property
+    def bounded_filter(self) -> Optional[BoundedFilter]:
+        return self._bounded_filter
+
     def is_noop(self) -> bool:
         return False
 
     def as_api_model(self) -> afm_models.RelativeDateFilter:
-        body = RelativeDateFilterBody(
-            dataset=self.dataset.as_afm_id(),
-            granularity=self.granularity,
-            _from=self.from_shift,
-            to=self.to_shift,
-            _check_type=False,
-        )
+        body_params = {
+            "dataset": self.dataset.as_afm_id(),
+            "granularity": self.granularity,
+            "_from": self.from_shift,
+            "to": self.to_shift,
+            "_check_type": False,
+        }
+
+        if self.bounded_filter is not None:
+            body_params["bounded_filter"] = self.bounded_filter.as_api_model()
+
+        body = RelativeDateFilterBody(**body_params)
         return afm_models.RelativeDateFilter(body)
 
     def description(self, labels: dict[str, str], format_locale: Optional[str] = None) -> str: