def _check_if_column_should_be_regenerated()

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

• 20 lines of code
• 7 McCabe index (conditional complexity)

    def _check_if_column_should_be_regenerated(self, table_fqn,column_name):
        """Updates the draft description for a column from a BigQuery table in Dataplex.

        Args:
            table_fqn (str): The fully qualified name of the table (e.g., 'project.dataset.table')
            column_name (str): The name of the column to update
            description (str): The new draft description for the column

        Raises:
            Exception: If there is an error updating the column description in Dataplex
        """

        # Create a client
        client = self._cloud_clients[constants["CLIENTS"]["DATAPLEX_CATALOG"]]        

        project_id, dataset_id, table_id = self._split_table_fqn(table_fqn) 
    
        entry = dataplex_v1.Entry()
        entry.name = f"projects/{project_id}/locations/{self._get_dataset_location(table_fqn)}/entryGroups/@bigquery/entries/bigquery.googleapis.com/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}"
        #entry.aspects[f"""{project_id}.global.{constants["ASPECT_TEMPLATE"]["name"]}"""] = aspect
        aspect_types=[f"""projects/{self._project_id}/locations/global/aspectTypes/{constants["ASPECT_TEMPLATE"]["name"]}"""]

        # Check if the aspect already exists
        try:
            get_request=dataplex_v1.GetEntryRequest(name=entry.name,view=dataplex_v1.EntryView.CUSTOM,aspect_types=aspect_types)
            entry = client.get_entry(request=get_request)
        except Exception as e:
            logger.error(f"Exception: {e}.")
            raise e
        
        for i in entry.aspects:
            #logger.info(f"""i: {i} path: "{entry.aspects[i].path}" """)
            if i.endswith(f"""global.{constants["ASPECT_TEMPLATE"]["name"]}@Schema.{column_name}""") and entry.aspects[i].path==f"Schema.{column_name}" :                
                data_dict = entry.aspects[i].data
                if data_dict["to-be-regenerated"] == True:
                    return True
                else:
                    return False
        return False