feat(ingestion/transformer): create tag if not exist

metadata-ingestion/src/datahub/ingestion/graph/client.py

@@ -787,9 +787,11 @@ def get_aspect_counts(self, aspect: str, urn_like: Optional[str] = None) -> int:
 
     def execute_graphql(self, query: str, variables: Optional[Dict] = None) -> Dict:
         url = f"{self.config.server}/api/graphql"
+
         body: Dict = {
             "query": query,
         }
+
         if variables:
             body["variables"] = variables
 
@@ -1065,6 +1067,22 @@ def parse_sql_lineage(
             default_schema=default_schema,
         )
 
+    def create_tag(self, tag_name: str) -> Dict[Any, Any]:
+        graph_query: str = """
+            mutation($tag_detail: CreateTagInput!) {
+                createTag(input: $tag_detail)
+            }
+        """
+
+        variables = {
+            "tag_detail": {"name": tag_name},
+        }
+
+        return self.execute_graphql(
+            query=graph_query,
+            variables=variables,
+        )
+
     def close(self) -> None:
         self._make_schema_resolver.cache_clear()
         super().close()

metadata-ingestion/src/datahub/ingestion/transformer/add_dataset_tags.py

@@ -1,3 +1,4 @@
+import logging
 from typing import Callable, List, Optional, cast
 
 from datahub.configuration.common import (
@@ -9,6 +10,9 @@
 from datahub.ingestion.api.common import PipelineContext
 from datahub.ingestion.transformer.dataset_transformer import DatasetTagsTransformer
 from datahub.metadata.schema_classes import GlobalTagsClass, TagAssociationClass
+from datahub.utilities.urns.tag_urn import TagUrn
+
+logger = logging.getLogger(__name__)
 
 
 class AddDatasetTagsConfig(TransformerSemanticsConfigModel):
@@ -33,6 +37,26 @@ def create(cls, config_dict: dict, ctx: PipelineContext) -> "AddDatasetTags":
         config = AddDatasetTagsConfig.parse_obj(config_dict)
         return cls(config, ctx)
 
+    def create_tag_if_not_exist(
+        self, tag_associations: List[TagAssociationClass]
+    ) -> None:
+        if self.ctx.graph is None:
+            logger.debug("graph instance is None. Skip tag creation")
+            return  # graph instance in not available
+
+        for tag_association in tag_associations:
+            if self.ctx.graph.exists(tag_association.tag):
+                continue
+
+            ids: List[str] = TagUrn.create_from_string(
+                tag_association.tag
+            ).get_entity_id()
+
+            assert len(ids) == 1, "Invalid Tag Urn"
+
+            response: dict = self.ctx.graph.create_tag(tag_name=ids[0])
+            logger.debug(f"Tag creation response: {response}")
+
     def transform_aspect(
         self, entity_urn: str, aspect_name: str, aspect: Optional[Aspect]
     ) -> Optional[Aspect]:
@@ -45,6 +69,9 @@ def transform_aspect(
         tags_to_add = self.config.get_tags_to_add(entity_urn)
         if tags_to_add is not None:
             out_global_tags_aspect.tags.extend(tags_to_add)
+            self.create_tag_if_not_exist(
+                tag_associations=out_global_tags_aspect.tags,
+            )
 
         return self.get_result_semantics(
             self.config, self.ctx.graph, entity_urn, out_global_tags_aspect