in tools/datacatalog-tagging-using-tag-templates/tagging/attach_delete_tag_to_table_column.py [0:0]
def delete_tag(config_table):
"""function for deletion of tags"""
try:
catalogclient = datacatalog_v1.DataCatalogClient()
bqclient = bigquery.Client()
job_config = bigquery.QueryJobConfig(use_query_cache=False)
query = f"""
SELECT id,projectname,datasetname,tablename,columnname,tagtemplate,level,tag
FROM {config_table} where upper(mode)="DELETE"
"""
results = bqclient.query(query, job_config=job_config).result()
if results.total_rows == 0:
print("No tags to delete")
else:
for row in results:
id1 = row["id"]
if row["level"] == "table":
search_results = tag_util.search_entries(
row["projectname"],
row["projectname"]
+ "."
+ row["datasetname"]
+ "."
+ row["tablename"],
row["level"],
logs_enabled=True,
)
else:
search_results = tag_util.search_entries(
row["projectname"],
row["columnname"],
row["level"],
logs_enabled=True,
)
lst_tables = []
colname = row["columnname"]
tbl_name = row["tablename"]
if row["level"] == "column" and row["tablename"] == "":
print(
f"Deleting tags for column={colname} in all tables except:"
)
colquery = f"""
SELECT tablename
FROM {config_table} where
columnname = '{colname}' and tablename != ''
"""
col_results = bqclient.query(
colquery, job_config=job_config
).result()
if col_results.total_rows != 0:
for r in col_results:
lst_tables.append(r["tablename"])
print(lst_tables)
else:
print(
f"Deleting tags for column={colname} in specific table:{tbl_name}"
)
for search_result in search_results:
entry = search_result.relative_resource_name
result = catalogclient.list_tags(parent=entry)
for response in result:
tbl_str = search_result.linked_resource.split("/", 4)[-1]
tbl_name = tbl_str[tbl_str.rfind("/") + 1 :]
if row["level"] == "table":
if (
response.template.split("/")[-1] == row["tagtemplate"]
and response.column == ""
):
catalogclient.delete_tag(name=response.name)
print(
f"Tag deleted successfully for table : {tbl_name}"
)
elif row["level"] == "column" and row["tablename"] != "":
if (
response.template.split("/")[-1] == row["tagtemplate"]
and response.column == row["columnname"]
):
if row["tablename"] == tbl_name:
catalogclient.delete_tag(name=response.name)
print(
"Tag deleted successfully for table "
+ tbl_name
+ " for column : "
+ str(response.column)
)
elif row["level"] == "column" and row["tablename"] == "":
if (
response.template.split("/")[-1] == row["tagtemplate"]
and response.column == row["columnname"]
):
if tbl_name not in lst_tables:
catalogclient.delete_tag(name=response.name)
print(
"Tag deleted successfully for table "
+ tbl_name
+ " for column : "
+ str(response.column)
)
update_query = f"""UPDATE {config_table}
set tagflag=false,activeflag=false,mode='DELETED' WHERE id = '{id1}'
AND activeflag is true and mode='DELETE'"""
query_job = bqclient.query(update_query)
query_job.result()
except (exceptions.BadRequest,exceptions.PermissionDenied,exceptions.NotFound) as exception:
print(f"A {type(exception).__name__} has occurred.")
print(f"Caught error: {repr(exception)}")