fix: remove catalog endpoint parameter

Author: botanical

dags/veda_data_pipeline/veda_tenant_tagging_pipeline.py

@@ -17,12 +17,7 @@
     "collections": Param(
         default=None,
         type=["null", "array"],
-        description="List of collection IDs to tag (optional if catalog_endpoint is provided)"
-    ),
-    "catalog_endpoint": Param(
-        default=None,
-        type=["null", "string"],
-        description="STAC catalog endpoint URL to fetch all collections from (optional if collections is provided)"
+        description="List of collection IDs to tag"
     ),
     "tenant": Param(default=None, type="string", description="Tenant ID to tag the collection with"),
     "tenant_field": Param(
@@ -42,7 +37,7 @@
 Tags existing collections with tenant information by updating their properties field.
 
 This pipeline:
-1. Fetches existing collections from the STAC catalog (either from a list or from a catalog endpoint)
+1. Fetches existing collections from the STAC catalog
 2. Updates each collection's properties with tenant tags
 3. Re-ingests the updated collections
 
@@ -51,10 +46,8 @@
 **Required Parameters:**
 - `tenant` (string): Tenant ID to tag collections with
 
-**Collection Source (provide one of the following):**
+**Collection Source:**
 - `collections` (array of strings): List of collection IDs to tag
-- `catalog_endpoint` (string): STAC catalog endpoint URL (e.g., `https://dev.openveda.cloud/api/stac/collections`) to fetch all collections from
-
 **Optional Parameters:**
 - `tenant_field` (string): Properties key to write tenant into (default: `eic:tenant`)
 - `properties` (object): Additional properties to add/update on collections
@@ -78,13 +71,6 @@
 }
 ```
 
-**Tag all collections from a catalog:**
-```json
-{
-    "catalog_endpoint": "https://dev.openveda.cloud/api/stac/collections",
-    "tenant": "tenant-123"
-}
-```
 
 **With additional properties:**
 ```json
@@ -109,51 +95,16 @@
 
 @task()
 def get_collection_ids(ti=None):
-    """Extract and validate collection IDs from configuration or fetch from catalog endpoint"""
+    """Extract and validate collection IDs from configuration"""
     try:
         config = ti.dag_run.conf
         collections = config.get("collections")
-        catalog_endpoint = config.get("catalog_endpoint")
         tenant = config.get("tenant")
 
         logger.info(f"Starting collection ID validation. Tenant: {tenant}")
 
-        # If catalog_endpoint is provided, fetch all collections
-        if catalog_endpoint:
-            logger.info(f"Fetching all collections from catalog endpoint: {catalog_endpoint}")
-
-            try:
-                response = requests.get(catalog_endpoint, timeout=30)
-                response.raise_for_status()
-            except requests.exceptions.RequestException as e:
-                error_msg = f"Failed to fetch collections from catalog endpoint {catalog_endpoint}: {str(e)}"
-                logger.error(error_msg)
-                raise ValueError(error_msg) from e
-
-            try:
-                catalog_data = response.json()
-            except (ValueError, requests.exceptions.JSONDecodeError) as json_error:
-                error_msg = f"Failed to parse JSON response from catalog endpoint {catalog_endpoint}"
-                logger.error(error_msg)
-                raise ValueError(error_msg) from json_error
-
-            # Extract collection IDs
-            if "collections" in catalog_data:
-                collections = [coll.get("id") for coll in catalog_data["collections"] if coll.get("id")]
-            else:
-                error_msg = f"Unexpected response format from catalog endpoint {catalog_endpoint}"
-                logger.error(error_msg)
-                raise ValueError(error_msg)
-
-            if not collections:
-                error_msg = f"No collections found at catalog endpoint {catalog_endpoint}"
-                logger.error(error_msg)
-                raise ValueError(error_msg)
-
-            logger.info(f"Found {len(collections)} collections from catalog endpoint")
-
         if not collections:
-            error_msg = "Either 'collections' list or 'catalog_endpoint' must be provided in DAG configuration"
+            error_msg = "The 'collections' list must be provided in DAG configuration"
             logger.error(error_msg)
             raise ValueError(error_msg)