def _create_resource()

in src/common/data_mesh/src/data_mesh_client.py [0:0]
32 lines of code
6 McCabe index (conditional complexity)

    def _create_resource(self, resource_spec: Any, parent: str,
                         **kwargs: Any) -> str:
        """Returns the specified created resource unique name.

        Args:
            resource_spec: An instance of a data_mesh_types dataclass associated
                with a resource.
            parent: String representing the parent project or resource. This
                doesn't include the parent policy tag if that applies, which is
                a different arg.
            kwargs:
                DataPolicy: parent_policy
                PolicyTag: parent_policy (optional)
                CatalogTag: template_spec, template_name, column (optional)
        """
        resource_type = type(resource_spec)

        if resource_type == dmt.CatalogTagTemplate:
            resource = dmt_util.get_request_type(resource_spec)
            resource = self._catalog_client.create_tag_template(
                parent=parent,
                tag_template_id=resource_spec.display_name,
                tag_template=resource,
                retry=self._retry_options)
        elif resource_type == dmt.CatalogTag:
            resource = dmt_util.get_request_type(resource_spec, **kwargs)
            _ = self._catalog_client.create_tag(parent=parent,
                                                tag=resource,
                                                retry=self._retry_options)
        elif resource_type == dmt.PolicyTaxonomy:
            resource = dmt_util.get_request_type(resource_spec)
            resource = self._policy_tag_client.create_taxonomy(
                parent=parent, taxonomy=resource, retry=self._retry_options)
        elif resource_type == dmt.PolicyTag:
            resource = dmt_util.get_request_type(resource_spec, **kwargs)
            resource = self._policy_tag_client.create_policy_tag(
                parent=parent, policy_tag=resource, retry=self._retry_options)
        elif resource_type == dmt.DataPolicy:
            resource = dmt_util.get_request_type(resource_spec, **kwargs)
            resource = self._bq_datapolicy_client.create_data_policy(
                parent=parent, data_policy=resource, retry=self._retry_options)
        else:
            raise cortex_exc.NotImplementedCError(f"Type: {resource_type}")

        logging.info("Created %s: '%s'", resource_type.__name__,
                     resource_spec.display_name)
        return resource.name