def create_catalog_tag()

in common/data_mesh/src/data_mesh_client.py [0:0]

• 27 lines of code
• 8 McCabe index (conditional complexity)

    def create_catalog_tag(
            self,
            project: str,
            tag_spec: dmt.CatalogTag,
            template_spec: dmt.CatalogTagTemplate,
            table_entry_name: str,
            column: Optional[str] = None,
            existing_catalog_tags: Optional[Dict[str, str]] = None) -> None:
        """Create the specified Catalog Tag.

        Args:
            project: String representing project to create resources in.
            tag_spec: data_mesh_types.CatalogTag that specifies how to create
                the Catalog Tag.
            template_spec: data_mesh_types.CatalogTagTemplate that defines the
                given tag schema.
            table_entry_name: String representing the Data Catalog table entry
                that the tag is annotating.
            column: String representing the name of the column if the tag is
                column level rather than asset level. Nested fields are
                specified with '.' separators, e.g. `struct.nested`.
            existing_catalog_tags: Dictionary of
                template_display_name: unique_name containing the existing tags.
                Catalog Tags can be keyed by the template display name because
                there cannot be more than one tag on a particular entry for the
                same template.
        """
        relevant_existing_catalog_tags = {}
        if existing_catalog_tags:
            for k, v in existing_catalog_tags.items():
                prefix, _, tag_name = k.partition(":")
                # Add relevant column tags.
                if column is not None and prefix == column:
                    relevant_existing_catalog_tags[tag_name] = v
                # Add relevant asset tags.
                elif column is None and prefix == _ASSET_TAG_PREFIX:
                    relevant_existing_catalog_tags[tag_name] = v

        if self._make_ready_to_create(tag_spec, relevant_existing_catalog_tags):
            return

        template_name = self._get_resource_name(dmt.CatalogTagTemplate,
                                                template_spec.display_name,
                                                project,
                                                self._existing_tag_templates)
        _ = self._create_resource(tag_spec,
                                  table_entry_name,
                                  template_spec=template_spec,
                                  template=template_name,
                                  column=column)