metadata-ingestion:lintFix

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

@@ -21,7 +21,8 @@
     make_data_platform_urn,
     make_dataset_urn,
     make_dataset_urn_with_platform_instance,
-    make_domain_urn, make_term_urn,
+    make_domain_urn,
+    make_term_urn,
 )
 from datahub.emitter.mcp import MetadataChangeProposalWrapper
 from datahub.emitter.mcp_builder import add_domain_to_entity_wu
@@ -37,7 +38,7 @@
 )
 from datahub.ingestion.api.source import MetadataWorkUnitProcessor
 from datahub.ingestion.api.workunit import MetadataWorkUnit
-from datahub.ingestion.graph.client import DataHubGraph, DatahubClientConfig
+from datahub.ingestion.graph.client import DatahubClientConfig, DataHubGraph
 from datahub.ingestion.source.sql.sql_types import resolve_sql_type
 from datahub.ingestion.source.sql.sqlalchemy_uri_mapper import (
     get_platform_from_sqlalchemy_uri,
@@ -51,8 +52,12 @@
     StatefulIngestionConfigBase,
     StatefulIngestionSourceBase,
 )
-from datahub.metadata._schema_classes import AuditStampClass, GlossaryTermsClass, GlossaryTermAssociationClass, \
-    GlossaryTermInfoClass
+from datahub.metadata._schema_classes import (
+    AuditStampClass,
+    GlossaryTermAssociationClass,
+    GlossaryTermInfoClass,
+    GlossaryTermsClass,
+)
 from datahub.metadata.com.linkedin.pegasus2avro.common import (
     AuditStamp,
     ChangeAuditStamps,
@@ -589,7 +594,11 @@ def gen_dataset_urn(self, datahub_dataset_name: str) -> str:
 
     def check_if_term_exists(self, term_urn):
         graph = DataHubGraph(
-            DatahubClientConfig(server=self.sink_config.get("server", ""), token=self.sink_config.get("token", "")))
+            DatahubClientConfig(
+                server=self.sink_config.get("server", ""),
+                token=self.sink_config.get("token", ""),
+            )
+        )
         # Query multiple aspects from entity
         result = graph.get_entity_semityped(
             entity_urn=term_urn,
@@ -600,7 +609,9 @@ def check_if_term_exists(self, term_urn):
             return True
         return False
 
-    def parse_glossary_terms_from_metrics(self, metrics, last_modified) -> GlossaryTermsClass:
+    def parse_glossary_terms_from_metrics(
+        self, metrics, last_modified
+    ) -> GlossaryTermsClass:
         glossary_term_urns = []
         for metric in metrics:
             expression = metric.get("expression", "")
@@ -626,8 +637,10 @@ def parse_glossary_terms_from_metrics(self, metrics, last_modified) -> GlossaryT
             )
 
             # Create rest emitter
-            rest_emitter = DatahubRestEmitter(gms_server=self.sink_config.get("server", ""),
-                                              token=self.sink_config.get("token", ""))
+            rest_emitter = DatahubRestEmitter(
+                gms_server=self.sink_config.get("server", ""),
+                token=self.sink_config.get("token", ""),
+            )
             rest_emitter.emit(event)
             logger.info(f"Created Glossary term {term_urn}")
             glossary_term_urns.append(GlossaryTermAssociationClass(urn=term_urn))
@@ -649,7 +662,9 @@ def construct_dataset_from_dataset_data(
     ) -> DatasetSnapshot:
         dataset_response = self.get_dataset_info(dataset_data.get("id"))
         dataset = SupersetDataset(**dataset_response["result"])
-        datasource_urn = self.get_datasource_urn_from_id(dataset_response, self.platform)
+        datasource_urn = self.get_datasource_urn_from_id(
+            dataset_response, self.platform
+        )
         now = datetime.now().strftime("%I:%M%p on %B %d, %Y")
         modified_ts = int(
             dp.parse(dataset_data.get("changed_on") or now).timestamp() * 1000
@@ -679,7 +694,9 @@ def construct_dataset_from_dataset_data(
         response_result = dataset_response.get("result", {})
 
         if self._is_certified_metric(response_result):
-            glossary_terms = self.parse_glossary_terms_from_metrics(response_result.get("metrics", {}), last_modified)
+            glossary_terms = self.parse_glossary_terms_from_metrics(
+                response_result.get("metrics", {}), last_modified
+            )
             aspects_items.append(glossary_terms)
 
         dataset_snapshot = DatasetSnapshot(