rest_api: filter, exclude, transform API responses

sources/rest_api/__init__.py

@@ -211,6 +211,9 @@ def create_resources(
         request_params = endpoint_config.get("params", {})
         request_json = endpoint_config.get("json", None)
         paginator = create_paginator(endpoint_config.get("paginator"))
+        row_filter = endpoint_resource.pop("filter", None)
+        transform = endpoint_resource.pop("transform", None)
+        exclude_columns = endpoint_resource.pop("exclude_columns", [])
 
         resolved_param: ResolvedParam = resolved_param_map[resource_name]
 
@@ -254,21 +257,39 @@ def paginate_resource(
                 client: RESTClient = client,
                 incremental_object: Optional[Incremental[Any]] = incremental_object,
                 incremental_param: IncrementalParam = incremental_param,
+                row_filter: Any = row_filter,
+                transform: Any = transform,
+                exclude_columns: List[str] = exclude_columns,
             ) -> Generator[Any, None, None]:
+                def exclude_elements(item: Any, exclude_columns: List[str]) -> Any:
+                    for key in exclude_columns:
+                        del item[key]
+                    return item
+
                 if incremental_object:
                     params[incremental_param.start] = incremental_object.last_value
                     if incremental_param.end:
                         params[incremental_param.end] = incremental_object.end_value
 
-                yield from client.paginate(
+                for page in client.paginate(
                     method=method,
                     path=path,
                     params=params,
-                    json=json,
                     paginator=paginator,
                     data_selector=data_selector,
                     hooks=hooks,
-                )
+                ):
+                    filtered_page = filter(row_filter, page) if row_filter else page
+                    transformed_page = (
+                        map(transform, filtered_page) if transform else filtered_page
+                    )
+                    final_page = (
+                        map(exclude_elements, transformed_page, exclude_columns)
+                        if exclude_elements
+                        else transformed_page
+                    )
+
+                    yield final_page
 
             resources[resource_name] = dlt.resource(
                 paginate_resource,
@@ -281,6 +302,9 @@ def paginate_resource(
                 paginator=paginator,
                 data_selector=endpoint_config.get("data_selector"),
                 hooks=hooks,
+                row_filter=row_filter,
+                transform=transform,
+                exclude_columns=exclude_columns,
             )
 
         else:

sources/rest_api/typing.py

@@ -1,5 +1,6 @@
 from typing import (
     Any,
+    Callable,
     Dict,
     List,
     Literal,
@@ -243,6 +244,9 @@ class ResourceBase(TypedDict, total=False):
     table_format: Optional[TTableHintTemplate[TTableFormat]]
     selected: Optional[bool]
     parallelized: Optional[bool]
+    row_filter: Optional[Callable[[Any], None]]
+    exclude_columns: Optional[List[str]]
+    transform: Optional[Callable[[Any], None]]
 
 
 class EndpointResourceBase(ResourceBase, total=False):