Feature to add a filtering options on the API call

tap_clickup/streams.py

@@ -4,6 +4,50 @@
 import requests
 from singer_sdk.helpers.jsonpath import extract_jsonpath
 from tap_clickup.client import ClickUpStream
+import yaml #added to read information specifying spaces and workspaces
+
+# Load the YAML config file from meltano
+with open("meltano.yml", "r") as yaml_file:
+    cu_config = yaml.safe_load(yaml_file)
+
+#funcions to read the extra configuration data from the YAML file
+#read and split lists to create list
+def extract_and_convert_list(config, key):
+    value = config.get(key, "")
+    if value:
+        split_values = value.split(',')
+        return [int(item) for item in split_values]
+    return []
+
+#specify which setting from which tap to read the config.info from ISSUE HERE the yaml read is hardcoded so breaks when i have multiple "tap-clickup" one for each env that i want to switch between
+def find_tap_clickup_config(plugins):
+    for plugin in plugins:
+        if plugin.get("name") == "tap-clickup":
+            return plugin.get("config", {})
+    return {}
+
+# Find the tap-clickup configuration
+tap_clickup_config = find_tap_clickup_config(cu_config["plugins"]["extractors"])
+
+# Extract and convert workspace ID
+cu_workspace = tap_clickup_config.get("workspace_id")
+
+# Extract and convert spaces and lists into lists
+spaces_id_list = extract_and_convert_list(tap_clickup_config, "spaces_id")
+lists_id_list = extract_and_convert_list(tap_clickup_config, "list_ids")
+
+workteamid = cu_workspace  # store workspace id E.g:30979640  
+space_ids = spaces_id_list # store list of spaces to be fetched E.g: [90100432266,90100432289,90100437857]
+list_ids = lists_id_list #store lists of  lists to be fetched E.g: [900101856869,901002404954]
+
+# Check if there's only one value in the list, this is necessary because of a bug on click up's API. 
+if len(list_ids) == 1:
+    # Duplicate the value to ensure at least two parameters due to a bug on clickups API (the values returned by the API are NOT duplicated)
+    list_ids.append(list_ids[0])
+if len(space_ids) == 1:
+    # Duplicate the value to ensure at least two parameters due to a bug on clickups API (the values returned by the API are NOT duplicated)
+    space_ids.append(space_ids[0])    
+
 
 SCHEMAS_DIR = Path(__file__).parent / Path("./schemas")
 
@@ -16,7 +60,9 @@ class TeamsStream(ClickUpStream):
     primary_keys = ["id"]
     replication_key = None
     schema_filepath = SCHEMAS_DIR / "team.json"
-    records_jsonpath = "$.teams[*]"
+    # Necessary because if you have access to multiple workteams, the responses are replicated N times where N = # of worspaces you have access to
+    records_jsonpath = f"$.teams[?(@.id == {workteamid})]" 
+
 
     def get_child_context(self, record: dict, context: Optional[dict]) -> dict:
         """Return a context dictionary for child streams."""
@@ -25,6 +71,7 @@ def get_child_context(self, record: dict, context: Optional[dict]) -> dict:
         }
 
 
+
 class TimeEntries(ClickUpStream):
     """Time Entries"""
 
@@ -220,20 +267,26 @@ class TasksStream(ClickUpStream):
 
     name = "task"
     # Date_updated_gt is greater than or equal to not just greater than
-    path = "/team/{team_id}/task"
     primary_keys = ["id"]
     replication_key = "date_updated"
     is_sorted = True
     # ignore_parent_replication_key = True
     schema_filepath = SCHEMAS_DIR / "task.json"
     records_jsonpath = "$.tasks[*]"
     parent_stream_type = TeamsStream
-
     # Need this stub as a hack on _sync to force it to use Partitions
     # Since this is a child stream we want each team_id to create a request for
     # archived:true and archived:false. And we want state to track properly
     partitions = []
-
+    basepath = "/team/{team_id}/task"
+    space_ids_param = "&space_ids=" + "&space_ids=".join(map(str, space_ids)) if space_ids else "" #Dinamically create the spaces path to be passed on API call
+    list_ids_param = "?list_ids=" + "&list_ids=".join(map(str, list_ids)) if list_ids else "" #Dinamically create the spaces path to be passed on API call
+    # Necessary because if you pass list and space, lists MUST go first in the parameter string
+    if list_ids_param:
+        path = f"{basepath}{list_ids_param}{space_ids_param}"
+    else:
+        path = f"{basepath}?{space_ids_param}"
+
     @property
     def base_partition(self):
         return [{"archived": "true"}, {"archived": "false"}]
@@ -262,7 +315,6 @@ def get_next_page_token(
 
         for _ in extract_jsonpath(self.records_jsonpath, input=response.json()):
             recordcount = recordcount + 1
-
         # I wonder if a better approach is to just check for 0 records and stop
         # For now I'll follow the docs verbatium
         # From the api docs, https://clickup.com/api.
@@ -273,4 +325,4 @@ def get_next_page_token(
         else:
             newtoken = None
 
-        return newtoken
+        return newtoken

tap_clickup/tap.py

@@ -47,6 +47,15 @@ class TapClickUp(Tap):
         th.Property(
             "api_token", th.StringType, required=True, description="Example: 'pk_12345"
         ),
+        th.Property(
+            "workspace_id", th.IntegerType, required=False, description="Example: '20214542" # fetches the data for workspace_id
+        ),
+        th.Property(
+            "spaces_id", th.StringType, required=False, description="Example: '[45215477,4547547]" # fetches the data for workspace_id
+        ),
+        th.Property(
+            "list_ids", th.StringType, required=False, description="Example: '[454455478,784552187]" # fetches the data for workspace_id
+        ),                
         # Removing "official" start_date support re https://github.com/AutoIDM/tap-clickup/issues/118
         #        th.Property(
         #            "start_date",
@@ -64,3 +73,4 @@ class TapClickUp(Tap):
     def discover_streams(self) -> List[Stream]:
         """Return a list of discovered streams."""
         return [stream_class(tap=self) for stream_class in STREAM_TYPES]
+