def get_namespaces_by_checking_access_permission()

in src/hyperpod_cli/service/discover_namespaces.py [0:0]

• 29 lines of code
• 6 McCabe index (conditional complexity)

    def get_namespaces_by_checking_access_permission(
            self,
            namespaces,
            resource_attributes_template, 
            max_workers=10
        ):
        """
        Get the accessible namespaces by performing the SelfSubjectAccessReview. For each of the namespace,
        check if user has specified permission to it, if the answer is NO, the namespace will be skipped.

        Performing access check can take quite long if the number of namespaces is large. Thus the implementation
        leverages the multi-threading to ensure that multiple access check can be performed in parallel.
        """
        subject_access_review = SelfSubjectAccessReview()
        accessible_namespaces = list()
        resource_attributes = list()

        for namespace in namespaces:
            resource_attribute = copy.deepcopy(resource_attributes_template)
            resource_attribute.namespace = namespace
            resource_attributes.append(resource_attribute)

        # Multi-thread the self subject access review to improve the performance
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(
                    subject_access_review.self_subject_access_review, resource_attribute
                ) for resource_attribute in resource_attributes
            }
            
            for future in concurrent.futures.as_completed(futures):
                try:
                    response = future.result()
                    if response.status.allowed:
                        accessible_namespaces.append(
                            response.spec.resource_attributes.namespace
                        )
                except Exception as e:
                    raise(e)

        return accessible_namespaces