def exists()

in airavata_django_portal_sdk/user_storage/backends/mft_provider.py [0:0]

• 26 lines of code
• 10 McCabe index (conditional complexity)

    def exists(self, resource_path):
        with grpc.insecure_channel(self.mft_api_endpoint) as channel:
            child_path = self._get_child_path(resource_path)
            # TODO: is this still needed?
            # parent_path, child_path = os.path.split(f"/tmp/{resource_path}".rstrip("/"))
            # get metadata for the parent path
            if child_path is not None:
                parent_path, child_path = os.path.split(child_path)
            else:
                parent_path = None
            stub = MFTApi_pb2_grpc.MFTApiServiceStub(channel)
            # Get metadata for parent directory and see if child_path exists
            request = MFTApi_pb2.FetchResourceMetadataRequest(
                # resourceId="remote-ssh-dir-resource",
                resourceId=self.resource_id,
                resourceType="SCP",
                # resourceToken="local-ssh-cred",
                resourceToken=self.resource_token,
                resourceBackend="FILE",
                resourceCredentialBackend="FILE",
                targetAgentId="agent0",
                childPath=parent_path,
                mftAuthorizationToken=self.auth_token,
            )
            try:
                response = stub.getDirectoryResourceMetadata(request)
            except Exception:
                # Could not find the parent path, so apparently doesn't exist
                logger.warning(f"Could not get metadata for {parent_path} on {self.resource_id}")
                return False
            # if not child_path, then return True since the response was
            # successful and we just need to confirm the existence of the root dir
            if child_path is None:
                return True
            return child_path in map(lambda f: f.friendlyName, list(response.directories) + list(response.files))