def is_dir()

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

• 20 lines of code
• 5 McCabe index (conditional complexity)

    def is_dir(self, resource_path):
        with grpc.insecure_channel(self.mft_api_endpoint) as channel:
            child_path = self._get_child_path(resource_path)
            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=child_path,
                mftAuthorizationToken=self.auth_token,
            )
            try:
                stub.getDirectoryResourceMetadata(request)
                return True
            except Exception:
                # assume that it doesn't exist or isn't a file
                logger.warning(f"Could not get metadata for {child_path} on {self.resource_id}")
                return False