check if conn_is is uuid to prevent failed request

src/datacat_integration/secrets.py

@@ -4,13 +4,22 @@ from airflow.secrets import BaseSecretsBackend
 import requests
 import logging
 import json
+import uuid
 
-from datacat_integration.connection import DataCatConnection, DataCatalogEntry, get_connection_from_entry
+from datacat_integration.connection import DataCatConnection, get_connection_from_entry
 
 connection_backend_type = "airflow_connections"
 
 log = logging.getLogger(__name__)
 
+def validate_uuid(uuid_to_test: str, version=4):
+    """Check if uuid_to_test id a valid UUID. Returns True if it is and false if it is not."""
+    try:
+        uuid_obj = uuid.UUID(uuid_to_test, version=version)
+    except ValueError:
+        return False
+    return str(uuid_obj) == uuid_to_test
+
 def get_connection_with_secrets_from_entry(data: Dict[str, Any], secrets: Dict[str, str] , datacat_type: str, oid: str):
     """returns an aiflow connection from the data provided in the datacat entry and the secrets."""
     conn = get_connection_from_entry(data, datacat_type, oid)
@@ -85,7 +94,10 @@ class DatacatSecretsBackend(BaseSecretsBackend):
     def get_connection(self, conn_id: str):
         """returns a Connection created from the <conenction_type>/<conn_id> entry in the datacatalog"""
         
-        log.debug(f"Get connection: {conn_id}")
+        log.debug(f"Get connection from datacat secrets backend: {conn_id}")
+
+        if not validate_uuid(conn_id):
+            return None # can not be a connection in the datacat, immediatly return None to allow airflow to look elsewhere
 
         secrets_conn = DataCatConnectionWithSecrets(self.url, self.user, self.password)
         data: Dict[str,str] = json.loads(secrets_conn.get_entry(connection_backend_type, conn_id))