feat(ingest/qlik): Qlik cloud connector integration

metadata-ingestion/docs/sources/qlik-sense/qlik-sense_recipe.yml

@@ -14,10 +14,10 @@ source:
 
     ingest_owner: true
 
-    # Optional -- This mapping is optional and only required to configure platform-instance for Qlik app dataset upstream source table
-    # A mapping of the Qlik app dataset upstream source to platform instance. Use <database>.<schema>.<table> as key.
-    # app_dataset_source_to_platform_instance:
-    #   <database>.<schema>.<table>:
+    # Optional -- This mapping is optional and only required to configure platform-instance for Qlik app dataset upstream source tables
+    # A mapping of the Qlik app dataset upstream tables from data connection to platform instance. Use 'data_connection_name' as key.
+    # data_connection_to_platform_instance:
+    #   data_connection_name:
     #     platform_instance: cloud_instance
     #     env: DEV
 

metadata-ingestion/src/datahub/ingestion/source/qlik_sense/config.py

@@ -21,20 +21,6 @@
 QLIK_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
 
 
-class WebSocketRequest:
-    """
-    Web socket send request dict
-    """
-
-    OPEN_DOC = {
-        "jsonrpc": "2.0",
-        "id": 1,
-        "handle": -1,
-        "method": "OpenDoc",
-        "params": {"qDocName": "f0714ca7-7093-49e4-8b58-47bb38563647"},
-    }
-
-
 class Constant:
     """
     keys used in qlik plugin
@@ -122,8 +108,9 @@ class QlikSourceConfig(
         default=True,
         description="Ingest Owner from source. This will override Owner info entered from UI",
     )
-    # Qlik app dataset upstream source to platform instance mapping
-    app_dataset_source_to_platform_instance: Dict[str, PlatformDetail] = pydantic.Field(
+    # Qlik app dataset upstream tables from data connection to platform instance mapping
+    data_connection_to_platform_instance: Dict[str, PlatformDetail] = pydantic.Field(
         default={},
-        description="A mapping of the Qlik app dataset upstream source to platform instance. Use <database>.<schema>.<table> as key.",
+        description="A mapping of the Qlik app dataset upstream tables from data connection to platform instance."
+        "Use 'data_connection_name' as key.",
     )

metadata-ingestion/src/datahub/ingestion/source/qlik_sense/qlik_api.py

@@ -24,6 +24,7 @@
 class QlikAPI:
     def __init__(self, config: QlikSourceConfig) -> None:
         self.spaces: Dict = {}
+        self.users: Dict = {}
         self.config = config
         self.session = requests.Session()
         self.session.headers.update(
@@ -74,6 +75,21 @@ def _get_dataset(self, dataset_id: str) -> Optional[QlikDataset]:
             )
         return None
 
+    def get_user_name(self, user_id: str) -> Optional[str]:
+        try:
+            if user_id in self.users:
+                # To avoid fetching same user details again
+                return self.users[user_id]
+            else:
+                response = self.session.get(f"{self.rest_api_url}/users/{user_id}")
+                response.raise_for_status()
+                user_name = response.json()[Constant.NAME]
+                self.users[user_id] = user_name
+                return user_name
+        except Exception:
+            self._log_http_error(message=f"Unable to fetch user with id {user_id}")
+        return None
+
     def _get_sheet(
         self,
         websocket_connection: WebsocketConnection,

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

@@ -171,14 +171,17 @@ def _gen_space_workunit(self, space: Space) -> Iterable[MetadataWorkUnit]:
         """
         Map Qlik space to Datahub container
         """
+        owner_username: Optional[str] = None
+        if space.ownerId:
+            owner_username = self.qlik_api.get_user_name(space.ownerId)
         yield from gen_containers(
             container_key=self._gen_space_key(space.id),
             name=space.name,
             description=space.description,
             sub_types=[BIContainerSubTypes.QLIK_SPACE],
             extra_properties={Constant.TYPE: str(space.type)},
-            owner_urn=builder.make_user_urn(space.ownerId)
-            if self.config.ingest_owner and space.ownerId
+            owner_urn=builder.make_user_urn(owner_username)
+            if self.config.ingest_owner and owner_username
             else None,
             created=int(space.createdAt.timestamp() * 1000),
             last_modified=int(space.updatedAt.timestamp() * 1000),
@@ -190,12 +193,12 @@ def _gen_entity_status_aspect(self, entity_urn: str) -> MetadataWorkUnit:
         ).as_workunit()
 
     def _gen_entity_owner_aspect(
-        self, entity_urn: str, owner_id: str
+        self, entity_urn: str, user_name: str
     ) -> MetadataWorkUnit:
         aspect = OwnershipClass(
             owners=[
                 OwnerClass(
-                    owner=builder.make_user_urn(owner_id),
+                    owner=builder.make_user_urn(user_name),
                     type=OwnershipTypeClass.DATAOWNER,
                 )
             ]
@@ -281,8 +284,9 @@ def _gen_sheets_workunit(
             if dpi_aspect:
                 yield dpi_aspect
 
-            if self.config.ingest_owner:
-                yield self._gen_entity_owner_aspect(dashboard_urn, sheet.ownerId)
+            owner_username = self.qlik_api.get_user_name(sheet.ownerId)
+            if self.config.ingest_owner and owner_username:
+                yield self._gen_entity_owner_aspect(dashboard_urn, owner_username)
 
             yield from self._gen_charts_workunit(sheet.charts)
 
@@ -292,8 +296,8 @@ def _gen_app_dataset_upstream_lineage(
         dataset_identifier = self._get_app_dataset_identifier(dataset)
         dataset_urn = self._gen_qlik_dataset_urn(dataset_identifier)
         upstream_dataset_platform_detail = (
-            self.config.app_dataset_source_to_platform_instance.get(
-                dataset_identifier,
+            self.config.data_connection_to_platform_instance.get(
+                dataset.dataconnectorName,
                 PlatformDetail(
                     env=self.config.env, platform_instance=self.config.platform_instance
                 ),
@@ -339,7 +343,9 @@ def _gen_app_dataset_properties(self, dataset: QlikAppDataset) -> MetadataWorkUn
         ).as_workunit()
 
     def _get_app_dataset_identifier(self, dataset: QlikAppDataset) -> str:
-        return f"{dataset.databaseName}.{dataset.schemaName}.{dataset.tableName}"
+        return (
+            f"{dataset.databaseName}.{dataset.schemaName}.{dataset.tableName}".lower()
+        )
 
     def _gen_app_dataset_workunit(
         self, datasets: List[QlikAppDataset], app_id: str
@@ -375,15 +381,16 @@ def _gen_app_workunit(self, app: App) -> Iterable[MetadataWorkUnit]:
         """
         Map Qlik App to Datahub container
         """
+        owner_username = self.qlik_api.get_user_name(app.ownerId)
         yield from gen_containers(
             container_key=self._gen_app_key(app.id),
             name=app.qTitle,
             description=app.description,
             sub_types=[BIContainerSubTypes.QLIK_APP],
             parent_container_key=self._gen_space_key(app.spaceId),
             extra_properties={Constant.QRI: app.qri, Constant.USAGE: app.qUsage},
-            owner_urn=builder.make_user_urn(app.ownerId)
-            if self.config.ingest_owner
+            owner_urn=builder.make_user_urn(owner_username)
+            if self.config.ingest_owner and owner_username
             else None,
             created=int(app.createdAt.timestamp() * 1000),
             last_modified=int(app.updatedAt.timestamp() * 1000),
@@ -427,7 +434,6 @@ def _gen_schema_fields(
                     if field.dataType
                     else NullType()
                 ),
-                # NOTE: nativeDataType will not be in sync with older connector
                 nativeDataType=field.dataType if field.dataType else "",
                 nullable=field.nullable,
                 isPartOfKey=field.primaryKey,
@@ -496,8 +502,9 @@ def _gen_dataset_workunit(self, dataset: QlikDataset) -> Iterable[MetadataWorkUn
         if dpi_aspect:
             yield dpi_aspect
 
-        if self.config.ingest_owner:
-            yield self._gen_entity_owner_aspect(dataset_urn, dataset.ownerId)
+        owner_username = self.qlik_api.get_user_name(dataset.ownerId)
+        if self.config.ingest_owner and owner_username:
+            yield self._gen_entity_owner_aspect(dataset_urn, owner_username)
 
         yield MetadataChangeProposalWrapper(
             entityUrn=dataset_urn,