Client examples

Python is an excellent choice for writing a client that consumes the Storage API. This document provides examples for both REST and gRPC APIs.

Discovery

The Storage API endpoint is not auto-discovered. To connect, you need the Storage API endpoint address. An example REST endpoint on localhost could be http://127.0.0.1:8011/fileobject, and a valid gRPC connection string could be 127.0.0.1:50051.

REST

Client library selection

The REST interface is a pure HTTP interface. Any library supporting basic web operations will work. We’ll use the Python requests module for this text, a popular library with comprehensive documentation.

It can be installed for example with pip:

pip install requests

Initial call

Here’s the minimal code required for a successful roundtrip to the Storage API:

def stat(example_server_address: str, resource_address: str):
    api_url = example_server_address + f"/by-address/{urllib.parse.quote_plus(resource_address)}"
    response = requests.head(api_url)
    if response.status_code == 204:
        result = json.loads(response.headers["x-nvidia-omniverse-storage-metadata"])
        print(result)
    else:
        print(f"Error calling stat, result code is {response.status_code}")

This code issues a single HTTP HEAD command on the calculated address. A successful stat returns the HTTP code 204, indicating success without body. The metadata is provided as a JSON struct within the x-nvidia-omniverse-storage-metadata response header, which is parsed and printed to stdout.

Data download

The Storage API service determines how data is downloaded. The simplest method is to stream the data using the HTTP response’s content body. A simple client can request the download method via a query parameter, and a well-behaved storage service will respect the preference given if the performance envelop permits it.

Here’s a working download loop:

def download_via_body(example_server_address: str, resource_address: str, destination_file_name: str):
    api_url = example_server_address + f"/by-address/{urllib.parse.quote_plus(resource_address)}?download_preference=body"
    response = requests.get(api_url, stream=True)
    if response.status_code == 200:
        with open(destination_file_name, "wb") as downloaded_file:
            for chunk in response.iter_content(chunk_size=None):
                downloaded_file.write(chunk)
        print(f"finished writing to file {destination_file_name}")
    else:
        print(f"Error downloading {resource_address}, result code is {response.status_code}")

This method is suitable for small files and simple clients. However, in a highly scalable environment, the Storage API service should not bear the load of all data downloads. In such cases, the service will likely return a redirect message with instructions for downloading data from the origin server.

Here’s how to implement this:

def download_via_redirect(example_server_address: str, resource_address: str, destination_file_name: str):
    api_url = example_server_address + f"/by-address/{urllib.parse.quote_plus(resource_address)}?download_preference=redirect"
    response = requests.get(api_url, stream=True)
    if response.status_code == 300:
        redirection_properties = json.loads(response.content)
        redirected_response = requests.get(
            redirection_properties["redirect_target_url"], headers=redirection_properties["additional_headers"], stream=True
        )
        if redirected_response.status_code == 200:
            with open(destination_file_name, "wb") as downloaded_file:
                for chunk in redirected_response.iter_content(chunk_size=None):
                    downloaded_file.write(chunk)
            print(f"finished writing to file {destination_file_name}")
        else:
            print(f"Error downloading {resource_address} after redirection, result code is {redirected_response.status_code}")
    else:
        print(f"Error downloading {resource_address}, result code is {response.status_code}")

Combining these elements, we get a complete flexible download client:

def download(example_server_address: str, resource_address: str, destination_file_name: str):
    api_url = example_server_address + f"/by-address/{urllib.parse.quote_plus(resource_address)}"
    response = requests.get(api_url, stream=True)
    if response.status_code == 200:
        with open(destination_file_name, "wb") as downloaded_file:
            for chunk in response.iter_content(chunk_size=None):
                downloaded_file.write(chunk)
        print(f"finished writing to file {destination_file_name}")
    elif response.status_code == 300:
        redirection_properties = json.loads(response.content)
        if redirection_properties["method"] == "get":
            redirected_response = requests.get(redirection_properties["redirect_target_url"], stream=True)
        elif redirection_properties["method"] == "post":
            redirected_response = requests.post(redirection_properties["redirect_target_url"], stream=True)
        else:
            raise RuntimeError(f"Download verb {redirection_properties['method']} not implemented")

        if redirected_response.status_code == 200:
            with open(destination_file_name, "wb") as downloaded_file:
                for chunk in redirected_response.iter_content(chunk_size=None):
                    downloaded_file.write(chunk)
            print(f"finished writing to file {destination_file_name}")
        else:
            print(f"Error downloading {resource_address} after redirection, result code is {redirected_response.status_code}")
    else:
        print(f"Error downloading {resource_address}, result code is {response.status_code}")

Note it is the client’s responsibility to optimize the redirected HTTP download, e.g. using proper multipart operations.

Data upload

HTTP provides some built-in large file transmission control mechanisms, so the mainstream HTTP client libraries won’t require additional actions to perform an upload.

To write a file object, call the PUT /fileobject/by-address/<address> endpoint providing the contents of the file object as the payload and specifying the length of the file object in the data_object_size query parameter. You can optionally provide an upload_preference query parameter, specifying the preferred content upload method, just like on download. Upon processing the request, the service will respond with either HTTP 201 or HTTP 300 status code, identifying either a successful write or the following write instruction respectively:

def upload(address: str, upload_preference: str | None, content: IO[bytes]):
    """Upload a file object at the specified address using REST interface."""

    content_length = get_content_length(content)
    if not upload_preference:
        upload_preference = _get_upload_preference(address, content_length)

    write_response = requests.put(
        f"{BASE_URL}/fileobject/by-address/{quote_plus(address)}",
        data=content,
        headers={
            "Expect": "100-continue",
        },
        params={
            key: value
            for key, value in (
                ("data_object_size", content_length),
                ("upload_preference", upload_preference),
            )
            if value is not None
        },
    )
    write_response.raise_for_status()

    if write_response.status_code == 201:
        return

    if write_response.status_code == 300:
        content.seek(0)

        write_response_json: WriteResponse = write_response.json()
        if (redirect := write_response_json.get("redirect")) is not None:
            return _write_via_redirect(address, redirect, content)
        if (multipart := write_response_json.get("multipart")) is not None:
            return _write_via_multipart(address, multipart, content)

    raise Exception("PUT write got unexpected response.")

HTTP 300 should be handled manually, namely by introspecting the contents of the response body. If the redirect response property is present and is not null, a redirect write must be performed:

def _write_via_redirect(address: str, params: WriteRedirectParams, content: IO[bytes]):
    response = requests.request(
        url=params["redirect_target_url"],
        method=params["method"].upper(),
        headers={h["name"]: h["value"] for h in params["additional_headers"]},
        data=content,
    )
    response.raise_for_status()

    # Optionally, obtain the resource information of the just uploaded file object
    # This is not required to store the data, but helpful to keep the reference to the stored object.
    requested_headers = {name.lower() for name in params["completion_header_names"]}
    completion = requests.post(
        f"{BASE_URL}/fileobject/by-address/{quote_plus(address)}/redirect/complete",
        json={
            "additional_headers": [
                {"name": response_key, "value": response.headers[response_key]}
                for response_key in response.headers
                if response_key.lower() in requested_headers
            ]
        },
    )
    completion.raise_for_status()

Otherwise, if the response contains a non-null multipart response property, the contents of the file object must be cut into several chunks, not exceeding the multipart upload limits and uploaded as follows:

def _write_via_multipart(address: str, params: MultipartUploadParams, content: IO[bytes]):
    # Build a list of redirect URLs to upload the parts to. The first we got from the CreateMultipartUploadResponse
    redirects = [params["first_part_write_redirect"]]

    # Calculate the number of parts, and retrieve the pre-signed URLs from the service to use for individual part upload
    total_part_count, part_size, content_length = part_count_for_multipart_upload(
        content, params["minimum_size_per_part"], params["maximum_size_per_part"]
    )
    if total_part_count > 1:
        response = requests.post(
            f"{BASE_URL}/fileobject/by-address/{quote_plus(address)}/multipart/prepare",
            json={
                "upload_id": params["upload_id"],
                "part_number": 1,
                "part_count": total_part_count - 1,
            },
        )
        if response.status_code != 200:
            print(response)
            raise Exception("POST multipart prepare got unexpected response.")

        redirects.extend(response.json()["part_write_redirects"])

    response = requests.post(
        f"{BASE_URL}/fileobject/by-address/{quote_plus(address)}/multipart/complete",
        json={
            "upload_id": params["upload_id"],
            "parts": [
                _write_part(part_number, redirects[part_number], part)
                for part_number, part in enumerate(
                    split_contents_for_multipart_upload(
                        content,
                        max_part_size=params.get("maximum_size_per_part"),
                        min_part_size=params.get("minimum_size_per_part"),
                        max_parts=params.get("maximum_parts_number"),
                    ),
                )
            ],
        },
    )
    if response.status_code != 200:
        raise Exception("Post multipart complete got unexpected response.")


def _write_part(part_number: int, redirect, content: SupportsRead) -> CompletedUploadPart:
    return {
        "part_number": part_number,
        "additional_headers": [
            {
                "name": name,
                "value": value,
            }
            for name, value in upload_part(
                redirect["redirect_target_url"],
                redirect["method"],
                content,
                upload_headers=dict([(h["name"], h["value"]) for h in redirect["additional_headers"]]),
                return_headers=redirect["completion_header_names"],
            )
        ],
    }

Upload preference selection

A client may prepare itself in advance to a certain write workflow, deducing the upload preference from the response of the GET /fileobject/by-address/<address>/upload-options endpoint, based on the size of the written file object:

def _get_upload_preference(address: str, size: int) -> str | None:
    response = requests.get(f"{BASE_URL}/fileobject/upload-options/by-address/{quote_plus(address)}")
    response.raise_for_status()

    for interval in response.json()["write_type_intervals"]:
        min_size = interval.get("minimum_data_object_size")
        max_size = interval.get("maximum_data_object_size")
        if min_size and min_size > size:
            continue

        if max_size and max_size < size:
            continue

        return interval["preferred_upload_method"]

    return None

In such a case, when a matching upload_preference is passed in the query parameters of the following PUT /fileobject/by-address/<address> request, it’s guaranteed that the service will select this exact upload method.

Listing folders

REST interface exposes a token-based pagination “list” and “list with stat” endpoints allowing to list the contents of folders in one or more chunks:

def list_file_folders(resource_address: str, stat: bool = False, max_page_size: int | None = None):
    with Session() as session:
        if stat:
            for list_stat_response in list_stat_file_folders_paginated(session, resource_address, max_page_size):
                handle_list_stat_file_folders_response(list_stat_response)
        else:
            for list_response in list_file_folders_paginated(session, resource_address, max_page_size):
                handle_list_file_folders_response(list_response)


def fetch_paginated(session: Session, url: str, max_page_size: int | None) -> Iterable[dict]:
    query_params = {}
    if max_page_size:
        query_params["max_page_size"] = max_page_size

    while True:
        response = session.get(url, params=query_params)
        response.raise_for_status()

        response_json = response.json()
        yield response_json

        if continuation_handle := response_json.get("next_continuation_handle"):
            query_params["continuation_handle"] = continuation_handle
        else:
            break


def list_file_folders_paginated(
    session: Session,
    address: str,
    max_page_size: int | None,
) -> Iterable[ListFileFoldersResponse]:
    return fetch_paginated(
        session,
        f"{REST_SERVICE_ADDRESS}/filefolder/list/{quote_plus(address)}",
        max_page_size,
    )  # type: ignore[return-value]


def list_stat_file_folders_paginated(
    session: Session,
    address: str,
    max_page_size: int | None,
) -> Iterable[ListStatFileFoldersResponse]:
    return fetch_paginated(
        session,
        f"{REST_SERVICE_ADDRESS}/filefolder/liststat/{quote_plus(address)}",
        max_page_size,
    )  # type: ignore[return-value]

Deleting file

To delete a file object, call the DELETE /fileobject/by-address/<address> endpoint. The endpoint always returns a 204 status code, regardless of whether the object exists.

def delete(
    example_server_address: str,
    resource_address: str,
):
    api_url = example_server_address + f"/by-address/{urllib.parse.quote_plus(resource_address)}"
    response = requests.delete(api_url)
    if response.status_code == 204:
        print(f"Done deleting via rest, file {resource_address}!")
    else:
        print(f"Error deleting {resource_address}, result code is {response.status_code}")
        print(response.json())
        raise Exception("Could not delete")

GRPC

Client library selection

Like REST, multiple different client libraries for Python are available to implement the gRPC client code. We’ll use the official grpcio library for our examples. Install it with

pip install grpcio grpcio-tools

The grpcio-tools package includes the protobuf compiler needed to generate the service stub files.

Python language bindings generation

Assuming the storage proto files are in a proto subdirectory, create the language bindings in the current directory with the following command:

python -m grpc_tools.protoc --python_out=. --pyi_out=. --grpc_python_out=. --proto_path=proto proto\nvidia\omniverse\storage\fileobject\v1beta\fileobject.proto proto\nvidia\omniverse\storage\fileobject\v1beta\fileobject_service.proto proto\nvidia\omniverse\storage\capabilities\v1beta\capabilities.proto proto\nvidia\omniverse\storage\filefolder\v1beta\filefolder_service.proto

This command creates the required Python files in the nvidia subdirectory, from which they can be imported. Make sure to adjust the paths and number of files matching the version of the storage API definitions, new files might be added and paths might change.

Simple stat

With the generated files, calling stat is even simpler than the REST version:

from nvidia.omniverse.storage.fileobject.v1alpha.fileobject_service_pb2_grpc import (
    FileObjectServiceStub,
)


def stat(grpc_address: str, resource_address: str):
    with grpc.insecure_channel(grpc_address) as channel:
        storage_api_server = FileObjectServiceStub(channel)
        try:
            response = storage_api_server.Stat(StatRequest(resource_address=resource_address))
            print(f"{resource_address}: size {response.resource_info.metadata.data_object_size}")
        except grpc.RpcError as e:
            print(f"Failure to stat {resource_address}: {str(e)}")

A connection to the gRPC server is established once via a channel context manager. The created service stub connecting via that channel can be reused for multiple calls.

Data download

Downloading data from the Storage API is straightforward. We just need to loop over the replies:

from nvidia.omniverse.storage.fileobject.v1alpha.fileobject_service_pb2_grpc import (
    FileObjectServiceStub,
)


def download(
    grpc_address: str,
    resource_address: str,
    destination_file_name: str,
    download_preference: DownloadPreference,
):
    with grpc.insecure_channel(grpc_address) as channel:
        storage_api_server = FileObjectServiceStub(channel)
        try:
            with open(destination_file_name, "wb") as destination_file:
                for reply in storage_api_server.ReadFromAddress(
                    ReadFromAddressRequest(resource_address=resource_address, download_preference=download_preference)
                ):
                    if reply.HasField("resource_info"):
                        print(f"Found {resource_address}, downloading {reply.resource_info.metadata.data_object_size} bytes")
                    elif reply.HasField("chunk"):
                        destination_file.write(reply.chunk.chunk)
                    elif reply.HasField("redirect"):
                        redirected_response = requests.get(reply.redirect.redirect_target_url, stream=True)
                        if redirected_response.ok:
                            for chunk in redirected_response.iter_content(chunk_size=None):
                                destination_file.write(chunk)
                            print(f"Done downloading from webserver file {destination_file_name} written!")
                            return
                        else:
                            raise Exception(f"{reply.redirect.redirect_target_url} failed to download")
                    else:
                        raise Exception(f"Unexpected reply {reply}")
                print(f"Done downloading via grpc, file {destination_file_name} written!")
        except grpc.RpcError as e:
            print(f"Failure to download {resource_address}: {str(e)}")

In this case, we handle the download by iterating over the response stream from the ReadFromAddress method. This method returns first the metadata and then either a stream of chunks, which we write into the destination file as they come in, or responds with redirection data similar to the REST redirect described above.

Data upload

In contrast to HTTP, gRPC requires a more explicit transmission management, to avoid traffic congestion and ensure optional performance of a client.

Write must be initiated by sending a WriteRequest, including at least the resource_address of the written file object and its size specified in the data_object_size field. A client may optionally provide an upload_preference parameter identifying a preferred file object upload method as before.

Upon receiving a WriteRequest, the service will respond with a control message, which can be either a WriteChunksAccepted message, identifying the willingness of the service to accept the file object contents in one or more following Chunk messages, or a WriteRedirect message, which contains an instruction to follow either an indirect or a multipart write flow.

On a successful write, the service will respond with a ResourceInfo message containing the metadata of the newly created data object:

def upload(channel: Channel, address: str, upload_preference: str | None, content: IO[bytes]):
    """Upload the contents of a file object at the specified address."""

    content_length = get_content_length(content)
    if upload_preference:
        chosen_upload_preference = _string_to_upload_preference(upload_preference)
    else:
        chosen_upload_preference = _get_upload_preference(channel, address, content_length)

    service = FileObjectServiceStub(channel)
    with _write_message_queue() as (requests, request_iterator):
        requests.put(
            WriteRequest(
                params=WriteParameters(
                    destination_resource_address=address,
                    data_object_size=content_length,
                    upload_preference=chosen_upload_preference,
                ),
            ),
        )

        write_responses: Iterator[WriteResponse] = service.Write(request_iterator)

        flow_control_message = next(write_responses)

        # A service may immediately respond with a "resource info" message, meaning that a 0-byte object write has
        # been performed immediately without the need for any further actions on the client end.
        if flow_control_message.HasField("resource_info"):
            return

        # If we receive a "write chunks accepted" message, stream the chunks and finally await for a "resource info"
        # response message.
        if flow_control_message.HasField("write_chunks_accepted"):
            for chunk in _slice_content(content):
                requests.put(WriteRequest(chunk=chunk))

            # End chunk transmission.
            requests.put(None)

            resource_info_response = next(write_responses)
            if resource_info_response.HasField("resource_info"):
                return

            raise Exception("Resource info message is expected.")

    # Write redirect and multipart upload methods don't operate over the same stream of Write requests, so we should
    # close the write stream.
    if flow_control_message.HasField("write_redirect"):
        _write_via_redirect(service, address, flow_control_message.write_redirect, content)
        return

    if flow_control_message.HasField("multipart_upload"):
        _write_via_multipart(service, address, flow_control_message.multipart_upload, content)
        return

    raise Exception("Unexpected flow control message.")

Redirect and multipart write routines are similar to those of the HTTP protocol.

def _write_via_redirect(service: FileObjectServiceStub, address: str, parameters: WriteRedirectProperties, content: IO[bytes]):
    response = requests.request(
        url=parameters.redirect_target_url,
        method=_upload_method_to_string(parameters.method),
        headers={header.name: header.value for header in parameters.additional_headers},
        data=content,
    )
    response.raise_for_status()

    # Optionally, obtain the resource information of the just uploaded file object
    # This is not required to store the data, but helpful to keep the reference to the stored object.
    requested_headers = set(parameters.completion_header_names)

    # Create lowercase mapping for case-insensitive header matching
    response_headers_lower = {key.lower(): key for key in response.headers}

    service.CompleteRedirectUpload(
        CompleteRedirectUploadRequest(
            destination_resource_address=address,
            additional_headers=[
                Header(name=x, value=response.headers[response_headers_lower[x.lower()]])
                for x in requested_headers
                if x.lower() in response_headers_lower
            ],
        )
    )

def _write_via_multipart(service: FileObjectServiceStub, address: str, parameters: CreateMultipartUploadResponse, content: IO[bytes]):
    # Build a list of redirect URLs to upload the parts to. The first we got from the CreateMultipartUploadResponse
    redirects = [parameters.first_part_write_redirect]

    # Calculate the number of parts, and retrieve the pre-signed URLs from the service to use for individual part upload
    total_part_count, part_size, content_length = part_count_for_multipart_upload(
        content, parameters.minimum_size_per_part, parameters.maximum_size_per_part
    )
    if total_part_count > 1:
        response: UploadPartResponse = service.UploadPart(
            UploadPartRequest(
                upload_id=parameters.upload_id,
                destination_resource_address=address,
                part_number=1,  # Part number 0 has already been delivered by the first CreateMultipartUploadReponse
                part_count=total_part_count - 1,
            ),
        )
        redirects.extend(response.part_write_redirects)

    completed_parts = []
    for part_number, part in enumerate(
        split_contents_for_multipart_upload(
            content,
            min_part_size=parameters.minimum_size_per_part,
            max_part_size=parameters.maximum_size_per_part,
            max_parts=parameters.maximum_parts_number,
        )
    ):
        completed_parts.append(_write_part(redirects[part_number], part_number, part))

    service.CompleteMultipartUpload(
        CompleteMultipartUploadRequest(
            upload_id=parameters.upload_id,
            destination_resource_address=address,
            parts=completed_parts,
        ),
    )


def _write_part(redirect: WriteRedirectProperties, part: int, content: SupportsRead) -> CompletedUploadPart:
    return CompletedUploadPart(
        part_number=part,
        headers=[
            Header(name=name, value=value)
            for name, value in upload_part(
                url=redirect.redirect_target_url,
                method=_upload_method_to_string(redirect.method),
                upload_headers=dict([(header.name, header.value) for header in redirect.additional_headers]),
                return_headers=[name for name in redirect.completion_header_names],
                content=content,
            )
        ],
    )

Opportunistic writes

In some cases, clients can use “opportunistic writes”, that is such write operations, where Chunk messages are streamed without awaiting for the WriteChunksAccepted response from the service. While this can be beneficial performance-wise for small uploads, for any “substantially large” file objects, clients are advised to call the FetchWriteTypeInfo in advance to familiarize themselves with the method upload limits, as otherwise a rejected sequence of Chunk messages may cause traffic congestion.

Upload preference selection

In a similar way to the REST API, gRPC API provides means of deducing the upload method selected by the storage service, namely by introspecting the response of the FetchWriteTypeInfo method call:

def _get_upload_preference(channel: Channel, resource_address: str, size: int) -> UploadPreference:
    client = FileObjectServiceStub(channel)
    response = client.FetchWriteTypeInfo(FetchWriteTypeInfoRequest(destination_resource_address=resource_address))

    for interval in response.write_type_intervals:
        if interval.minimum_data_object_size and interval.minimum_data_object_size > size:
            continue

        if interval.maximum_data_object_size and interval.maximum_data_object_size < size:
            continue

        return interval.preferred_upload_method

    return UploadPreference.UPLOAD_PREFERENCE_UNSPECIFIED

Listing folders

To perform file and folder listings, use the List operation provided by the FileFolderService API, which returns a stream of file and folder entries. For clients requiring additional metadata for file entries, the ListStat operation should be employed as follows:

def list_file_folders(channel: Channel, resource_address: str, stat: bool = False):
    client = FileFolderServiceStub(channel)

    if stat:
        for list_stat_message in client.ListStat(ListStatRequest(folder=FolderAddress(uri=resource_address))):
            handle_list_stat_message(list_stat_message)
    else:
        for list_message in client.List(ListRequest(folder=FolderAddress(uri=resource_address))):
            handle_list_message(list_message)

Depending on the storage backend’s capabilities, ListStat might be significantly slower than List or there might be no performance difference at all.

Deleting file

To delete a file object, call the Delete operation provided by the FileFolderService API. This operation is idempotent, meaning it returns an OK response even if the specified file object does not exist. Additionally, it does not return any content in the response.

from nvidia.omniverse.storage.fileobject.v1alpha.fileobject_service_pb2_grpc import (
    FileObjectServiceStub,
)


def delete(
    grpc_address: str,
    resource_address: str,
):
    with grpc.insecure_channel(grpc_address) as channel:
        storage_api_server = FileObjectServiceStub(channel)
        try:
            reply = storage_api_server.Delete(DeleteRequest(resource_address=resource_address))
            print(f"Done deleting via grpc, file {resource_address}!")
        except grpc.RpcError as e:
            print(f"Failure to delete {resource_address}: {str(e)}")