def _list_resources()

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

• 42 lines of code
• 15 McCabe index (conditional complexity)

    def _list_resources(self, resource_type: Type[object],
                        parent: str) -> Dict[str, str]:
        """Gets a dictionary of the existing resources of the given type.

        Args:
            resource_type: The resource type to list. This supports a subset of
                data_mesh_types dataclasses.
            parent: String that uniquely identifies a parent resource.
                CatalogTagTemplate: project ID
                CatalogTag: entry/asset name
                PolicyTaxonomy: project location path
                PolicyTag: taxonomy name
                DataPolicy: project location path
                Lake: project location path of lake region
                Zone: lake name
                asset: zone name

        Returns:
            A dict representing existing resources that maps resource display
            names to their uniquely identifing name.
        """
        list_pager = None

        # Catalog tag templates are searched for using syntax described below:
        # https://cloud.google.com/data-catalog/docs/how-to/search-reference
        if resource_type == dmt.CatalogTagTemplate:
            scope = datacatalog_v1.SearchCatalogRequest.Scope(
                include_project_ids=[parent])
            list_pager = self._catalog_client.search_catalog(
                scope=scope,
                query="type=tag_template",
                retry=self._retry_options)
        elif resource_type == dmt.CatalogTag:
            list_pager = self._catalog_client.list_tags(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.PolicyTaxonomy:
            list_pager = self._policy_tag_client.list_taxonomies(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.PolicyTag:
            list_pager = self._policy_tag_client.list_policy_tags(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.DataPolicy:
            list_pager = self._bq_datapolicy_client.list_data_policies(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.Lake:
            list_pager = self._dataplex_client.list_lakes(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.Zone:
            list_pager = self._dataplex_client.list_zones(
                parent=parent, retry=self._retry_options)
        elif resource_type == dmt.Asset:
            list_pager = self._dataplex_client.list_assets(
                parent=parent, retry=self._retry_options)
        else:
            raise cortex_exc.NotImplementedCError(f"Type: {resource_type}")

        resources = {}
        for resource in list_pager:
            # Catalog Tags need special handling because the returned results
            # can have multiple resources with the same display name for cases
            # where a single Tag Template is used at the asset and column level.
            if resource_type == dmt.CatalogTag:
                prefix = resource.column or _ASSET_TAG_PREFIX
                key = f"{prefix}:{dmt_util.get_display_name(resource)}"
            else:
                key = dmt_util.get_display_name(resource)

            resources[key] = dmt_util.get_unique_name(resource)
        return resources