fix the dataplex entity for multiple matches for bigquery uri (#192)

* fix the dataplex entity for multiple matches for bigquery uri

* fix the dataplex entity for multiple matches for bigquery uri

* incorporated review comments

* updated loggers

Author: AmandeepSinghCS

clouddq/classes/dataplex_entity.py

@@ -174,7 +174,7 @@ def to_dict(self: DataplexEntity) -> dict:
             "dataPath": self.dataPath,
             "system": self.system,
             "format": self.format,
-            "schema": self.schema,
+            "schema": self.schema.to_dict(),
             "project_id": self.project_id,
             "location": self.location,
             "lake": self.lake,

clouddq/classes/dq_configs_cache.py

@@ -534,16 +534,24 @@ def is_dataplex_entity(
             data_path=entity_uri.get_entity_id(),
         )
         logger.info(f"Retrieved Dataplex Entities:\n{pformat(dataplex_entities_match)}")
-        if len(dataplex_entities_match) != 1:
+        if len(dataplex_entities_match) == 0:
             logger.info(
                 "Failed to retrieve Dataplex Metadata entry for "
                 f"entity_uri '{entity_uri.complete_uri_string}' "
                 f"with error:\n"
-                f"{pformat(json.dumps(dataplex_entities_match))}\n\n"
+                f"{pformat(json.dumps([entity.to_dict() for entity in dataplex_entities_match]))}\n\n"
                 f"Parsed entity_uri configs:\n"
                 f"{pformat(entity_uri.to_dict())}\n\n"
             )
             return False
+        if len(dataplex_entities_match) > 1:
+            logger.warning(
+                "Unexpected number of Dataplex entities found for entity_uri \n"
+                f"${entity_uri.complete_uri_string}:\n"
+                f"${pformat([entity.to_dict() for entity in dataplex_entities_match])}. \n"
+                f"Only one match is expected."
+            )
+            return False
         else:
             dataplex_entity = dataplex_entities_match[0]
             clouddq_entity = dq_entity.DqEntity.from_dataplex_entity(

clouddq/integration/dataplex/clouddq_dataplex.py

@@ -307,7 +307,7 @@ def list_dataplex_entities(
         if prefix:
             params.update({"filter": f"id=starts_with({prefix})"})
         if data_path:
-            params.update({"filter": f"data_path=starts_with({data_path})"})
+            params.update({"filter": f"data_path={data_path}"})
 
         response_dict = {}
         response = self._client.list_entities(

tests/unit/test_dataplex_entity.py

@@ -90,7 +90,10 @@ def test_validate_dataplex_entity_to_dict(self, mock_valid_dataplex_input):
         schema = {"fields": [{"name": "column1", "type": "STRING", "mode": "REQUIRED"},
                               {"name": "column2", "type": "STRING", "mode": "REQUIRED"},  # noqa: E127
                               {"name": "column3", "type": "STRING", "mode": "REQUIRED"},  # noqa: E127
-                              {"name": "column4", "type": "TIMESTAMP", "mode": "REQUIRED"}]}  # noqa: E127
+                              {"name": "column4", "type": "TIMESTAMP", "mode": "REQUIRED"}],
+                  "partitionFields": None,
+                  "partitionStyle": None
+                  }  # noqa: E127
 
         dataplex_entity_expected = {
             "name": name,
@@ -103,7 +106,7 @@ def test_validate_dataplex_entity_to_dict(self, mock_valid_dataplex_input):
             "dataPath": "projects/project-id/datasets/bigquery_dataset_id/tables/table_name",
             "system": "BIGQUERY",
             "format": {"format": "OTHER"},
-            "schema": DataplexEntitySchema.from_dict(entity_id=name, kwargs=schema),
+            "schema": schema,
             "project_id": "project-id",
             "location": "location-id",
             "lake": "lake_name",
@@ -122,7 +125,10 @@ def test_validate_dataplex_entity_to_dict_no_format(self, mock_valid_dataplex_in
         schema = {"fields": [{"name": "column1", "type": "STRING", "mode": "REQUIRED"},
                               {"name": "column2", "type": "STRING", "mode": "REQUIRED"},  # noqa: E127
                               {"name": "column3", "type": "STRING", "mode": "REQUIRED"},  # noqa: E127
-                              {"name": "column4", "type": "TIMESTAMP", "mode": "REQUIRED"}]}  # noqa: E127
+                              {"name": "column4", "type": "TIMESTAMP", "mode": "REQUIRED"}],
+                  "partitionFields": None,
+                  "partitionStyle": None
+                  }  # noqa: E127
 
         dataplex_entity_expected = {
             "name": name,
@@ -135,7 +141,7 @@ def test_validate_dataplex_entity_to_dict_no_format(self, mock_valid_dataplex_in
             "dataPath": "projects/project-id/datasets/bigquery_dataset_id/tables/table_name",
             "system": "BIGQUERY",
             "format": None,
-            "schema": DataplexEntitySchema.from_dict(entity_id=name, kwargs=schema),
+            "schema": schema,
             "project_id": "project-id",
             "location": "location-id",
             "lake": "lake_name",