def _get_table_scan_reference()

in src/package/dataplexutils/metadata/wizard.py [0:0]

• 19 lines of code
• 4 McCabe index (conditional complexity)

    def _get_table_scan_reference(self, table_fqn):
        """Retrieves data scan references for a BigQuery table.

        Args:
            table_fqn (str): The fully qualified name of the table
                (e.g., 'project.dataset.table')

        Returns:
            list: List of scan reference names associated with the table

        Raises:
            Exception: If there is an error retrieving scan references
        """
        try:
            scan_references = None
            scan_client = self._cloud_clients[
                constants["CLIENTS"]["DATAPLEX_DATA_SCAN"]
            ]
            logger.info(f"Getting table scan reference for table:{table_fqn}.")
            data_scans = scan_client.list_data_scans(
                parent=f"projects/{self._project_id}/locations/{self._dataplex_location}"
            )
            bq_resource_string = self._construct_bq_resource_string(table_fqn)
            scan_references = []
            for scan in data_scans:
                if scan.data.resource == bq_resource_string:
                    scan_references.append(scan.name)
            return scan_references
        except Exception as e:
            logger.error(f"Exception: {e}.")
            raise e