fix(ingest/fivetran): handle missing user_id for a connection (datahub-project#9460)

Co-authored-by: Harshal Sheth <[email protected]>

Author: ryaminal

metadata-ingestion/src/datahub/ingestion/source/fivetran/data_classes.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List
+from typing import List, Optional
 
 
 @dataclass
@@ -23,7 +23,7 @@ class Connector:
     paused: bool
     sync_frequency: int
     destination_id: str
-    user_name: str
+    user_name: Optional[str]
     table_lineage: List[TableLineage]
     jobs: List["Job"]
 

metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran.py

@@ -179,7 +179,7 @@ def _generate_datajob_from_connector(self, connector: Connector) -> DataJob:
             id=connector.connector_id,
             flow_urn=dataflow_urn,
             name=connector.connector_name,
-            owners={connector.user_name},
+            owners={connector.user_name} if connector.user_name else set(),
         )
 
         job_property_bag: Dict[str, str] = {}

metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_log_api.py

@@ -117,7 +117,9 @@ def _get_jobs_list(self, connector_id: str) -> List[Job]:
             )
         return jobs
 
-    def _get_user_name(self, user_id: str) -> str:
+    def _get_user_name(self, user_id: Optional[str]) -> Optional[str]:
+        if not user_id:
+            return None
         user_details = self._query(FivetranLogQuery.get_user_query(user_id=user_id))[0]
         return (
             f"{user_details[Constant.GIVEN_NAME]} {user_details[Constant.FAMILY_NAME]}"