Forward Compatibility

Forward Compatibility¶

AWS keeps adding enum values, union members and event-stream events to existing APIs. A client generated before such a change keeps working: it doesn't fail on a name it doesn't know, it hands it to you.

Enum values are typed as Literal[...], but they are plain strings at runtime. A value newer than your client comes back as the string AWS sent, so match enums with a fallback branch. You can send a new value too. The type checker doesn't know about it, so cast it:

from typing import cast

from capo_s3 import AsyncS3Client
from capo_s3.types.storage_class import StorageClass

async with AsyncS3Client() as s3:
    await s3.put_object("bucket_name", "key", body=b"data", storage_class=cast(StorageClass, "NEW_STORAGE_CLASS"))

A union variant your client doesn't know comes back as one more variant, keyed by $unknown: {"$unknown": {"name": ...}}, where name is the variant's name on the wire, or "" if the service sent no variant. No Smithy member name contains $, so the key never clashes with a real one, and you check for it like any other variant:

from capo_bedrock_runtime import AsyncBedrockRuntimeClient

async with AsyncBedrockRuntimeClient() as bedrock:
    response = await bedrock.converse(
        "amazon.nova-lite-v1:0",
        messages=[{"role": "user", "content": [{"text": "Hello"}]}],
    )

output = response["output"]
if "message" in output:
    for block in output["message"]["content"]:
        if "text" in block:
            print(block["text"])
        elif "$unknown" in block:
            print(f"skipping a {block['$unknown']['name']} block, newer than this client")

The type checker narrows on "$unknown" in block as it does on "text" in block, so ending a chain of checks with typing.assert_never(block) tells you which variants a new client version added. An unknown variant carries no data and can't be sent back: passing one as input raises SerializationError.

A response that leaves out a required member doesn't fail either. The client fills in the member's @default if it has one, and otherwise an empty value for its type: 0, False, "", b"", [], {}, None for documents, the Unix epoch for timestamps, and "" (or 0) for enums. A missing union comes back as {"$unknown": {"name": ""}}. A missing required structure still raises DeserializationError.

In an event stream, an event your client doesn't know comes back the same way, and the stream goes on. Next to the event's name (its :event-type), it keeps the raw message, with its headers and undecoded payload:

from capo_s3 import AsyncS3Client

async with AsyncS3Client() as s3:
    async with s3.select_object_content(
        "bucket_name",
        "key",
        expression="SELECT * FROM S3Object",
        expression_type="SQL",
        input_serialization={"csv": {}},
        output_serialization={"csv": {}},
    ) as response:
        async for event in response["payload"]:
            if "Records" in event:
                print(event["Records"]["payload"])
            elif "$unknown" in event:
                unknown = event["$unknown"]
                print(f"skipping {unknown['name']}: {unknown['message'].payload!r}")