smithy-core

HostType

Bases: Enum

Enumeration of possible host types.

Source code in packages/smithy-core/src/smithy_core/__init__.py

class HostType(Enum):
    """Enumeration of possible host types."""

    IPv6 = "IPv6"
    """Host is an IPv6 address."""

    IPv4 = "IPv4"
    """Host is an IPv4 address."""

    DOMAIN = "DOMAIN"
    """Host type is a domain name."""

    UNKNOWN = "UNKNOWN"
    """Host type is unknown."""

DOMAIN = 'DOMAIN' class-attribute instance-attribute

Host type is a domain name.

IPv4 = 'IPv4' class-attribute instance-attribute

Host is an IPv4 address.

IPv6 = 'IPv6' class-attribute instance-attribute

Host is an IPv6 address.

UNKNOWN = 'UNKNOWN' class-attribute instance-attribute

Host type is unknown.

URI dataclass

Bases: URI

Universal Resource Identifier, target location for a :py:class:HTTPRequest.

Source code in packages/smithy-core/src/smithy_core/__init__.py

@dataclass(kw_only=True, frozen=True)
class URI(interfaces.URI):
    """Universal Resource Identifier, target location for a :py:class:`HTTPRequest`."""

    scheme: str = "https"
    """For example ``http`` or ``https``."""

    username: str | None = None
    """Username part of the userinfo URI component."""

    password: str | None = None
    """Password part of the userinfo URI component."""

    host: str
    """The hostname, for example ``amazonaws.com``."""

    port: int | None = None
    """An explicit port number."""

    path: str | None = None
    """Path component of the URI."""

    query: str | None = None
    """Query component of the URI as string."""

    fragment: str | None = None
    """Part of the URI specification, but may not be transmitted by a client."""

    def __post_init__(self) -> None:
        """Validate host component."""
        if not rfc3986.HOST_MATCHER.match(self.host) and not rfc3986.IPv6_MATCHER.match(
            f"[{self.host}]"
        ):
            raise SmithyError(f"Invalid host: {self.host}")

    @property
    def netloc(self) -> str:
        """Construct netloc string in format ``{username}:{password}@{host}:{port}``

        ``username``, ``password``, and ``port`` are only included if set. ``password``
        is ignored, unless ``username`` is also set. Add square brackets around the host
        if it is a valid IPv6 endpoint URI per :rfc:`3986#section-3.2.2`.
        """
        return self._netloc

    # cached_property does NOT behave like property, it actually allows for setting.
    # Therefore we need a layer of indirection.
    @cached_property
    def _netloc(self) -> str:
        if self.username is not None:
            password = "" if self.password is None else f":{self.password}"
            userinfo = f"{self.username}{password}@"
        else:
            userinfo = ""

        if self.port is not None:
            port = f":{self.port}"
        else:
            port = ""

        if self.host_type == HostType.IPv6:
            host = f"[{self.host}]"
        else:
            host = self.host

        return f"{userinfo}{host}{port}"

    @property
    def host_type(self) -> HostType:
        """Return the type of host."""
        return self._host_type

    @cached_property
    def _host_type(self) -> HostType:
        if rfc3986.IPv6_MATCHER.match(f"[{self.host}]"):
            return HostType.IPv6
        if rfc3986.IPv4_MATCHER.match(self.host):
            return HostType.IPv4
        if rfc3986.HOST_MATCHER.match(self.host):
            return HostType.DOMAIN
        return HostType.UNKNOWN

    def build(self) -> str:
        """Construct URI string representation.

        Validate host. Returns a string of the form
        ``{scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}``
        """
        components = (
            self.scheme,
            self.netloc,
            self.path or "",
            "",  # params
            self.query,
            self.fragment,
        )
        return urlunparse(components)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, URI):
            return False
        return (
            self.scheme == other.scheme
            and self.host == other.host
            and self.port == other.port
            and self.path == other.path
            and self.query == other.query
            and self.username == other.username
            and self.password == other.password
            and self.fragment == other.fragment
        )

fragment = None class-attribute instance-attribute

Part of the URI specification, but may not be transmitted by a client.

host instance-attribute

The hostname, for example amazonaws.com.

host_type property

Return the type of host.

netloc property

Construct netloc string in format {username}:{password}@{host}:{port}

username, password, and port are only included if set. password is ignored, unless username is also set. Add square brackets around the host if it is a valid IPv6 endpoint URI per :rfc:3986#section-3.2.2.

password = None class-attribute instance-attribute

Password part of the userinfo URI component.

path = None class-attribute instance-attribute

Path component of the URI.

port = None class-attribute instance-attribute

An explicit port number.

query = None class-attribute instance-attribute

Query component of the URI as string.

scheme = 'https' class-attribute instance-attribute

For example http or https.

username = None class-attribute instance-attribute

Username part of the userinfo URI component.

__post_init__()

Validate host component.

Source code in packages/smithy-core/src/smithy_core/__init__.py

def __post_init__(self) -> None:
    """Validate host component."""
    if not rfc3986.HOST_MATCHER.match(self.host) and not rfc3986.IPv6_MATCHER.match(
        f"[{self.host}]"
    ):
        raise SmithyError(f"Invalid host: {self.host}")

build()

Construct URI string representation.

Validate host. Returns a string of the form {scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}

Source code in packages/smithy-core/src/smithy_core/__init__.py

def build(self) -> str:
    """Construct URI string representation.

    Validate host. Returns a string of the form
    ``{scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}``
    """
    components = (
        self.scheme,
        self.netloc,
        self.path or "",
        "",  # params
        self.query,
        self.fragment,
    )
    return urlunparse(components)

aio

client

ClientCall dataclass

A data class containing all the initial information about an operation invocation.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

@dataclass(kw_only=True, frozen=True)
class ClientCall[I: SerializeableShape, O: DeserializeableShape]:
    """A data class containing all the initial information about an operation
    invocation."""

    input: I
    """The input of the operation."""

    operation: APIOperation[I, O] = field(repr=False)
    """The schema of the operation."""

    context: TypedProperties
    """The initial context of the operation."""

    interceptor: Interceptor[I, O, Any, Any]
    """The interceptor to use in the course of the operation invocation.

    This SHOULD be an InterceptorChain.
    """

    auth_scheme_resolver: AuthSchemeResolver
    """The auth scheme resolver for the operation."""

    supported_auth_schemes: dict[ShapeID, AuthScheme[Any, Any, Any, Any]]
    """The supported auth schemes for the operation."""

    endpoint_resolver: EndpointResolver
    """The endpoint resolver for the operation."""

    retry_strategy: RetryStrategy
    """The retry strategy to use for the operation."""

    retry_scope: str | None = None
    """The retry scope for the operation."""

    def retryable(self) -> bool:
        # TODO: check to see if the stream is seekable
        return self.operation.input_stream_member is None

auth_scheme_resolver instance-attribute

The auth scheme resolver for the operation.

context instance-attribute

The initial context of the operation.

endpoint_resolver instance-attribute

The endpoint resolver for the operation.

input instance-attribute

The input of the operation.

interceptor instance-attribute

The interceptor to use in the course of the operation invocation.

This SHOULD be an InterceptorChain.

operation = field(repr=False) class-attribute instance-attribute

The schema of the operation.

retry_scope = None class-attribute instance-attribute

The retry scope for the operation.

retry_strategy instance-attribute

The retry strategy to use for the operation.

supported_auth_schemes instance-attribute

The supported auth schemes for the operation.

RequestPipeline

Invokes client operations asynchronously.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

class RequestPipeline[TRequest: Request, TResponse: Response]:
    """Invokes client operations asynchronously."""

    protocol: ClientProtocol[TRequest, TResponse]
    """The protocol to use to serialize the request and deserialize the response."""

    transport: ClientTransport[TRequest, TResponse]
    """The transport to use to send the request and receive the response (e.g. an HTTP
    Client)."""

    def __init__(
        self,
        protocol: ClientProtocol[TRequest, TResponse],
        transport: ClientTransport[TRequest, TResponse],
    ) -> None:
        self.protocol = protocol
        self.transport = transport

    async def __call__[I: SerializeableShape, O: DeserializeableShape](
        self, call: ClientCall[I, O], /
    ) -> O:
        """Invoke an operation asynchronously.

        :param call: The operation to invoke and associated context.
        """
        output, _ = await self._execute_request(call, None)
        return output

    async def input_stream[
        I: SerializeableShape,
        O: DeserializeableShape,
        E: SerializeableShape,
    ](
        self, call: ClientCall[I, O], event_type: "TypeForm[E]", /
    ) -> InputEventStream[E, O]:
        """Invoke an input stream operation asynchronously.

        :param call: The operation to invoke and associated context.
        :param event_type: The event type to send in the input stream.
        """
        request_future = Future[RequestContext[I, TRequest]]()
        output_future = asyncio.create_task(
            self._await_output(self._execute_request(call, request_future))
        )
        request_context = await request_future
        input_stream = self.protocol.create_event_publisher(
            operation=call.operation,
            request=request_context.transport_request,
            event_type=event_type,
            context=request_context.properties,
            auth_scheme=request_context.properties.get(AUTH_SCHEME),
        )
        return InputEventStream(input_stream=input_stream, output_future=output_future)

    async def _await_output[I: SerializeableShape, O: DeserializeableShape](
        self,
        execute_task: Awaitable[tuple[O, OutputContext[I, O, TRequest, TResponse]]],
    ) -> O:
        output, _ = await execute_task
        return output

    async def output_stream[
        I: SerializeableShape,
        O: DeserializeableShape,
        E: DeserializeableShape,
    ](
        self,
        call: ClientCall[I, O],
        event_type: "TypeForm[E]",
        event_deserializer: Callable[[ShapeDeserializer], E],
        /,
    ) -> OutputEventStream[E, O]:
        """Invoke an input stream operation asynchronously.

        :param call: The operation to invoke and associated context.
        :param event_type: The event type to receive in the output stream.
        :param event_deserializer: The method used to deserialize events.
        """
        output, output_context = await self._execute_request(call, None)
        output_stream = self.protocol.create_event_receiver(
            operation=call.operation,
            request=output_context.transport_request,
            response=output_context.transport_response,
            event_type=event_type,
            event_deserializer=event_deserializer,
            context=output_context.properties,
        )
        return OutputEventStream(output_stream=output_stream, output=output)

    async def duplex_stream[
        I: SerializeableShape,
        O: DeserializeableShape,
        IE: SerializeableShape,
        OE: DeserializeableShape,
    ](
        self,
        call: ClientCall[I, O],
        input_event_type: "TypeForm[IE]",
        output_event_type: "TypeForm[OE]",
        event_deserializer: Callable[[ShapeDeserializer], OE],
        /,
    ) -> DuplexEventStream[IE, OE, O]:
        """Invoke an input stream operation asynchronously.

        :param call: The operation to invoke and associated context.
        :param input_event_type: The event type to send in the input stream.
        :param output_event_type: The event type to receive in the output stream.
        :param event_deserializer: The method used to deserialize events.
        """
        request_future = Future[RequestContext[I, TRequest]]()
        execute_task = asyncio.create_task(self._execute_request(call, request_future))
        request_context = await request_future
        input_stream = self.protocol.create_event_publisher(
            operation=call.operation,
            request=request_context.transport_request,
            event_type=input_event_type,
            context=request_context.properties,
            auth_scheme=request_context.properties.get(AUTH_SCHEME),
        )
        output_future = asyncio.create_task(
            self._await_output_stream(
                call=call,
                execute_task=execute_task,
                output_event_type=output_event_type,
                event_deserializer=event_deserializer,
            )
        )
        return DuplexEventStream(input_stream=input_stream, output_future=output_future)

    async def _await_output_stream[
        I: SerializeableShape,
        O: DeserializeableShape,
        OE: DeserializeableShape,
    ](
        self,
        call: ClientCall[I, O],
        execute_task: Awaitable[tuple[O, OutputContext[I, O, TRequest, TResponse]]],
        output_event_type: "TypeForm[OE]",
        event_deserializer: Callable[[ShapeDeserializer], OE],
    ) -> tuple[O, EventReceiver[OE]]:
        output, output_context = await execute_task
        output_stream = self.protocol.create_event_receiver(
            operation=call.operation,
            request=output_context.transport_request,
            response=output_context.transport_response,
            event_type=output_event_type,
            event_deserializer=event_deserializer,
            context=output_context.properties,
        )
        return output, output_stream

    async def _execute_request[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        request_future: Future[RequestContext[I, TRequest]] | None,
    ) -> tuple[O, OutputContext[I, O, TRequest, TResponse]]:
        _LOGGER.debug(
            'Making request for operation "%s" with parameters: %s',
            call.operation.schema.id.name,
            call.input,
        )
        output_context = await self._handle_execution(call, request_future)
        output_context = self._finalize_execution(call, output_context)

        if isinstance(output_context.response, Exception):
            e = output_context.response
            if not isinstance(e, SmithyError):
                raise SmithyError(e) from e
            raise e

        return output_context.response, output_context  # type: ignore

    async def _handle_execution[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        request_future: Future[RequestContext[I, TRequest]] | None,
    ) -> OutputContext[I, O, TRequest | None, TResponse | None]:
        try:
            interceptor = call.interceptor

            input_context = InputContext(request=call.input, properties=call.context)
            interceptor.read_before_execution(input_context)

            input_context = replace(
                input_context,
                request=interceptor.modify_before_serialization(input_context),
            )

            interceptor.read_before_serialization(input_context)
            _LOGGER.debug("Serializing request for: %s", input_context.request)

            transport_request = self.protocol.serialize_request(
                operation=call.operation,
                input=call.input,
                endpoint=_UNRESOLVED,
                context=input_context.properties,
            )
            request_context = RequestContext(
                request=input_context.request,
                transport_request=transport_request,
                properties=input_context.properties,
            )

            _LOGGER.debug(
                "Serialization complete. Transport request: %s", transport_request
            )
        except Exception as e:
            return OutputContext(
                request=call.input,
                response=e,
                transport_request=None,
                transport_response=None,
                properties=call.context,
            )

        try:
            interceptor.read_after_serialization(request_context)
            request_context = replace(
                request_context,
                transport_request=interceptor.modify_before_retry_loop(request_context),
            )

            return await self._retry(call, request_context, request_future)
        except Exception as e:
            return OutputContext(
                request=request_context.request,
                response=e,
                transport_request=request_context.transport_request,
                transport_response=None,
                properties=request_context.properties,
            )

    async def _retry[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        request_context: RequestContext[I, TRequest],
        request_future: Future[RequestContext[I, TRequest]] | None,
    ) -> OutputContext[I, O, TRequest | None, TResponse | None]:
        if not call.retryable():
            return await self._handle_attempt(call, request_context, request_future)

        retry_strategy = call.retry_strategy
        retry_token = retry_strategy.acquire_initial_retry_token(
            token_scope=call.retry_scope
        )

        while True:
            if retry_token.retry_delay:
                await sleep(retry_token.retry_delay)

            output_context = await self._handle_attempt(
                call,
                replace(
                    request_context,
                    transport_request=copy(request_context.transport_request),
                ),
                request_future,
            )

            if isinstance(output_context.response, Exception):
                try:
                    retry_strategy.refresh_retry_token_for_retry(
                        token_to_renew=retry_token,
                        error=output_context.response,
                    )
                except RetryError:
                    raise output_context.response

                _LOGGER.debug(
                    "Retry needed. Attempting request #%s in %.4f seconds.",
                    retry_token.retry_count + 1,
                    retry_token.retry_delay,
                )

                await seek(request_context.transport_request.body, 0)
            else:
                retry_strategy.record_success(token=retry_token)
                return output_context

    async def _handle_attempt[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        request_context: RequestContext[I, TRequest],
        request_future: Future[RequestContext[I, TRequest]] | None,
    ) -> OutputContext[I, O, TRequest, TResponse | None]:
        output_context: OutputContext[I, O, TRequest, TResponse | None]
        try:
            interceptor = call.interceptor
            interceptor.read_before_attempt(request_context)

            endpoint_params = EndpointResolverParams(
                operation=call.operation,
                input=call.input,
                context=request_context.properties,
            )
            _LOGGER.debug("Calling endpoint resolver with params: %s", endpoint_params)
            endpoint: Endpoint = await call.endpoint_resolver.resolve_endpoint(
                endpoint_params
            )
            _LOGGER.debug("Endpoint resolver result: %s", endpoint)

            request_context = replace(
                request_context,
                transport_request=self.protocol.set_service_endpoint(
                    request=request_context.transport_request, endpoint=endpoint
                ),
            )

            request_context = replace(
                request_context,
                transport_request=interceptor.modify_before_signing(request_context),
            )
            interceptor.read_before_signing(request_context)

            auth_params = AuthParams[I, O](
                protocol_id=self.protocol.id,
                operation=call.operation,
                context=request_context.properties,
            )
            auth = self._resolve_auth(call, auth_params)
            if auth is not None:
                option, scheme = auth
                request_context.properties[AUTH_SCHEME] = scheme
                identity_resolver = scheme.identity_resolver(context=call.context)

                identity_properties = scheme.identity_properties(
                    context=request_context.properties
                )
                identity_properties.update(option.identity_properties)

                identity = await identity_resolver.get_identity(
                    properties=identity_properties
                )

                signer_properties = scheme.signer_properties(
                    context=request_context.properties
                )
                signer_properties.update(option.identity_properties)
                _LOGGER.debug("Request to sign: %s", request_context.transport_request)
                _LOGGER.debug("Signer properties: %s", signer_properties)

                signer = scheme.signer()
                request_context = replace(
                    request_context,
                    transport_request=await signer.sign(
                        request=request_context.transport_request,
                        identity=identity,
                        properties=signer_properties,
                    ),
                )

            interceptor.read_after_signing(request_context)
            request_context = replace(
                request_context,
                transport_request=interceptor.modify_before_transmit(request_context),
            )
            interceptor.read_before_transmit(request_context)

            _LOGGER.debug("Sending request %s", request_context.transport_request)

            if request_future is not None:
                # If we have an input event stream (or duplex event stream) then we
                # need to let the client return ASAP so that it can start sending
                # events. So here we start the transport send in a background task
                # then set the result of the request future. It's important to sequence
                # it just like that so that the client gets a stream that's ready
                # to send.
                transport_task = asyncio.create_task(
                    self.transport.send(request=request_context.transport_request)
                )
                request_future.set_result(request_context)
                transport_response = await transport_task
            else:
                # If we don't have an input stream, there's no point in creating a
                # task, so we just immediately await the coroutine.
                transport_response = await self.transport.send(
                    request=request_context.transport_request
                )

            _LOGGER.debug("Received response: %s", transport_response)

            response_context = ResponseContext(
                request=request_context.request,
                transport_request=request_context.transport_request,
                transport_response=transport_response,
                properties=request_context.properties,
            )

            interceptor.read_after_transmit(response_context)

            response_context = replace(
                response_context,
                transport_response=interceptor.modify_before_deserialization(
                    response_context
                ),
            )

            interceptor.read_before_deserialization(response_context)

            _LOGGER.debug(
                "Deserializing response: %s", response_context.transport_response
            )

            output = await self.protocol.deserialize_response(
                operation=call.operation,
                request=response_context.transport_request,
                response=response_context.transport_response,
                error_registry=call.operation.error_registry,
                context=response_context.properties,
            )

            _LOGGER.debug("Deserialization complete. Output: %s", output)

            output_context = OutputContext(
                request=response_context.request,
                response=output,
                transport_request=response_context.transport_request,
                transport_response=response_context.transport_response,
                properties=response_context.properties,
            )

            interceptor.read_after_deserialization(output_context)
        except Exception as e:
            output_context = OutputContext(
                request=request_context.request,
                response=e,
                transport_request=request_context.transport_request,
                transport_response=None,
                properties=request_context.properties,
            )

        return self._finalize_attempt(call, output_context)

    def _resolve_auth[I: SerializeableShape, O: DeserializeableShape](
        self, call: ClientCall[Any, Any], params: AuthParams[I, O]
    ) -> tuple[AuthOption, AuthScheme[TRequest, Any, Any, Any]] | None:
        auth_options: Sequence[AuthOption] = (
            call.auth_scheme_resolver.resolve_auth_scheme(auth_parameters=params)
        )

        for option in auth_options:
            if (
                scheme := call.supported_auth_schemes.get(option.scheme_id)
            ) is not None:
                return option, scheme

        return None

    def _finalize_attempt[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        output_context: OutputContext[I, O, TRequest, TResponse | None],
    ) -> OutputContext[I, O, TRequest, TResponse | None]:
        interceptor = call.interceptor
        try:
            output_context = replace(
                output_context,
                response=interceptor.modify_before_attempt_completion(output_context),
            )
        except Exception as e:
            output_context = replace(output_context, response=e)

        try:
            interceptor.read_after_attempt(output_context)
        except Exception as e:
            output_context = replace(output_context, response=e)

        return output_context

    def _finalize_execution[I: SerializeableShape, O: DeserializeableShape](
        self,
        call: ClientCall[I, O],
        output_context: OutputContext[I, O, TRequest | None, TResponse | None],
    ) -> OutputContext[I, O, TRequest | None, TResponse | None]:
        interceptor = call.interceptor
        try:
            output_context = replace(
                output_context,
                response=interceptor.modify_before_completion(output_context),
            )

            # TODO trace probe
        except Exception as e:
            output_context = replace(output_context, response=e)

        try:
            interceptor.read_after_execution(output_context)
        except Exception as e:
            output_context = replace(output_context, response=e)

        return output_context

protocol = protocol instance-attribute

The protocol to use to serialize the request and deserialize the response.

transport = transport instance-attribute

The transport to use to send the request and receive the response (e.g. an HTTP Client).

__call__(call) async

Invoke an operation asynchronously.

:param call: The operation to invoke and associated context.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

async def __call__[I: SerializeableShape, O: DeserializeableShape](
    self, call: ClientCall[I, O], /
) -> O:
    """Invoke an operation asynchronously.

    :param call: The operation to invoke and associated context.
    """
    output, _ = await self._execute_request(call, None)
    return output

duplex_stream(call, input_event_type, output_event_type, event_deserializer) async

Invoke an input stream operation asynchronously.

:param call: The operation to invoke and associated context. :param input_event_type: The event type to send in the input stream. :param output_event_type: The event type to receive in the output stream. :param event_deserializer: The method used to deserialize events.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

async def duplex_stream[
    I: SerializeableShape,
    O: DeserializeableShape,
    IE: SerializeableShape,
    OE: DeserializeableShape,
](
    self,
    call: ClientCall[I, O],
    input_event_type: "TypeForm[IE]",
    output_event_type: "TypeForm[OE]",
    event_deserializer: Callable[[ShapeDeserializer], OE],
    /,
) -> DuplexEventStream[IE, OE, O]:
    """Invoke an input stream operation asynchronously.

    :param call: The operation to invoke and associated context.
    :param input_event_type: The event type to send in the input stream.
    :param output_event_type: The event type to receive in the output stream.
    :param event_deserializer: The method used to deserialize events.
    """
    request_future = Future[RequestContext[I, TRequest]]()
    execute_task = asyncio.create_task(self._execute_request(call, request_future))
    request_context = await request_future
    input_stream = self.protocol.create_event_publisher(
        operation=call.operation,
        request=request_context.transport_request,
        event_type=input_event_type,
        context=request_context.properties,
        auth_scheme=request_context.properties.get(AUTH_SCHEME),
    )
    output_future = asyncio.create_task(
        self._await_output_stream(
            call=call,
            execute_task=execute_task,
            output_event_type=output_event_type,
            event_deserializer=event_deserializer,
        )
    )
    return DuplexEventStream(input_stream=input_stream, output_future=output_future)

input_stream(call, event_type) async

Invoke an input stream operation asynchronously.

:param call: The operation to invoke and associated context. :param event_type: The event type to send in the input stream.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

async def input_stream[
    I: SerializeableShape,
    O: DeserializeableShape,
    E: SerializeableShape,
](
    self, call: ClientCall[I, O], event_type: "TypeForm[E]", /
) -> InputEventStream[E, O]:
    """Invoke an input stream operation asynchronously.

    :param call: The operation to invoke and associated context.
    :param event_type: The event type to send in the input stream.
    """
    request_future = Future[RequestContext[I, TRequest]]()
    output_future = asyncio.create_task(
        self._await_output(self._execute_request(call, request_future))
    )
    request_context = await request_future
    input_stream = self.protocol.create_event_publisher(
        operation=call.operation,
        request=request_context.transport_request,
        event_type=event_type,
        context=request_context.properties,
        auth_scheme=request_context.properties.get(AUTH_SCHEME),
    )
    return InputEventStream(input_stream=input_stream, output_future=output_future)

output_stream(call, event_type, event_deserializer) async

Invoke an input stream operation asynchronously.

:param call: The operation to invoke and associated context. :param event_type: The event type to receive in the output stream. :param event_deserializer: The method used to deserialize events.

Source code in packages/smithy-core/src/smithy_core/aio/client.py

async def output_stream[
    I: SerializeableShape,
    O: DeserializeableShape,
    E: DeserializeableShape,
](
    self,
    call: ClientCall[I, O],
    event_type: "TypeForm[E]",
    event_deserializer: Callable[[ShapeDeserializer], E],
    /,
) -> OutputEventStream[E, O]:
    """Invoke an input stream operation asynchronously.

    :param call: The operation to invoke and associated context.
    :param event_type: The event type to receive in the output stream.
    :param event_deserializer: The method used to deserialize events.
    """
    output, output_context = await self._execute_request(call, None)
    output_stream = self.protocol.create_event_receiver(
        operation=call.operation,
        request=output_context.transport_request,
        response=output_context.transport_response,
        event_type=event_type,
        event_deserializer=event_deserializer,
        context=output_context.properties,
    )
    return OutputEventStream(output_stream=output_stream, output=output)

endpoints

StaticEndpointResolver

Bases: EndpointResolver

A basic endpoint resolver that forwards a static URI.

Source code in packages/smithy-core/src/smithy_core/aio/endpoints.py

class StaticEndpointResolver(EndpointResolver):
    """A basic endpoint resolver that forwards a static URI."""

    async def resolve_endpoint(self, params: EndpointResolverParams[Any]) -> _Endpoint:
        static_uri = resolve_static_uri(params)
        if static_uri is None:
            raise EndpointResolutionError(
                "Unable to resolve endpoint: endpoint_uri is required"
            )

        return Endpoint(uri=static_uri)

eventstream

DuplexEventStream

An event stream that both sends and receives messages.

To ensure that streams are closed upon exiting, this class may be used as an async context manager.

.. code-block:: python

async def main():
    client = ChatClient()
    input = StreamMessagesInput(chat_room="aws-python-sdk", username="hunter7")

    async with client.stream_messages(input=input) as stream:
        stream.input_stream.send(
            MessageStreamMessage("Chat logger starting up.")
        )
        response_task = asyncio.create_task(handle_output(stream))
        stream.input_stream.send(MessageStreamMessage("Chat logger active."))
        await response_handler

    async def handle_output(stream: EventStream) -> None:
        _, output_stream = await stream.await_output()
        async for event in output_stream:
            match event:
                case MessageStreamMessage():
                    print(event.value)
                case MessageStreamShutdown():
                    return
                case _:
                    stream.input_stream.send(
                        MessageStreamMessage(
                            "Unknown message type received. Shutting down."
                        )
                    )
                    return

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

class DuplexEventStream[
    IE: SerializeableShape,
    OE: DeserializeableShape,
    O: DeserializeableShape,
]:
    """An event stream that both sends and receives messages.

    To ensure that streams are closed upon exiting, this class may be used as an async
    context manager.

    .. code-block:: python

        async def main():
            client = ChatClient()
            input = StreamMessagesInput(chat_room="aws-python-sdk", username="hunter7")

            async with client.stream_messages(input=input) as stream:
                stream.input_stream.send(
                    MessageStreamMessage("Chat logger starting up.")
                )
                response_task = asyncio.create_task(handle_output(stream))
                stream.input_stream.send(MessageStreamMessage("Chat logger active."))
                await response_handler

            async def handle_output(stream: EventStream) -> None:
                _, output_stream = await stream.await_output()
                async for event in output_stream:
                    match event:
                        case MessageStreamMessage():
                            print(event.value)
                        case MessageStreamShutdown():
                            return
                        case _:
                            stream.input_stream.send(
                                MessageStreamMessage(
                                    "Unknown message type received. Shutting down."
                                )
                            )
                            return
    """

    input_stream: EventPublisher[IE]
    """An event stream that sends events to the service."""

    output_stream: EventReceiver[OE] | None = None
    """An event stream that receives events from the service.

    This value may be None until ``await_output`` has been called.

    This value will also be None if the operation has no output stream.
    """

    output: O | None = None
    """The initial response from the service.

    This value may be None until ``await_output`` has been called.

    This may include context necessary to interpret output events or prepare
    input events. It will always be available before any events.
    """

    def __init__(
        self,
        *,
        input_stream: EventPublisher[IE],
        output_future: Future[tuple[O, EventReceiver[OE]]],
    ) -> None:
        self.input_stream = input_stream
        self._output_future = output_future

    async def await_output(self) -> tuple[O, EventReceiver[OE]]:
        """Await the operation's output.

        The EventStream will be returned as soon as the input stream is ready to
        receive events, which may be before the initial response has been received
        and the service is ready to send events.

        Awaiting this method will wait until the initial response was received and the
        service is ready to send events. The initial response and output stream will
        be returned by this operation and also cached in ``response`` and
        ``output_stream``, respectively.

        The default implementation of this method performs the caching behavior,
        delegating to the abstract ``_await_output`` method to actually retrieve the
        initial response and output stream.

        :returns: A tuple containing the initial response and output stream. If the
            operation has no output stream, the second value will be None.
        """
        self.output, self.output_stream = await self._output_future
        return self.output, self.output_stream

    async def close(self) -> None:
        """Closes the event stream.

        This closes both the input and output streams.
        """
        if self.output_stream is None:
            _, self.output_stream = await self.await_output()

        await self.input_stream.close()
        await self.output_stream.close()

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        await self.close()

input_stream = input_stream instance-attribute

An event stream that sends events to the service.

output = None class-attribute instance-attribute

The initial response from the service.

This value may be None until await_output has been called.

This may include context necessary to interpret output events or prepare input events. It will always be available before any events.

output_stream = None class-attribute instance-attribute

An event stream that receives events from the service.

This value may be None until await_output has been called.

This value will also be None if the operation has no output stream.

await_output() async

Await the operation's output.

The EventStream will be returned as soon as the input stream is ready to receive events, which may be before the initial response has been received and the service is ready to send events.

Awaiting this method will wait until the initial response was received and the service is ready to send events. The initial response and output stream will be returned by this operation and also cached in response and output_stream, respectively.

The default implementation of this method performs the caching behavior, delegating to the abstract _await_output method to actually retrieve the initial response and output stream.

:returns: A tuple containing the initial response and output stream. If the operation has no output stream, the second value will be None.

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

async def await_output(self) -> tuple[O, EventReceiver[OE]]:
    """Await the operation's output.

    The EventStream will be returned as soon as the input stream is ready to
    receive events, which may be before the initial response has been received
    and the service is ready to send events.

    Awaiting this method will wait until the initial response was received and the
    service is ready to send events. The initial response and output stream will
    be returned by this operation and also cached in ``response`` and
    ``output_stream``, respectively.

    The default implementation of this method performs the caching behavior,
    delegating to the abstract ``_await_output`` method to actually retrieve the
    initial response and output stream.

    :returns: A tuple containing the initial response and output stream. If the
        operation has no output stream, the second value will be None.
    """
    self.output, self.output_stream = await self._output_future
    return self.output, self.output_stream

close() async

Closes the event stream.

This closes both the input and output streams.

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

async def close(self) -> None:
    """Closes the event stream.

    This closes both the input and output streams.
    """
    if self.output_stream is None:
        _, self.output_stream = await self.await_output()

    await self.input_stream.close()
    await self.output_stream.close()

InputEventStream

An event stream that streams messages to the service.

To ensure that streams are closed upon exiting, this class may be used as an async context manager.

.. code-block:: python

async def main():
    client = ChatClient()
    input = PublishMessagesInput(chat_room="aws-python-sdk", username="hunter7")

    async with client.publish_messages(input=input) as stream:
        stream.input_stream.send(
            MessageStreamMessage("High severity ticket alert!")
        )
        await stream.await_output()

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

class InputEventStream[IE: SerializeableShape, O]:
    """An event stream that streams messages to the service.

    To ensure that streams are closed upon exiting, this class may be used as an async
    context manager.

    .. code-block:: python

        async def main():
            client = ChatClient()
            input = PublishMessagesInput(chat_room="aws-python-sdk", username="hunter7")

            async with client.publish_messages(input=input) as stream:
                stream.input_stream.send(
                    MessageStreamMessage("High severity ticket alert!")
                )
                await stream.await_output()
    """

    input_stream: EventPublisher[IE]
    """An event stream that sends events to the service."""

    output: O | None = None
    """The initial response from the service.

    This value may be None until ``await_output`` has been called.

    This may include context necessary to interpret output events or prepare
    input events. It will always be available before any events.
    """

    def __init__(
        self,
        *,
        input_stream: EventPublisher[IE],
        output_future: Future[O],
    ) -> None:
        self.input_stream = input_stream
        self._output_future = output_future

    async def await_output(self) -> O:
        """Await the operation's initial response.

        The EventStream will be returned as soon as the input stream is ready to receive
        events, which may be before the initial response has been received and the
        service is ready to send events.

        Awaiting this method will wait until the initial response was received.

        :returns: The service's initial response.
        """
        if self.output is None:
            self.output = await self._output_future
        return self.output

    async def close(self) -> None:
        """Closes the event stream."""
        await self.input_stream.close()

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        await self.close()

input_stream = input_stream instance-attribute

An event stream that sends events to the service.

output = None class-attribute instance-attribute

The initial response from the service.

This value may be None until await_output has been called.

This may include context necessary to interpret output events or prepare input events. It will always be available before any events.

await_output() async

Await the operation's initial response.

The EventStream will be returned as soon as the input stream is ready to receive events, which may be before the initial response has been received and the service is ready to send events.

Awaiting this method will wait until the initial response was received.

:returns: The service's initial response.

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

async def await_output(self) -> O:
    """Await the operation's initial response.

    The EventStream will be returned as soon as the input stream is ready to receive
    events, which may be before the initial response has been received and the
    service is ready to send events.

    Awaiting this method will wait until the initial response was received.

    :returns: The service's initial response.
    """
    if self.output is None:
        self.output = await self._output_future
    return self.output

close() async

Closes the event stream.

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

async def close(self) -> None:
    """Closes the event stream."""
    await self.input_stream.close()

OutputEventStream

An event stream that streams messages from the service.

To ensure that streams are closed upon exiting, this class may be used as an async context manager.

.. code-block:: python

async def main():
    client = ChatClient()
    input = ReceiveMessagesInput(chat_room="aws-python-sdk")

    async with client.receive_messages(input=input) as stream:
        async for event in stream.output_stream:
            match event:
                case MessageStreamMessage():
                    print(event.value)
                case _:
                    return

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

class OutputEventStream[OE: DeserializeableShape, O: DeserializeableShape]:
    """An event stream that streams messages from the service.

    To ensure that streams are closed upon exiting, this class may be used as an async
    context manager.

    .. code-block:: python

        async def main():
            client = ChatClient()
            input = ReceiveMessagesInput(chat_room="aws-python-sdk")

            async with client.receive_messages(input=input) as stream:
                async for event in stream.output_stream:
                    match event:
                        case MessageStreamMessage():
                            print(event.value)
                        case _:
                            return
    """

    output_stream: EventReceiver[OE]
    """An event stream that receives events from the service.

    This value will also be None if the operation has no output stream.
    """

    output: O
    """The initial response from the service.

    This may include context necessary to interpret output events or prepare input
    events. It will always be available before any events.
    """

    def __init__(self, output_stream: EventReceiver[OE], output: O) -> None:
        self.output_stream = output_stream
        self.output = output

    async def close(self) -> None:
        """Closes the event stream."""
        await self.output_stream.close()

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        await self.close()

output = output instance-attribute

The initial response from the service.

This may include context necessary to interpret output events or prepare input events. It will always be available before any events.

output_stream = output_stream instance-attribute

An event stream that receives events from the service.

This value will also be None if the operation has no output stream.

close() async

Closes the event stream.

Source code in packages/smithy-core/src/smithy_core/aio/eventstream.py

async def close(self) -> None:
    """Closes the event stream."""
    await self.output_stream.close()

identity

ChainedIdentityResolver

Bases: CachingIdentityResolver[I, IP]

Attempts to resolve an identity by checking a sequence of sub-resolvers.

If a nested resolver raises a :py:class:SmithyIdentityError, the next resolver in the chain will be attempted.

Source code in packages/smithy-core/src/smithy_core/aio/identity.py

class ChainedIdentityResolver[I: Identity, IP: Mapping[str, Any]](
    CachingIdentityResolver[I, IP]
):
    """Attempts to resolve an identity by checking a sequence of sub-resolvers.

    If a nested resolver raises a :py:class:`SmithyIdentityError`, the next
    resolver in the chain will be attempted.
    """

    def __init__(self, resolvers: Sequence[IdentityResolver[I, IP]]) -> None:
        """Construct a ChainedIdentityResolver.

        :param resolvers: The sequence of resolvers to resolve identity from.
        """
        super().__init__()
        self._resolvers = resolvers

    async def _get_identity(self, *, properties: IP) -> I:
        logger.debug("Attempting to resolve identity from resolver chain.")
        for resolver in self._resolvers:
            try:
                logger.debug("Attempting to resolve identity from %s.", type(resolver))
                return await resolver.get_identity(properties=properties)
            except SmithyIdentityError as e:
                logger.debug(
                    "Failed to resolve identity from %s: %s", type(resolver), e
                )

        raise SmithyIdentityError("Failed to resolve identity from resolver chain.")

__init__(resolvers)

Construct a ChainedIdentityResolver.

:param resolvers: The sequence of resolvers to resolve identity from.

Source code in packages/smithy-core/src/smithy_core/aio/identity.py

def __init__(self, resolvers: Sequence[IdentityResolver[I, IP]]) -> None:
    """Construct a ChainedIdentityResolver.

    :param resolvers: The sequence of resolvers to resolve identity from.
    """
    super().__init__()
    self._resolvers = resolvers

interfaces

AsyncByteStream

Bases: Protocol

A file-like object with an async read method.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

@runtime_checkable
class AsyncByteStream(Protocol):
    """A file-like object with an async read method."""

    async def read(self, size: int = -1) -> bytes: ...

AsyncWriter

Bases: Protocol

An object with an async write method.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

@runtime_checkable
class AsyncWriter(Protocol):
    """An object with an async write method."""

    async def write(self, data: bytes) -> None: ...

ClientProtocol

Bases: Protocol

A protocol used by a client to communicate with a server.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

class ClientProtocol[I: Request, O: Response](Protocol):
    """A protocol used by a client to communicate with a server."""

    @property
    def id(self) -> "ShapeID":
        """The ID of the protocol."""
        ...

    def serialize_request[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
    ](
        self,
        *,
        operation: "APIOperation[OperationInput, OperationOutput]",
        input: OperationInput,
        endpoint: URI,
        context: TypedProperties,
    ) -> I:
        """Serialize an operation input into a transport request.

        :param operation: The operation whose request is being serialized.
        :param input: The input shape to be serialized.
        :param endpoint: The base endpoint to serialize.
        :param context: A context bag for the request.
        """
        ...

    def set_service_endpoint(
        self,
        *,
        request: I,
        endpoint: Endpoint,
    ) -> I:
        """Update the endpoint of a transport request.

        :param request: The request whose endpoint should be updated.
        :param endpoint: The endpoint to set on the request.
        """
        ...

    async def deserialize_response[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
    ](
        self,
        *,
        operation: "APIOperation[OperationInput, OperationOutput]",
        request: I,
        response: O,
        error_registry: TypeRegistry,
        context: TypedProperties,
    ) -> OperationOutput:
        """Deserializes the output from the tranport response or throws an exception.

        :param operation: The operation whose response is being deserialized.
        :param request: The transport request that was sent for this response.
        :param response: The response to deserialize.
        :param error_registry: A TypeRegistry used to deserialize errors.
        :param context: A context bag for the request.
        """
        ...

    def create_event_publisher[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
        Event: "SerializeableShape",
    ](
        self,
        *,
        operation: "APIOperation[OperationInput, OperationOutput]",
        request: I,
        event_type: "TypeForm[Event]",
        context: TypedProperties,
        auth_scheme: "AuthScheme[Any, Any, Any, Any] | None" = None,
    ) -> EventPublisher[Event]:
        """Creates an event publisher for a protocol event stream.

        :param operation: The event stream operation.
        :param request: The transport request that was sent for this stream.
        :param event_type: The type of event to publish.
        :param context: A context bag for the request.
        :param auth_scheme: The optional auth scheme used to sign events.
        """
        raise UnsupportedStreamError()

    def create_event_receiver[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
        Event: "DeserializeableShape",
    ](
        self,
        *,
        operation: "APIOperation[OperationInput, OperationOutput]",
        request: I,
        response: O,
        event_type: "TypeForm[Event]",
        event_deserializer: Callable[["ShapeDeserializer"], Event],
        context: TypedProperties,
    ) -> EventReceiver[Event]:
        """Creates an event receiver for a protocol event stream.

        :param operation: The event stream operation.
        :param request: The transport request that was sent for this stream.
        :param response: The transport response that was received for this stream.
        :param event_type: The type of event to publish.
        :param event_deserializer: The deserializer to be used to deserialize events.
        :param context: A context bag for the request.
        """
        raise UnsupportedStreamError()

id property

The ID of the protocol.

create_event_publisher(*, operation, request, event_type, context, auth_scheme=None)

Creates an event publisher for a protocol event stream.

:param operation: The event stream operation. :param request: The transport request that was sent for this stream. :param event_type: The type of event to publish. :param context: A context bag for the request. :param auth_scheme: The optional auth scheme used to sign events.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def create_event_publisher[
    OperationInput: "SerializeableShape",
    OperationOutput: "DeserializeableShape",
    Event: "SerializeableShape",
](
    self,
    *,
    operation: "APIOperation[OperationInput, OperationOutput]",
    request: I,
    event_type: "TypeForm[Event]",
    context: TypedProperties,
    auth_scheme: "AuthScheme[Any, Any, Any, Any] | None" = None,
) -> EventPublisher[Event]:
    """Creates an event publisher for a protocol event stream.

    :param operation: The event stream operation.
    :param request: The transport request that was sent for this stream.
    :param event_type: The type of event to publish.
    :param context: A context bag for the request.
    :param auth_scheme: The optional auth scheme used to sign events.
    """
    raise UnsupportedStreamError()

create_event_receiver(*, operation, request, response, event_type, event_deserializer, context)

Creates an event receiver for a protocol event stream.

:param operation: The event stream operation. :param request: The transport request that was sent for this stream. :param response: The transport response that was received for this stream. :param event_type: The type of event to publish. :param event_deserializer: The deserializer to be used to deserialize events. :param context: A context bag for the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def create_event_receiver[
    OperationInput: "SerializeableShape",
    OperationOutput: "DeserializeableShape",
    Event: "DeserializeableShape",
](
    self,
    *,
    operation: "APIOperation[OperationInput, OperationOutput]",
    request: I,
    response: O,
    event_type: "TypeForm[Event]",
    event_deserializer: Callable[["ShapeDeserializer"], Event],
    context: TypedProperties,
) -> EventReceiver[Event]:
    """Creates an event receiver for a protocol event stream.

    :param operation: The event stream operation.
    :param request: The transport request that was sent for this stream.
    :param response: The transport response that was received for this stream.
    :param event_type: The type of event to publish.
    :param event_deserializer: The deserializer to be used to deserialize events.
    :param context: A context bag for the request.
    """
    raise UnsupportedStreamError()

deserialize_response(*, operation, request, response, error_registry, context) async

Deserializes the output from the tranport response or throws an exception.

:param operation: The operation whose response is being deserialized. :param request: The transport request that was sent for this response. :param response: The response to deserialize. :param error_registry: A TypeRegistry used to deserialize errors. :param context: A context bag for the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

async def deserialize_response[
    OperationInput: "SerializeableShape",
    OperationOutput: "DeserializeableShape",
](
    self,
    *,
    operation: "APIOperation[OperationInput, OperationOutput]",
    request: I,
    response: O,
    error_registry: TypeRegistry,
    context: TypedProperties,
) -> OperationOutput:
    """Deserializes the output from the tranport response or throws an exception.

    :param operation: The operation whose response is being deserialized.
    :param request: The transport request that was sent for this response.
    :param response: The response to deserialize.
    :param error_registry: A TypeRegistry used to deserialize errors.
    :param context: A context bag for the request.
    """
    ...

serialize_request(*, operation, input, endpoint, context)

Serialize an operation input into a transport request.

:param operation: The operation whose request is being serialized. :param input: The input shape to be serialized. :param endpoint: The base endpoint to serialize. :param context: A context bag for the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def serialize_request[
    OperationInput: "SerializeableShape",
    OperationOutput: "DeserializeableShape",
](
    self,
    *,
    operation: "APIOperation[OperationInput, OperationOutput]",
    input: OperationInput,
    endpoint: URI,
    context: TypedProperties,
) -> I:
    """Serialize an operation input into a transport request.

    :param operation: The operation whose request is being serialized.
    :param input: The input shape to be serialized.
    :param endpoint: The base endpoint to serialize.
    :param context: A context bag for the request.
    """
    ...

set_service_endpoint(*, request, endpoint)

Update the endpoint of a transport request.

:param request: The request whose endpoint should be updated. :param endpoint: The endpoint to set on the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def set_service_endpoint(
    self,
    *,
    request: I,
    endpoint: Endpoint,
) -> I:
    """Update the endpoint of a transport request.

    :param request: The request whose endpoint should be updated.
    :param endpoint: The endpoint to set on the request.
    """
    ...

ClientTransport

Bases: Protocol

Protocol-agnostic representation of a client tranport (e.g. an HTTP client).

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

class ClientTransport[I: Request, O: Response](Protocol):
    """Protocol-agnostic representation of a client tranport (e.g. an HTTP client)."""

    async def send(self, request: I) -> O:
        """Send a request over the transport and receive the response."""
        ...

send(request) async

Send a request over the transport and receive the response.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

async def send(self, request: I) -> O:
    """Send a request over the transport and receive the response."""
    ...

EndpointResolver

Bases: Protocol

Resolves an operation's endpoint based given parameters.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

class EndpointResolver(Protocol):
    """Resolves an operation's endpoint based given parameters."""

    async def resolve_endpoint(self, params: EndpointResolverParams[Any]) -> Endpoint:
        """Resolve an endpoint for the given operation.

        :param params: The parameters available to resolve the endpoint.
        """
        ...

resolve_endpoint(params) async

Resolve an endpoint for the given operation.

:param params: The parameters available to resolve the endpoint.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

async def resolve_endpoint(self, params: EndpointResolverParams[Any]) -> Endpoint:
    """Resolve an endpoint for the given operation.

    :param params: The parameters available to resolve the endpoint.
    """
    ...

Request

Bases: Protocol

Protocol-agnostic representation of a request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

class Request(Protocol):
    """Protocol-agnostic representation of a request."""

    destination: URI
    """The URI where the request should be sent to."""

    body: StreamingBlob = b""
    """The request payload."""

    async def consume_body_async(self) -> bytes:
        """Iterate over request body and return as bytes."""
        ...

    def consume_body(self) -> bytes:
        """Iterate over request body and return as bytes."""
        ...

body = b'' class-attribute instance-attribute

The request payload.

destination instance-attribute

The URI where the request should be sent to.

consume_body()

Iterate over request body and return as bytes.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def consume_body(self) -> bytes:
    """Iterate over request body and return as bytes."""
    ...

consume_body_async() async

Iterate over request body and return as bytes.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

async def consume_body_async(self) -> bytes:
    """Iterate over request body and return as bytes."""
    ...

Response

Bases: Protocol

Protocol-agnostic representation of a response.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

class Response(Protocol):
    """Protocol-agnostic representation of a response."""

    @property
    def body(self) -> StreamingBlob:
        """The response payload as iterable of chunks of bytes."""
        ...

    async def consume_body_async(self) -> bytes:
        """Iterate over response body and return as bytes."""
        ...

    def consume_body(self) -> bytes:
        """Iterate over request body and return as bytes."""
        ...

body property

The response payload as iterable of chunks of bytes.

consume_body()

Iterate over request body and return as bytes.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

def consume_body(self) -> bytes:
    """Iterate over request body and return as bytes."""
    ...

consume_body_async() async

Iterate over response body and return as bytes.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

async def consume_body_async(self) -> bytes:
    """Iterate over response body and return as bytes."""
    ...

auth

AuthScheme

Bases: Protocol

A class that coordinates identity and auth.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

class AuthScheme[R: Request, I: Identity, IP: Mapping[str, Any], SP: Mapping[str, Any]](
    Protocol
):
    """A class that coordinates identity and auth."""

    scheme_id: ShapeID
    """The ID of the auth scheme."""

    def identity_properties(self, *, context: _TypedProperties) -> IP:
        """Construct identity properties from the request context.

        The context will always include the client's config under "config". Other
        properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

        :param context: The context of the request.
        """
        ...

    def identity_resolver(
        self, *, context: _TypedProperties
    ) -> IdentityResolver[I, IP]:
        """Get an identity resolver for the request.

        The context will always include the client's config under "config". Other
        properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

        :param context: The context of the request.
        """
        ...

    def signer_properties(self, *, context: _TypedProperties) -> SP:
        """Construct signer properties from the request context.

        The context will always include the client's config under "config". Other
        properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

        :param context: The context of the request.
        """
        ...

    def signer(self) -> Signer[R, I, SP]:
        """Get a signer for the request."""
        ...

    def event_signer(self, *, request: R) -> EventSigner[I, SP] | None:
        """Construct a signer for event stream events.

        :param request: The request that will initiate the event stream. The request
            will not have been sent when this method is called.
        :returns: An event signer if the scheme supports signing events, otherwise None.
        """
        return None

scheme_id instance-attribute

The ID of the auth scheme.

event_signer(*, request)

Construct a signer for event stream events.

:param request: The request that will initiate the event stream. The request will not have been sent when this method is called. :returns: An event signer if the scheme supports signing events, otherwise None.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

def event_signer(self, *, request: R) -> EventSigner[I, SP] | None:
    """Construct a signer for event stream events.

    :param request: The request that will initiate the event stream. The request
        will not have been sent when this method is called.
    :returns: An event signer if the scheme supports signing events, otherwise None.
    """
    return None

identity_properties(*, context)

Construct identity properties from the request context.

The context will always include the client's config under "config". Other properties may be added by :py:class:smithy_core.interceptors.Interceptors.

:param context: The context of the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

def identity_properties(self, *, context: _TypedProperties) -> IP:
    """Construct identity properties from the request context.

    The context will always include the client's config under "config". Other
    properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

    :param context: The context of the request.
    """
    ...

identity_resolver(*, context)

Get an identity resolver for the request.

The context will always include the client's config under "config". Other properties may be added by :py:class:smithy_core.interceptors.Interceptors.

:param context: The context of the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

def identity_resolver(
    self, *, context: _TypedProperties
) -> IdentityResolver[I, IP]:
    """Get an identity resolver for the request.

    The context will always include the client's config under "config". Other
    properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

    :param context: The context of the request.
    """
    ...

signer()

Get a signer for the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

def signer(self) -> Signer[R, I, SP]:
    """Get a signer for the request."""
    ...

signer_properties(*, context)

Construct signer properties from the request context.

The context will always include the client's config under "config". Other properties may be added by :py:class:smithy_core.interceptors.Interceptors.

:param context: The context of the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

def signer_properties(self, *, context: _TypedProperties) -> SP:
    """Construct signer properties from the request context.

    The context will always include the client's config under "config". Other
    properties may be added by :py:class:`smithy_core.interceptors.Interceptor`s.

    :param context: The context of the request.
    """
    ...

EventSigner

Bases: Protocol

A class that signs requests before they are sent.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

class EventSigner[I, SP: Mapping[str, Any]](Protocol):
    """A class that signs requests before they are sent."""

    # TODO: add a protocol type for events
    async def sign(self, *, event: Any, identity: I, properties: SP) -> Any:
        """Get a signed version of the event.

        :param event: The event to be signed.
        """
        ...

sign(*, event, identity, properties) async

Get a signed version of the event.

:param event: The event to be signed.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

async def sign(self, *, event: Any, identity: I, properties: SP) -> Any:
    """Get a signed version of the event.

    :param event: The event to be signed.
    """
    ...

Signer

Bases: Protocol

A class that signs requests before they are sent.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

class Signer[R: Request, I, SP: Mapping[str, Any]](Protocol):
    """A class that signs requests before they are sent."""

    async def sign(self, *, request: R, identity: I, properties: SP) -> R:
        """Get a signed version of the request.

        :param request: The request to be signed.
        :param identity: The identity to use to sign the request.
        :param properties: Additional properties used to sign the request.
        """
        ...

sign(*, request, identity, properties) async

Get a signed version of the request.

:param request: The request to be signed. :param identity: The identity to use to sign the request. :param properties: Additional properties used to sign the request.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/auth.py

async def sign(self, *, request: R, identity: I, properties: SP) -> R:
    """Get a signed version of the request.

    :param request: The request to be signed.
    :param identity: The identity to use to sign the request.
    :param properties: Additional properties used to sign the request.
    """
    ...

eventstream

EventPublisher

Bases: Protocol

Asynchronously sends events to a service.

This may be used as a context manager to ensure the stream is closed before exiting.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

class EventPublisher[E: SerializeableShape](Protocol):
    """Asynchronously sends events to a service.

    This may be used as a context manager to ensure the stream is closed before exiting.
    """

    async def send(self, event: E) -> None:
        """Sends an event to the service.

        :param event: The event to send.
        """
        ...

    async def close(self) -> None:
        """Closes the event stream."""
        ...

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        await self.close()

close() async

Closes the event stream.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

async def close(self) -> None:
    """Closes the event stream."""
    ...

send(event) async

Sends an event to the service.

:param event: The event to send.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

async def send(self, event: E) -> None:
    """Sends an event to the service.

    :param event: The event to send.
    """
    ...

EventReceiver

Bases: Protocol

Asynchronously receives events from a service.

Events may be received via the receive method or by using this class as an async iterable.

This may also be used as a context manager to ensure the stream is closed before exiting.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

class EventReceiver[E: DeserializeableShape](Protocol):
    """Asynchronously receives events from a service.

    Events may be received via the ``receive`` method or by using this class as
    an async iterable.

    This may also be used as a context manager to ensure the stream is closed before
    exiting.
    """

    async def receive(self) -> E | None:
        """Receive a single event from the service.

        :returns: An event or None. None indicates that no more events will be sent by
            the service.
        """
        ...

    async def close(self) -> None:
        """Closes the event stream."""
        ...

    async def __anext__(self) -> E:
        result = await self.receive()
        if result is None:
            await self.close()
            raise StopAsyncIteration
        return result

    def __aiter__(self) -> Self:
        return self

    async def __enter__(self) -> Self:
        return self

    async def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        await self.close()

close() async

Closes the event stream.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

async def close(self) -> None:
    """Closes the event stream."""
    ...

receive() async

Receive a single event from the service.

:returns: An event or None. None indicates that no more events will be sent by the service.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/eventstream.py

async def receive(self) -> E | None:
    """Receive a single event from the service.

    :returns: An event or None. None indicates that no more events will be sent by
        the service.
    """
    ...

identity

IdentityResolver

Bases: Protocol

Used to load a user's Identity from a given source.

Each Identity may have one or more resolver implementations.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/identity.py

class IdentityResolver[I: Identity, IP: Mapping[str, Any]](Protocol):
    """Used to load a user's `Identity` from a given source.

    Each `Identity` may have one or more resolver implementations.
    """

    async def get_identity(self, *, properties: IP) -> I:
        """Load the user's identity from this resolver.

        :param properties: Properties used to help determine the identity to return.
        """
        ...

get_identity(*, properties) async

Load the user's identity from this resolver.

:param properties: Properties used to help determine the identity to return.

Source code in packages/smithy-core/src/smithy_core/aio/interfaces/identity.py

async def get_identity(self, *, properties: IP) -> I:
    """Load the user's identity from this resolver.

    :param properties: Properties used to help determine the identity to return.
    """
    ...

types

AsyncBytesProvider

A buffer that allows chunks of bytes to be exchanged asynchronously.

Bytes are written in chunks to an internal buffer, that is then drained via an async iterator.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

class AsyncBytesProvider:
    """A buffer that allows chunks of bytes to be exchanged asynchronously.

    Bytes are written in chunks to an internal buffer, that is then drained via an async
    iterator.
    """

    def __init__(
        self, intial_data: bytes | None = None, max_buffered_chunks: int = 16
    ) -> None:
        """Initialize the AsyncBytesProvider.

        :param initial_data: An initial chunk of bytes to make available.
        :param max_buffered_chunks: The maximum number of chunks of data to buffer.
            Calls to ``write`` will block until the number of chunks is less than this
            number. Default is 16.
        """
        self._data = deque[bytes]()
        if intial_data is not None:
            self._data.append(intial_data)

        if max_buffered_chunks < 1:
            raise ValueError(
                "The maximum number of buffered chunks must be greater than 0."
            )

        self._closed = False
        self._closing = False
        self._flushing = False
        self._max_buffered_chunks = max_buffered_chunks

        # Create a Condition to synchronize access to the data chunk pool.
        self._data_condition = asyncio.Condition()

    async def write(self, data: bytes) -> None:
        if self._closed:
            raise SmithyError("Attempted to write to a closed provider.")

        # Acquire a lock on the data buffer, releasing it automatically when the
        # block exits.
        async with self._data_condition:
            # Wait for the number of chunks in the buffer to be less than the
            # specified maximum. This also releases the lock until that condition
            # is met.
            await self._data_condition.wait_for(self._can_write)

            # The provider could have been closed while waiting to write, so an
            # additional check is done here for safety.
            if self._closed or self._closing:
                # Notify to allow other coroutines to check their conditions.
                self._data_condition.notify()
                raise SmithyError("Attempted to write to a closed or closing provider.")

            # Add a new chunk of data to the buffer and notify the next waiting
            # coroutine.
            self._data.append(data)
            self._data_condition.notify()

    def _can_write(self) -> bool:
        return (
            self._closed
            or self._closing
            or (len(self._data) < self._max_buffered_chunks and not self._flushing)
        )

    @property
    def closed(self) -> bool:
        """Returns whether the provider is closed."""
        return self._closed

    async def flush(self) -> None:
        """Waits for all buffered data to be consumed."""
        if self._closed:
            return

        # Acquire a lock on the data buffer, releasing it automatically when the
        # block exits.
        async with self._data_condition:
            # Block writes
            self._flushing = True

            # Wait for the stream to be closed or for the data buffer to be empty,
            # releasing the lock until the condition is met.
            await self._data_condition.wait_for(lambda: len(self._data) == 0)

            # Unblock writes
            self._flushing = False

    async def close(self, flush: bool = True) -> None:
        """Closes the provider.

        Pending writing tasks queued after this will fail, so such tasks should be
        awaited before this. Write tasks queued before this may succeed, however.

        :param flush: Whether to flush buffered data before closing. If false, all
            buffered data will be lost. Default is False.
        """
        if self._closed:
            return

        # Acquire a lock on the data buffer, releasing it automatically when the
        # block exits. Notably this will not wait on a condition to move forward.
        async with self._data_condition:
            self._closing = True
            if flush:
                # Release the lock until the buffer is empty.
                await self._data_condition.wait_for(lambda: len(self._data) == 0)
            else:
                # Clear out any pending data, freeing up memory.
                self._data.clear()

            self._closed = True
            self._closing = False

            # Notify all waiting coroutines that the provider has closed.
            self._data_condition.notify_all()

    def __aiter__(self) -> Self:
        return self

    async def __anext__(self) -> bytes:
        # Acquire a lock on the data buffer, releasing it automatically when the
        # block exits.
        async with self._data_condition:
            # Wait for the stream to be closed or for the data buffer to be non-empty.
            # This also releases the lock until that condition is met.
            await self._data_condition.wait_for(
                lambda: self._closed or len(self._data) > 0
            )

            # If the provider is closed, end the iteration.
            if self._closed:
                raise StopAsyncIteration

            # Pop the next chunk of data from the buffer, then notify any waiting
            # coroutines, returning immediately after.
            result = self._data.popleft()
            self._data_condition.notify()
            return result

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
        await self.close(flush=True)

closed property

Returns whether the provider is closed.

__init__(intial_data=None, max_buffered_chunks=16)

Initialize the AsyncBytesProvider.

:param initial_data: An initial chunk of bytes to make available. :param max_buffered_chunks: The maximum number of chunks of data to buffer. Calls to write will block until the number of chunks is less than this number. Default is 16.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def __init__(
    self, intial_data: bytes | None = None, max_buffered_chunks: int = 16
) -> None:
    """Initialize the AsyncBytesProvider.

    :param initial_data: An initial chunk of bytes to make available.
    :param max_buffered_chunks: The maximum number of chunks of data to buffer.
        Calls to ``write`` will block until the number of chunks is less than this
        number. Default is 16.
    """
    self._data = deque[bytes]()
    if intial_data is not None:
        self._data.append(intial_data)

    if max_buffered_chunks < 1:
        raise ValueError(
            "The maximum number of buffered chunks must be greater than 0."
        )

    self._closed = False
    self._closing = False
    self._flushing = False
    self._max_buffered_chunks = max_buffered_chunks

    # Create a Condition to synchronize access to the data chunk pool.
    self._data_condition = asyncio.Condition()

close(flush=True) async

Closes the provider.

Pending writing tasks queued after this will fail, so such tasks should be awaited before this. Write tasks queued before this may succeed, however.

:param flush: Whether to flush buffered data before closing. If false, all buffered data will be lost. Default is False.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def close(self, flush: bool = True) -> None:
    """Closes the provider.

    Pending writing tasks queued after this will fail, so such tasks should be
    awaited before this. Write tasks queued before this may succeed, however.

    :param flush: Whether to flush buffered data before closing. If false, all
        buffered data will be lost. Default is False.
    """
    if self._closed:
        return

    # Acquire a lock on the data buffer, releasing it automatically when the
    # block exits. Notably this will not wait on a condition to move forward.
    async with self._data_condition:
        self._closing = True
        if flush:
            # Release the lock until the buffer is empty.
            await self._data_condition.wait_for(lambda: len(self._data) == 0)
        else:
            # Clear out any pending data, freeing up memory.
            self._data.clear()

        self._closed = True
        self._closing = False

        # Notify all waiting coroutines that the provider has closed.
        self._data_condition.notify_all()

flush() async

Waits for all buffered data to be consumed.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def flush(self) -> None:
    """Waits for all buffered data to be consumed."""
    if self._closed:
        return

    # Acquire a lock on the data buffer, releasing it automatically when the
    # block exits.
    async with self._data_condition:
        # Block writes
        self._flushing = True

        # Wait for the stream to be closed or for the data buffer to be empty,
        # releasing the lock until the condition is met.
        await self._data_condition.wait_for(lambda: len(self._data) == 0)

        # Unblock writes
        self._flushing = False

AsyncBytesReader

A file-like object with an async read method.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

class AsyncBytesReader:
    """A file-like object with an async read method."""

    # BytesIO *is* a ByteStream, but mypy will nevertheless complain if it isn't here.
    _data: BytesReader | AsyncByteStream | AsyncIterable[bytes] | BytesIO | None
    _closed = False

    def __init__(self, data: StreamingBlob):
        """Initializes self.

        Data is read from the source on an as-needed basis and is not buffered.

        :param data: The source data to read from.
        """
        self._remainder = b""

        # pylint: disable-next=isinstance-second-argument-not-valid-type
        if isinstance(data, bytes | bytearray):
            self._data = BytesIO(data)
        else:
            self._data = data

        if hasattr(self._data, "seek"):
            if (seekable := getattr(self._data, "seekable", None)) is not None:
                self._seekable = seekable()
            else:
                self._seekable = True
        else:
            self._seekable = False

    async def read(self, size: int = -1) -> bytes:
        """Read a number of bytes from the stream.

        :param size: The maximum number of bytes to read. If less than 0, all bytes will
            be read.
        """
        if self._closed or not self._data:
            raise ValueError("I/O operation on closed file.")

        if isinstance(self._data, BytesReader) and not iscoroutinefunction(  # type: ignore - TODO(pyright)
            self._data.read
        ):
            # Python's runtime_checkable can't actually tell the difference between
            # sync and async, so we have to check ourselves.
            return self._data.read(size)

        if isinstance(self._data, AsyncByteStream):  # type: ignore - TODO(pyright)
            return await self._data.read(size)

        return await self._read_from_iterable(
            cast(AsyncIterable[bytes], self._data), size
        )

    async def seek(self, offset: int, whence: int = 0) -> int:
        """Moves the cursor to a position relatve to the position indicated by whence.

        Whence can have one of three values:

        * 0 => The offset is relative to the start of the stream.

        * 1 => The offset is relative to the current location of the cursor.

        * 2 => The offset is relative to the end of the stream.

        :param offset: The amount of movement to be done relative to whence.
        :param whence: The location the offset is relative to.
        :returns: Returns the new position of the cursor.
        """
        if not self._seekable:
            raise NotImplementedError

        return await seek(self._data, offset, whence)  # type: ignore

    def tell(self) -> int:
        """Returns the position of the cursor."""
        if self._seekable:
            if (tell := getattr(self._data, "tell", None)) is not None:
                return tell()
        raise NotImplementedError

    async def _read_from_iterable(
        self, iterator: AsyncIterable[bytes], size: int
    ) -> bytes:
        # This takes the iterator as an arg here just to avoid mypy complaints, since
        # we know it's an iterator where this is called.
        result = self._remainder
        if size < 0:
            async for element in iterator:
                result += element
            self._remainder = b""
            return result

        if len(result) < size:
            async for element in iterator:
                result += element
                if len(result) >= size:
                    break

        self._remainder = result[size:]
        return result[:size]

    def __aiter__(self) -> AsyncIterator[bytes]:
        return self.iter_chunks()

    def iter_chunks(
        self, chunk_size: int = _DEFAULT_CHUNK_SIZE
    ) -> AsyncIterator[bytes]:
        """Iterate over the reader in chunks of a given size.

        :param chunk_size: The maximum size of each chunk. If less than 0, the entire
            reader will be read into one chunk.
        """
        return _AsyncByteStreamIterator(self.read, chunk_size)

    def readable(self) -> bool:
        """Returns whether the stream is readable."""
        return True

    def writeable(self) -> bool:
        """Returns whether the stream is writeable."""
        return False

    def seekable(self) -> bool:
        """Returns whether the stream is seekable."""
        return self._seekable

    @property
    def closed(self) -> bool:
        """Returns whether the stream is closed."""
        return self._closed

    async def close(self) -> None:
        """Closes the stream, as well as the underlying stream where possible."""
        self._closed = True
        await close(self._data)
        self._data = None

closed property

Returns whether the stream is closed.

__init__(data)

Initializes self.

Data is read from the source on an as-needed basis and is not buffered.

:param data: The source data to read from.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def __init__(self, data: StreamingBlob):
    """Initializes self.

    Data is read from the source on an as-needed basis and is not buffered.

    :param data: The source data to read from.
    """
    self._remainder = b""

    # pylint: disable-next=isinstance-second-argument-not-valid-type
    if isinstance(data, bytes | bytearray):
        self._data = BytesIO(data)
    else:
        self._data = data

    if hasattr(self._data, "seek"):
        if (seekable := getattr(self._data, "seekable", None)) is not None:
            self._seekable = seekable()
        else:
            self._seekable = True
    else:
        self._seekable = False

close() async

Closes the stream, as well as the underlying stream where possible.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def close(self) -> None:
    """Closes the stream, as well as the underlying stream where possible."""
    self._closed = True
    await close(self._data)
    self._data = None

iter_chunks(chunk_size=_DEFAULT_CHUNK_SIZE)

Iterate over the reader in chunks of a given size.

:param chunk_size: The maximum size of each chunk. If less than 0, the entire reader will be read into one chunk.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def iter_chunks(
    self, chunk_size: int = _DEFAULT_CHUNK_SIZE
) -> AsyncIterator[bytes]:
    """Iterate over the reader in chunks of a given size.

    :param chunk_size: The maximum size of each chunk. If less than 0, the entire
        reader will be read into one chunk.
    """
    return _AsyncByteStreamIterator(self.read, chunk_size)

read(size=-1) async

Read a number of bytes from the stream.

:param size: The maximum number of bytes to read. If less than 0, all bytes will be read.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def read(self, size: int = -1) -> bytes:
    """Read a number of bytes from the stream.

    :param size: The maximum number of bytes to read. If less than 0, all bytes will
        be read.
    """
    if self._closed or not self._data:
        raise ValueError("I/O operation on closed file.")

    if isinstance(self._data, BytesReader) and not iscoroutinefunction(  # type: ignore - TODO(pyright)
        self._data.read
    ):
        # Python's runtime_checkable can't actually tell the difference between
        # sync and async, so we have to check ourselves.
        return self._data.read(size)

    if isinstance(self._data, AsyncByteStream):  # type: ignore - TODO(pyright)
        return await self._data.read(size)

    return await self._read_from_iterable(
        cast(AsyncIterable[bytes], self._data), size
    )

readable()

Returns whether the stream is readable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def readable(self) -> bool:
    """Returns whether the stream is readable."""
    return True

seek(offset, whence=0) async

Moves the cursor to a position relatve to the position indicated by whence.

Whence can have one of three values:

• 0 => The offset is relative to the start of the stream.

• 1 => The offset is relative to the current location of the cursor.

• 2 => The offset is relative to the end of the stream.

:param offset: The amount of movement to be done relative to whence. :param whence: The location the offset is relative to. :returns: Returns the new position of the cursor.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def seek(self, offset: int, whence: int = 0) -> int:
    """Moves the cursor to a position relatve to the position indicated by whence.

    Whence can have one of three values:

    * 0 => The offset is relative to the start of the stream.

    * 1 => The offset is relative to the current location of the cursor.

    * 2 => The offset is relative to the end of the stream.

    :param offset: The amount of movement to be done relative to whence.
    :param whence: The location the offset is relative to.
    :returns: Returns the new position of the cursor.
    """
    if not self._seekable:
        raise NotImplementedError

    return await seek(self._data, offset, whence)  # type: ignore

seekable()

Returns whether the stream is seekable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def seekable(self) -> bool:
    """Returns whether the stream is seekable."""
    return self._seekable

tell()

Returns the position of the cursor.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def tell(self) -> int:
    """Returns the position of the cursor."""
    if self._seekable:
        if (tell := getattr(self._data, "tell", None)) is not None:
            return tell()
    raise NotImplementedError

writeable()

Returns whether the stream is writeable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def writeable(self) -> bool:
    """Returns whether the stream is writeable."""
    return False

SeekableAsyncBytesReader

A file-like object with async read and seek methods.

Data is written into a buffer as it is read to enable seeking non-seekable streams.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

class SeekableAsyncBytesReader:
    """A file-like object with async read and seek methods.

    Data is written into a buffer as it is read to enable seeking non-seekable streams.
    """

    def __init__(self, data: StreamingBlob):
        """Initializes self.

        :param data: The source data to read from.
        """
        # pylint: disable-next=isinstance-second-argument-not-valid-type
        if isinstance(data, bytes | bytearray):
            self._buffer = BytesIO(data)
            self._data_source = None
        elif isinstance(data, AsyncByteStream) and iscoroutinefunction(data.read):  # type: ignore - TODO(pyright)
            # Note that we need that iscoroutine check because python won't actually check
            # whether or not the read function is async.
            self._buffer = BytesIO()
            self._data_source = data
        else:
            self._buffer = BytesIO()
            self._data_source = AsyncBytesReader(data)

    async def read(self, size: int = -1) -> bytes:
        """Read a number of bytes from the stream.

        :param size: The maximum number of bytes to read. If less than 0, all bytes will
            be read.
        """
        if self._data_source is None or size == 0:
            return self._buffer.read(size)

        start = self._buffer.tell()
        current_buffer_size = self._buffer.seek(0, 2)

        if size < 0:
            await self._read_into_buffer(size)
        elif (target := start + size) > current_buffer_size:
            amount_to_read = target - current_buffer_size
            await self._read_into_buffer(amount_to_read)

        self._buffer.seek(start, 0)
        return self._buffer.read(size)

    async def seek(self, offset: int, whence: int = 0) -> int:
        """Moves the cursor to a position relatve to the position indicated by whence.

        Whence can have one of three values:

        * 0 => The offset is relative to the start of the stream.

        * 1 => The offset is relative to the current location of the cursor.

        * 2 => The offset is relative to the end of the stream.

        :param offset: The amount of movement to be done relative to whence.
        :param whence: The location the offset is relative to.
        :returns: Returns the new position of the cursor.
        """
        if self._data_source is None:
            return self._buffer.seek(offset, whence)

        if whence >= 2:
            # If the seek is relative to the end of the stream, we need to read the
            # whole thing in from the source.
            self._buffer.seek(0, 2)
            self._buffer.write(await self._data_source.read())
            return self._buffer.seek(offset, whence)

        start = self.tell()
        target = offset
        if whence == 1:
            target += start

        current_buffer_size = self._buffer.seek(0, 2)
        if current_buffer_size < target:
            await self._read_into_buffer(target - current_buffer_size)

        return self._buffer.seek(target, 0)

    async def _read_into_buffer(self, size: int) -> None:
        if self._data_source is None:
            return

        read_bytes = await self._data_source.read(size)
        if len(read_bytes) < size or size < 0:
            self._data_source = None

        self._buffer.seek(0, 2)
        self._buffer.write(read_bytes)

    def tell(self) -> int:
        """Returns the position of the cursor."""
        return self._buffer.tell()

    def __aiter__(self) -> AsyncIterator[bytes]:
        return self.iter_chunks()

    def iter_chunks(
        self, chunk_size: int = _DEFAULT_CHUNK_SIZE
    ) -> AsyncIterator[bytes]:
        """Iterate over the reader in chunks of a given size.

        :param chunk_size: The maximum size of each chunk. If less than 0, the entire
            reader will be read into one chunk.
        """
        return _AsyncByteStreamIterator(self.read, chunk_size)

    def readable(self) -> bool:
        """Returns whether the stream is readable."""
        return True

    def writeable(self) -> bool:
        """Returns whether the stream is writeable."""
        return False

    def seekable(self) -> bool:
        """Returns whether the stream is seekable."""
        return True

    @property
    def closed(self) -> bool:
        """Returns whether the stream is closed."""
        return self._buffer.closed

    async def close(self) -> None:
        """Closes the stream, as well as the underlying stream where possible."""
        self._buffer.close()
        await close(self._data_source)
        self._data_source = None

closed property

Returns whether the stream is closed.

__init__(data)

Initializes self.

:param data: The source data to read from.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def __init__(self, data: StreamingBlob):
    """Initializes self.

    :param data: The source data to read from.
    """
    # pylint: disable-next=isinstance-second-argument-not-valid-type
    if isinstance(data, bytes | bytearray):
        self._buffer = BytesIO(data)
        self._data_source = None
    elif isinstance(data, AsyncByteStream) and iscoroutinefunction(data.read):  # type: ignore - TODO(pyright)
        # Note that we need that iscoroutine check because python won't actually check
        # whether or not the read function is async.
        self._buffer = BytesIO()
        self._data_source = data
    else:
        self._buffer = BytesIO()
        self._data_source = AsyncBytesReader(data)

close() async

Closes the stream, as well as the underlying stream where possible.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def close(self) -> None:
    """Closes the stream, as well as the underlying stream where possible."""
    self._buffer.close()
    await close(self._data_source)
    self._data_source = None

iter_chunks(chunk_size=_DEFAULT_CHUNK_SIZE)

Iterate over the reader in chunks of a given size.

:param chunk_size: The maximum size of each chunk. If less than 0, the entire reader will be read into one chunk.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def iter_chunks(
    self, chunk_size: int = _DEFAULT_CHUNK_SIZE
) -> AsyncIterator[bytes]:
    """Iterate over the reader in chunks of a given size.

    :param chunk_size: The maximum size of each chunk. If less than 0, the entire
        reader will be read into one chunk.
    """
    return _AsyncByteStreamIterator(self.read, chunk_size)

read(size=-1) async

Read a number of bytes from the stream.

:param size: The maximum number of bytes to read. If less than 0, all bytes will be read.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def read(self, size: int = -1) -> bytes:
    """Read a number of bytes from the stream.

    :param size: The maximum number of bytes to read. If less than 0, all bytes will
        be read.
    """
    if self._data_source is None or size == 0:
        return self._buffer.read(size)

    start = self._buffer.tell()
    current_buffer_size = self._buffer.seek(0, 2)

    if size < 0:
        await self._read_into_buffer(size)
    elif (target := start + size) > current_buffer_size:
        amount_to_read = target - current_buffer_size
        await self._read_into_buffer(amount_to_read)

    self._buffer.seek(start, 0)
    return self._buffer.read(size)

readable()

Returns whether the stream is readable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def readable(self) -> bool:
    """Returns whether the stream is readable."""
    return True

seek(offset, whence=0) async

Moves the cursor to a position relatve to the position indicated by whence.

Whence can have one of three values:

• 0 => The offset is relative to the start of the stream.

• 1 => The offset is relative to the current location of the cursor.

• 2 => The offset is relative to the end of the stream.

:param offset: The amount of movement to be done relative to whence. :param whence: The location the offset is relative to. :returns: Returns the new position of the cursor.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

async def seek(self, offset: int, whence: int = 0) -> int:
    """Moves the cursor to a position relatve to the position indicated by whence.

    Whence can have one of three values:

    * 0 => The offset is relative to the start of the stream.

    * 1 => The offset is relative to the current location of the cursor.

    * 2 => The offset is relative to the end of the stream.

    :param offset: The amount of movement to be done relative to whence.
    :param whence: The location the offset is relative to.
    :returns: Returns the new position of the cursor.
    """
    if self._data_source is None:
        return self._buffer.seek(offset, whence)

    if whence >= 2:
        # If the seek is relative to the end of the stream, we need to read the
        # whole thing in from the source.
        self._buffer.seek(0, 2)
        self._buffer.write(await self._data_source.read())
        return self._buffer.seek(offset, whence)

    start = self.tell()
    target = offset
    if whence == 1:
        target += start

    current_buffer_size = self._buffer.seek(0, 2)
    if current_buffer_size < target:
        await self._read_into_buffer(target - current_buffer_size)

    return self._buffer.seek(target, 0)

seekable()

Returns whether the stream is seekable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def seekable(self) -> bool:
    """Returns whether the stream is seekable."""
    return True

tell()

Returns the position of the cursor.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def tell(self) -> int:
    """Returns the position of the cursor."""
    return self._buffer.tell()

writeable()

Returns whether the stream is writeable.

Source code in packages/smithy-core/src/smithy_core/aio/types.py

def writeable(self) -> bool:
    """Returns whether the stream is writeable."""
    return False

utils

async_list(lst) async

Turn an Iterable into an AsyncIterable.

Source code in packages/smithy-core/src/smithy_core/aio/utils.py

async def async_list[E](lst: Iterable[E]) -> AsyncIterable[E]:
    """Turn an Iterable into an AsyncIterable."""
    for x in lst:
        await sleep(0)
        yield x

close(stream) async

Close a stream, awaiting it if it's async.

Source code in packages/smithy-core/src/smithy_core/aio/utils.py

async def close(stream: Any) -> None:
    """Close a stream, awaiting it if it's async."""
    if (close := getattr(stream, "close", None)) is not None:
        if iscoroutine(result := close()):
            await result

read_streaming_blob(body)

Synchronously reads a streaming blob into bytes.

:param body: The streaming blob to read from. :raises AsyncBodyError: If the body is an async type.

Source code in packages/smithy-core/src/smithy_core/aio/utils.py

def read_streaming_blob(body: StreamingBlob) -> bytes:
    """Synchronously reads a streaming blob into bytes.

    :param body: The streaming blob to read from.
    :raises AsyncBodyError: If the body is an async type.
    """
    match body:
        case bytes():
            return body
        case bytearray():
            return bytes(body)
        case BytesReader():
            return body.read()
        case _:
            raise AsyncBodyError(
                f"Expected type {SyncStreamingBlob}, but was {type(body)}"
            )

read_streaming_blob_async(body) async

Asynchronously reads a streaming blob into bytes.

:param body: The streaming blob to read from.

Source code in packages/smithy-core/src/smithy_core/aio/utils.py

async def read_streaming_blob_async(body: StreamingBlob) -> bytes:
    """Asynchronously reads a streaming blob into bytes.

    :param body: The streaming blob to read from.
    """
    match body:
        case AsyncByteStream():
            return await body.read()
        case AsyncIterable():
            full = b""
            async for chunk in body:
                full += chunk
            return full
        case _:
            return read_streaming_blob(body)

seek(stream, offset, whence=0) async

Seek a stream to a specified point.

Source code in packages/smithy-core/src/smithy_core/aio/utils.py

async def seek(stream: Any, offset: int, whence: int = 0) -> int | None:
    """Seek a stream to a specified point."""
    if (seekable := getattr(stream, "seekable", None)) is not None and not seekable():
        return

    if (seek := getattr(stream, "seek", None)) is not None:
        result = seek(offset, whence)
        if iscoroutine(result):
            return await result
        return result

auth

AuthOption dataclass

Auth scheme used for signing and identity resolution.

Source code in packages/smithy-core/src/smithy_core/auth.py

@dataclass(kw_only=True)
class AuthOption:
    """Auth scheme used for signing and identity resolution."""

    scheme_id: ShapeID
    """The ID of the auth scheme to use."""

    identity_properties: _TypedProperties = field(default_factory=TypedProperties)
    """Paramters to pass to the identity resolver method."""

    signer_properties: _TypedProperties = field(default_factory=TypedProperties)
    """Paramters to pass to the signing method."""

identity_properties = field(default_factory=TypedProperties) class-attribute instance-attribute

Paramters to pass to the identity resolver method.

scheme_id instance-attribute

The ID of the auth scheme to use.

signer_properties = field(default_factory=TypedProperties) class-attribute instance-attribute

Paramters to pass to the signing method.

AuthParams dataclass

Parameters passed to an AuthSchemeResolver's resolve_auth_scheme method.

Source code in packages/smithy-core/src/smithy_core/auth.py

@dataclass(kw_only=True, frozen=True)
class AuthParams[I: SerializeableShape, O: DeserializeableShape]:
    """Parameters passed to an AuthSchemeResolver's ``resolve_auth_scheme`` method."""

    protocol_id: ShapeID
    """The ID of the protocol being used for the operation invocation."""

    operation: APIOperation[I, O] = field(repr=False)
    """The schema and associated information about the operation being invoked."""

    context: _TypedProperties
    """The context of the operation invocation."""

context instance-attribute

The context of the operation invocation.

operation = field(repr=False) class-attribute instance-attribute

The schema and associated information about the operation being invoked.

protocol_id instance-attribute

The ID of the protocol being used for the operation invocation.

DefaultAuthResolver

Determines which authentication scheme to use based on modeled auth schemes.

Source code in packages/smithy-core/src/smithy_core/auth.py

class DefaultAuthResolver:
    """Determines which authentication scheme to use based on modeled auth schemes."""

    def resolve_auth_scheme(
        self, *, auth_parameters: AuthParams[Any, Any]
    ) -> Sequence[AuthOption]:
        """Resolve an ordered list of applicable auth schemes.

        :param auth_parameters: The parameters required for determining which
            authentication schemes to potentially use.
        """
        return [
            AuthOption(scheme_id=id)
            for id in auth_parameters.operation.effective_auth_schemes
        ]

resolve_auth_scheme(*, auth_parameters)

Resolve an ordered list of applicable auth schemes.

:param auth_parameters: The parameters required for determining which authentication schemes to potentially use.

Source code in packages/smithy-core/src/smithy_core/auth.py

def resolve_auth_scheme(
    self, *, auth_parameters: AuthParams[Any, Any]
) -> Sequence[AuthOption]:
    """Resolve an ordered list of applicable auth schemes.

    :param auth_parameters: The parameters required for determining which
        authentication schemes to potentially use.
    """
    return [
        AuthOption(scheme_id=id)
        for id in auth_parameters.operation.effective_auth_schemes
    ]

NoAuthResolver

Auth resolver that always returns no auth scheme options.

Source code in packages/smithy-core/src/smithy_core/auth.py

class NoAuthResolver:
    """Auth resolver that always returns no auth scheme options."""

    def resolve_auth_scheme(
        self, *, auth_parameters: AuthParams[Any, Any]
    ) -> Sequence[AuthOption]:
        return []

codecs

Codec

Bases: Protocol

A protocol for Smithy codecs.

Smithy codecs are responsible for serializing and deserializing shapes in a particular format.

Source code in packages/smithy-core/src/smithy_core/codecs.py

@runtime_checkable
class Codec(Protocol):
    """A protocol for Smithy codecs.

    Smithy codecs are responsible for serializing and deserializing shapes in a
    particular format.
    """

    @property
    def media_type(self) -> str:
        """The media type that the codec supports."""
        ...

    def create_serializer(self, sink: BytesWriter) -> "ShapeSerializer":
        """Create a serializer that writes to the given bytes writer.

        :param sink: The output class to write to.
        :returns: A serializer that will write to the given output.
        """
        ...

    def create_deserializer(self, source: bytes | BytesReader) -> "ShapeDeserializer":
        """Create a deserializer that reads from the given bytes reader.

        :param source: The source to read bytes from.
        :returns: A deserializer that reads from the given source.
        """
        ...

    def serialize(self, shape: "SerializeableShape") -> bytes:
        """Serialize a shape to bytes.

        :param shape: The shape to serialize.
        :returns: Bytes representing the shape serialized in the codec's media type.
        """
        stream = BytesIO()
        serializer = self.create_serializer(sink=stream)
        shape.serialize(serializer=serializer)
        serializer.flush()
        stream.seek(0)
        return stream.read()

    def deserialize[S: DeserializeableShape](
        self, source: bytes | BytesReader, shape: type[S]
    ) -> S:
        """Deserialize bytes into a shape.

        :param source: The bytes to deserialize.
        :param shape: The shape class to deserialize into.
        :returns: An instance of the given shape class with the data from the source.
        """
        deserializer = self.create_deserializer(source=source)
        return shape.deserialize(deserializer=deserializer)

media_type property

The media type that the codec supports.

create_deserializer(source)

Create a deserializer that reads from the given bytes reader.

:param source: The source to read bytes from. :returns: A deserializer that reads from the given source.

Source code in packages/smithy-core/src/smithy_core/codecs.py

def create_deserializer(self, source: bytes | BytesReader) -> "ShapeDeserializer":
    """Create a deserializer that reads from the given bytes reader.

    :param source: The source to read bytes from.
    :returns: A deserializer that reads from the given source.
    """
    ...

create_serializer(sink)

Create a serializer that writes to the given bytes writer.

:param sink: The output class to write to. :returns: A serializer that will write to the given output.

Source code in packages/smithy-core/src/smithy_core/codecs.py

def create_serializer(self, sink: BytesWriter) -> "ShapeSerializer":
    """Create a serializer that writes to the given bytes writer.

    :param sink: The output class to write to.
    :returns: A serializer that will write to the given output.
    """
    ...

deserialize(source, shape)

Deserialize bytes into a shape.

:param source: The bytes to deserialize. :param shape: The shape class to deserialize into. :returns: An instance of the given shape class with the data from the source.

Source code in packages/smithy-core/src/smithy_core/codecs.py

def deserialize[S: DeserializeableShape](
    self, source: bytes | BytesReader, shape: type[S]
) -> S:
    """Deserialize bytes into a shape.

    :param source: The bytes to deserialize.
    :param shape: The shape class to deserialize into.
    :returns: An instance of the given shape class with the data from the source.
    """
    deserializer = self.create_deserializer(source=source)
    return shape.deserialize(deserializer=deserializer)

serialize(shape)

Serialize a shape to bytes.

:param shape: The shape to serialize. :returns: Bytes representing the shape serialized in the codec's media type.

Source code in packages/smithy-core/src/smithy_core/codecs.py

def serialize(self, shape: "SerializeableShape") -> bytes:
    """Serialize a shape to bytes.

    :param shape: The shape to serialize.
    :returns: Bytes representing the shape serialized in the codec's media type.
    """
    stream = BytesIO()
    serializer = self.create_serializer(sink=stream)
    shape.serialize(serializer=serializer)
    serializer.flush()
    stream.seek(0)
    return stream.read()

deserializers

DeserializeableShape

Bases: Protocol

Protocol for shapes that are deserializeable using a ShapeDeserializer.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

@runtime_checkable
class DeserializeableShape(Protocol):
    """Protocol for shapes that are deserializeable using a ShapeDeserializer."""

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        """Construct an instance of this class using the given deserializer.

        :param deserializer: The deserializer to read from.
        :returns: An instance of this class created from the deserializer.
        """
        ...

deserialize(deserializer) classmethod

Construct an instance of this class using the given deserializer.

:param deserializer: The deserializer to read from. :returns: An instance of this class created from the deserializer.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

@classmethod
def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
    """Construct an instance of this class using the given deserializer.

    :param deserializer: The deserializer to read from.
    :returns: An instance of this class created from the deserializer.
    """
    ...

ShapeDeserializer

Bases: Protocol

Protocol used for deserializing shapes based on the Smithy data model.

If used as a base class, all non-float number methods default to calling read_integer and read_double defaults to calling read_float. These extra numeric methods are for types in the Smithy data model that don't have Python equivalents, but may have equivalents in the format being read.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

@runtime_checkable
class ShapeDeserializer(Protocol):
    """Protocol used for deserializing shapes based on the Smithy data model.

    If used as a base class, all non-float number methods default to calling
    ``read_integer`` and ``read_double`` defaults to calling ``read_float``.
    These extra numeric methods are for types in the Smithy data model that
    don't have Python equivalents, but may have equivalents in the format
    being read.
    """

    def read_struct(
        self,
        schema: "Schema",
        consumer: Callable[["Schema", "ShapeDeserializer"], None],
    ) -> None:
        """Read a struct value from the underlying data.

        :param schema: The shape's schema.
        :param consumer: A callable to read struct members with.
        """
        ...

    def read_list(
        self, schema: "Schema", consumer: Callable[["ShapeDeserializer"], None]
    ) -> None:
        """Read a list value from the underlying data.

        :param schema: The shape's schema.
        :param consumer: A callable to read list elements with.
        """
        ...

    def read_map(
        self,
        schema: "Schema",
        consumer: Callable[[str, "ShapeDeserializer"], None],
    ) -> None:
        """Read a map value from the underlying data.

        :param schema: The shape's schema.
        :param consumer: A callable to read map values with.
        """
        ...

    def is_null(self) -> bool:
        """Returns whether the next value in the underlying data represents null.

        :param schema: The shape's schema.
        """
        ...

    def read_null(self) -> None:
        """Read a null value from the underlying data."""
        ...

    def read_boolean(self, schema: "Schema") -> bool:
        """Read a boolean value from the underlying data.

        :param schema: The shape's schema.
        :returns: A bool from the underlying data.
        """
        ...

    def read_blob(self, schema: "Schema") -> bytes:
        """Read a blob value from the underlying data.

        :param schema: The shape's schema.
        :returns: A blob from the underlying data.
        """
        ...

    def read_byte(self, schema: "Schema") -> int:
        """Read a byte (8-bit integer) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A byte from the underlying data.
        """
        return self.read_integer(schema)

    def read_short(self, schema: "Schema") -> int:
        """Read a short (16-bit integer) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A short from the underlying data.
        """
        return self.read_integer(schema)

    def read_integer(self, schema: "Schema") -> int:
        """Read an integer (32-bit) value from the underlying data.

        :param schema: The shape's schema.
        :returns: An integer from the underlying data.
        """
        ...

    def read_long(self, schema: "Schema") -> int:
        """Read a long (64-bit integer) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A long from the underlying data.
        """
        return self.read_integer(schema)

    def read_float(self, schema: "Schema") -> float:
        """Read a float (32-bit) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A float from the underlying data.
        """
        ...

    def read_double(self, schema: "Schema") -> float:
        """Read a double (64-bit float) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A double from the underlying data.
        """
        return self.read_float(schema)

    def read_big_integer(self, schema: "Schema") -> int:
        """Read a big integer (arbitrarily large integer) value from the underlying
        data.

        :param schema: The shape's schema.
        :returns: A big integer from the underlying data.
        """
        return self.read_integer(schema)

    def read_big_decimal(self, schema: "Schema") -> Decimal:
        """Read a big decimal (arbitrarily large float) value from the underlying data.

        :param schema: The shape's schema.
        :returns: A big decimal from the underlying data.
        """
        ...

    def read_string(self, schema: "Schema") -> str:
        """Read a string value from the underlying data.

        :param schema: The shape's schema.
        :returns: A string from the underlying data.
        """
        ...

    def read_document(self, schema: "Schema") -> "Document":
        """Read a document value from the underlying data.

        :param schema: The shape's schema.
        :returns: A document from the underlying data.
        """
        ...

    def read_timestamp(self, schema: "Schema") -> datetime.datetime:
        """Read a timestamp value from the underlying data.

        :param schema: The shape's schema.
        :returns: A timestamp from the underlying data.
        """
        ...

    def read_data_stream(self, schema: "Schema") -> "_Stream":
        """Read a data stream from the underlying data.

        The data itself MUST NOT be read by this method. The value returned is intended
        to be read later by the consumer. In an HTTP implementation, for example, this
        would directly return the HTTP body stream. The stream MAY be wrapped to provide
        a more consistent interface or to avoid exposing implementation details.

        Data streams are only supported at the top-level input and output for
        operations.

        :param schema: The shape's schema.
        :returns: A data stream derived from the underlying data.
        """
        raise UnsupportedStreamError()

is_null()

Returns whether the next value in the underlying data represents null.

:param schema: The shape's schema.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def is_null(self) -> bool:
    """Returns whether the next value in the underlying data represents null.

    :param schema: The shape's schema.
    """
    ...

read_big_decimal(schema)

Read a big decimal (arbitrarily large float) value from the underlying data.

:param schema: The shape's schema. :returns: A big decimal from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_big_decimal(self, schema: "Schema") -> Decimal:
    """Read a big decimal (arbitrarily large float) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A big decimal from the underlying data.
    """
    ...

read_big_integer(schema)

Read a big integer (arbitrarily large integer) value from the underlying data.

:param schema: The shape's schema. :returns: A big integer from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_big_integer(self, schema: "Schema") -> int:
    """Read a big integer (arbitrarily large integer) value from the underlying
    data.

    :param schema: The shape's schema.
    :returns: A big integer from the underlying data.
    """
    return self.read_integer(schema)

read_blob(schema)

Read a blob value from the underlying data.

:param schema: The shape's schema. :returns: A blob from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_blob(self, schema: "Schema") -> bytes:
    """Read a blob value from the underlying data.

    :param schema: The shape's schema.
    :returns: A blob from the underlying data.
    """
    ...

read_boolean(schema)

Read a boolean value from the underlying data.

:param schema: The shape's schema. :returns: A bool from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_boolean(self, schema: "Schema") -> bool:
    """Read a boolean value from the underlying data.

    :param schema: The shape's schema.
    :returns: A bool from the underlying data.
    """
    ...

read_byte(schema)

Read a byte (8-bit integer) value from the underlying data.

:param schema: The shape's schema. :returns: A byte from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_byte(self, schema: "Schema") -> int:
    """Read a byte (8-bit integer) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A byte from the underlying data.
    """
    return self.read_integer(schema)

read_data_stream(schema)

Read a data stream from the underlying data.

The data itself MUST NOT be read by this method. The value returned is intended to be read later by the consumer. In an HTTP implementation, for example, this would directly return the HTTP body stream. The stream MAY be wrapped to provide a more consistent interface or to avoid exposing implementation details.

Data streams are only supported at the top-level input and output for operations.

:param schema: The shape's schema. :returns: A data stream derived from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_data_stream(self, schema: "Schema") -> "_Stream":
    """Read a data stream from the underlying data.

    The data itself MUST NOT be read by this method. The value returned is intended
    to be read later by the consumer. In an HTTP implementation, for example, this
    would directly return the HTTP body stream. The stream MAY be wrapped to provide
    a more consistent interface or to avoid exposing implementation details.

    Data streams are only supported at the top-level input and output for
    operations.

    :param schema: The shape's schema.
    :returns: A data stream derived from the underlying data.
    """
    raise UnsupportedStreamError()

read_document(schema)

Read a document value from the underlying data.

:param schema: The shape's schema. :returns: A document from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_document(self, schema: "Schema") -> "Document":
    """Read a document value from the underlying data.

    :param schema: The shape's schema.
    :returns: A document from the underlying data.
    """
    ...

read_double(schema)

Read a double (64-bit float) value from the underlying data.

:param schema: The shape's schema. :returns: A double from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_double(self, schema: "Schema") -> float:
    """Read a double (64-bit float) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A double from the underlying data.
    """
    return self.read_float(schema)

read_float(schema)

Read a float (32-bit) value from the underlying data.

:param schema: The shape's schema. :returns: A float from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_float(self, schema: "Schema") -> float:
    """Read a float (32-bit) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A float from the underlying data.
    """
    ...

read_integer(schema)

Read an integer (32-bit) value from the underlying data.

:param schema: The shape's schema. :returns: An integer from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_integer(self, schema: "Schema") -> int:
    """Read an integer (32-bit) value from the underlying data.

    :param schema: The shape's schema.
    :returns: An integer from the underlying data.
    """
    ...

read_list(schema, consumer)

Read a list value from the underlying data.

:param schema: The shape's schema. :param consumer: A callable to read list elements with.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_list(
    self, schema: "Schema", consumer: Callable[["ShapeDeserializer"], None]
) -> None:
    """Read a list value from the underlying data.

    :param schema: The shape's schema.
    :param consumer: A callable to read list elements with.
    """
    ...

read_long(schema)

Read a long (64-bit integer) value from the underlying data.

:param schema: The shape's schema. :returns: A long from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_long(self, schema: "Schema") -> int:
    """Read a long (64-bit integer) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A long from the underlying data.
    """
    return self.read_integer(schema)

read_map(schema, consumer)

Read a map value from the underlying data.

:param schema: The shape's schema. :param consumer: A callable to read map values with.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_map(
    self,
    schema: "Schema",
    consumer: Callable[[str, "ShapeDeserializer"], None],
) -> None:
    """Read a map value from the underlying data.

    :param schema: The shape's schema.
    :param consumer: A callable to read map values with.
    """
    ...

read_null()

Read a null value from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_null(self) -> None:
    """Read a null value from the underlying data."""
    ...

read_short(schema)

Read a short (16-bit integer) value from the underlying data.

:param schema: The shape's schema. :returns: A short from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_short(self, schema: "Schema") -> int:
    """Read a short (16-bit integer) value from the underlying data.

    :param schema: The shape's schema.
    :returns: A short from the underlying data.
    """
    return self.read_integer(schema)

read_string(schema)

Read a string value from the underlying data.

:param schema: The shape's schema. :returns: A string from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_string(self, schema: "Schema") -> str:
    """Read a string value from the underlying data.

    :param schema: The shape's schema.
    :returns: A string from the underlying data.
    """
    ...

read_struct(schema, consumer)

Read a struct value from the underlying data.

:param schema: The shape's schema. :param consumer: A callable to read struct members with.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_struct(
    self,
    schema: "Schema",
    consumer: Callable[["Schema", "ShapeDeserializer"], None],
) -> None:
    """Read a struct value from the underlying data.

    :param schema: The shape's schema.
    :param consumer: A callable to read struct members with.
    """
    ...

read_timestamp(schema)

Read a timestamp value from the underlying data.

:param schema: The shape's schema. :returns: A timestamp from the underlying data.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

def read_timestamp(self, schema: "Schema") -> datetime.datetime:
    """Read a timestamp value from the underlying data.

    :param schema: The shape's schema.
    :returns: A timestamp from the underlying data.
    """
    ...

SpecificShapeDeserializer

Bases: ShapeDeserializer

Expects to deserialize a specific kind of shape, failing if other shapes are deserialized.

Source code in packages/smithy-core/src/smithy_core/deserializers.py

class SpecificShapeDeserializer(ShapeDeserializer):
    """Expects to deserialize a specific kind of shape, failing if other shapes are
    deserialized."""

    def _invalid_state(
        self, schema: "Schema | None" = None, message: str | None = None
    ) -> Never:
        if message is None:
            message = f"Unexpected schema type: {schema}"
        raise SmithyError(message)

    def read_struct(
        self,
        schema: "Schema",
        consumer: Callable[["Schema", "ShapeDeserializer"], None],
    ) -> None:
        self._invalid_state(schema)

    def read_list(
        self, schema: "Schema", consumer: Callable[["ShapeDeserializer"], None]
    ) -> None:
        self._invalid_state(schema)

    def read_map(
        self,
        schema: "Schema",
        consumer: Callable[[str, "ShapeDeserializer"], None],
    ) -> None:
        self._invalid_state(schema)

    def is_null(self) -> bool:
        self._invalid_state(message="Unexpected attempt to read null.")

    def read_null(self) -> None:
        self._invalid_state(message="Unexpected attempt to read null.")

    def read_boolean(self, schema: "Schema") -> bool:
        self._invalid_state(schema)

    def read_blob(self, schema: "Schema") -> bytes:
        self._invalid_state(schema)

    def read_byte(self, schema: "Schema") -> int:
        self._invalid_state(schema)

    def read_short(self, schema: "Schema") -> int:
        self._invalid_state(schema)

    def read_integer(self, schema: "Schema") -> int:
        self._invalid_state(schema)

    def read_long(self, schema: "Schema") -> int:
        self._invalid_state(schema)

    def read_float(self, schema: "Schema") -> float:
        self._invalid_state(schema)

    def read_double(self, schema: "Schema") -> float:
        self._invalid_state(schema)

    def read_big_integer(self, schema: "Schema") -> int:
        self._invalid_state(schema)

    def read_big_decimal(self, schema: "Schema") -> Decimal:
        self._invalid_state(schema)

    def read_string(self, schema: "Schema") -> str:
        self._invalid_state(schema)

    def read_document(self, schema: "Schema") -> "Document":
        self._invalid_state(schema)

    def read_timestamp(self, schema: "Schema") -> datetime.datetime:
        self._invalid_state(schema)

    def read_data_stream(self, schema: "Schema") -> "_Stream":
        self._invalid_state(schema)

documents

DocumentValue = Mapping[str, DocumentValue] | Sequence[DocumentValue] | str | int | float | Decimal | bool | None | bytes | datetime.datetime

Protocol-agnostic open content.

Document

Wrapped protocol-agnostic open content.

This wrapping class facilitates serialization and deserialiazation and may contain a schema to enable serializing and deserializing protocol-specific documents into a protocol-agnostic form.

Source code in packages/smithy-core/src/smithy_core/documents.py

class Document:
    """Wrapped protocol-agnostic open content.

    This wrapping class facilitates serialization and deserialiazation and may contain a
    schema to enable serializing and deserializing protocol-specific documents into a
    protocol-agnostic form.
    """

    _value: _InnerDocumentValue = None
    _raw_value: Mapping[str, DocumentValue] | Sequence[DocumentValue] | None = None
    _type: ShapeType
    _schema: Schema

    def __init__(
        self,
        value: DocumentValue | dict[str, "Document"] | list["Document"] = None,
        *,
        schema: Schema = _DOCUMENT,
    ) -> None:
        """Initializes a document.

        :param value: The underlying value of a document.
        :param schema: A schema defining the document's structure. The default value is
            a plain document schema with no traits.
        """
        self._schema = schema

        # Mappings and Sequences are lazily converted to/from the inner value type.
        if isinstance(value, Mapping):
            if self._is_raw_map(value):
                self._raw_value = value
            else:
                self._value = value  # type: ignore
        elif isinstance(value, Sequence) and not isinstance(value, str | bytes):
            if self._is_raw_list(value):
                self._raw_value = value
            else:
                self._value = value  # type: ignore
        else:
            self._value = value

        if self._schema.shape_type not in (
            ShapeType.DOCUMENT,
            ShapeType.OPERATION,
            ShapeType.SERVICE,
        ):
            self._type = self._schema.shape_type
        else:
            # TODO: set an appropriate schema if one was not provided
            match value:
                case bool():
                    self._type = ShapeType.BOOLEAN
                case str():
                    self._type = ShapeType.STRING
                case int():
                    self._type = ShapeType.LONG
                case float():
                    self._type = ShapeType.DOUBLE
                case Decimal():
                    self._type = ShapeType.BIG_DECIMAL
                case bytes():
                    self._type = ShapeType.BLOB
                case datetime.datetime():
                    self._type = ShapeType.TIMESTAMP
                case Sequence():
                    self._type = ShapeType.LIST
                case Mapping():
                    self._type = ShapeType.MAP
                case _:
                    self._type = ShapeType.DOCUMENT

    def _is_raw_map(
        self, value: Mapping[str, "Document"] | Mapping[str, DocumentValue]
    ) -> TypeGuard[Mapping[str, DocumentValue]]:
        return len(value) != 0 and not isinstance(next(iter(value.values())), Document)

    def _is_raw_list(
        self, value: Sequence["Document"] | Sequence[DocumentValue]
    ) -> TypeGuard[Sequence[DocumentValue]]:
        return (
            len(value) != 0
            and not isinstance(value, str | bytes)
            and not isinstance(next(iter(value)), Document)
        )

    @property
    def shape_type(self) -> ShapeType:
        """The Smithy data model type for the underlying contents of the document."""
        return self._type

    @property
    def discriminator(self) -> ShapeID:
        """The shape ID that corresponds to the contents of the document."""
        if self._type is ShapeType.STRUCTURE:
            return self._schema.id
        raise DiscriminatorError(f"{self._type} document has no discriminator.")

    def is_none(self) -> bool:
        """Indicates whether the document contains a null value."""
        return self._value is None and self._raw_value is None

    def as_blob(self) -> bytes:
        """Asserts the document is a blob and returns it as bytes."""
        return expect_type(bytes, self._value)

    def as_boolean(self) -> bool:
        """Asserts the document is a boolean and returns it."""
        return expect_type(bool, self._value)

    def as_string(self) -> str:
        """Asserts the document is a string and returns it."""
        return expect_type(str, self._value)

    def as_timestamp(self) -> datetime.datetime:
        """Asserts the document is a timestamp and returns it."""
        return expect_type(datetime.datetime, self._value)

    def as_integer(self) -> int:
        """Asserts the document is an integer and returns it.

        This method is to be used for any type from the Smithy data model that
        translates to Python's int type. This includes: byte, short, integer, long, and
        bigInteger.
        """
        if isinstance(self._value, int) and not isinstance(self._value, bool):
            return self._value
        raise ExpectationNotMetError(
            f"Expected int, found {type(self._value)}: {self._value}"
        )

    def as_float(self) -> float:
        """Asserts the document is a float and returns it.

        This method is to be used for any type from the Smithy data model that
        translates to Python's float type. This includes float and double.
        """
        return expect_type(float, self._value)

    def as_decimal(self) -> Decimal:
        """Asserts the document is a Decimal and returns it."""
        return expect_type(Decimal, self._value)

    def as_list(self) -> list["Document"]:
        """Asserts the document is a list and returns it."""
        if isinstance(self._value, list):
            return self._value
        if (
            self._value is None
            and isinstance(self._raw_value, Sequence)
            and not isinstance(self._raw_value, str | bytes)
        ):
            self._value = self._wrap_list(self._raw_value)
            return self._value
        raise ExpectationNotMetError(
            f"Expected list, found {type(self._value)}: {self._value}"
        )

    def _wrap_list(self, value: Sequence[DocumentValue]) -> list["Document"]:
        schema = self._schema
        if schema.shape_type is ShapeType.LIST:
            schema = self._schema.members["member"]
        return [self._new_document(e, schema) for e in value]

    def _new_document(
        self,
        value: DocumentValue | dict[str, "Document"] | list["Document"],
        schema: Schema,
    ) -> "Document":
        return Document(value, schema=schema)

    def as_map(self) -> dict[str, "Document"]:
        """Asserts the document is a map and returns it."""
        if isinstance(self._value, Mapping):
            return self._value
        if self._value is None and isinstance(self._raw_value, Mapping):
            self._value = self._wrap_map(self._raw_value)
            return self._value
        raise ExpectationNotMetError(
            f"Expected map, found {type(self._value)}: {self._value}"
        )

    def _wrap_map(self, value: Mapping[str, DocumentValue]) -> dict[str, "Document"]:
        if self._schema.shape_type not in (ShapeType.STRUCTURE, ShapeType.UNION):
            member_schema = self._schema
            if self._schema.shape_type is ShapeType.MAP:
                member_schema = self._schema.members["value"]
            return {k: self._new_document(v, member_schema) for k, v in value.items()}

        result: dict[str, Document] = {}
        for k, v in value.items():
            result[k] = self._new_document(v, self._schema.members[k])
        return result

    def as_value(self) -> DocumentValue:
        """Converts the document to a plain, protocol-agnostic DocumentValue and returns
        it."""
        if self._value is not None and self._raw_value is None:
            match self._value:
                case dict():
                    self._raw_value = {k: v.as_value() for k, v in self._value.items()}
                case list():
                    self._raw_value = [e.as_value() for e in self._value]
                case _:
                    return self._value
        return self._raw_value

    def as_shape[S: DeserializeableShape](self, shape_class: type[S]) -> S:
        """Converts the document to an instance of the given shape type.

        :param shape_class: A class that implements the DeserializeableShape protocol.
        """
        return shape_class.deserialize(_DocumentDeserializer(self))

    def serialize(self, serializer: ShapeSerializer) -> None:
        serializer.write_document(self._schema, self)

    def serialize_contents(self, serializer: ShapeSerializer) -> None:
        if self.is_none():
            serializer.write_null(self._schema)
            return

        match self._type:
            case ShapeType.STRUCTURE | ShapeType.UNION:
                with serializer.begin_struct(self._schema) as struct_serializer:
                    self.serialize_members(struct_serializer)
            case ShapeType.LIST:
                list_value = self.as_list()
                with serializer.begin_list(
                    self._schema, len(list_value)
                ) as list_serializer:
                    for element in list_value:
                        element.serialize(list_serializer)
            case ShapeType.MAP:
                map_value = self.as_map()
                with serializer.begin_map(
                    self._schema, len(map_value)
                ) as map_serializer:
                    for key, value in map_value.items():
                        map_serializer.entry(key, lambda s: value.serialize(s))
            case ShapeType.STRING | ShapeType.ENUM:
                serializer.write_string(self._schema, self.as_string())
            case ShapeType.BOOLEAN:
                serializer.write_boolean(self._schema, self.as_boolean())
            case ShapeType.BYTE:
                serializer.write_byte(self._schema, self.as_integer())
            case ShapeType.SHORT:
                serializer.write_short(self._schema, self.as_integer())
            case ShapeType.INTEGER | ShapeType.INT_ENUM:
                serializer.write_integer(self._schema, self.as_integer())
            case ShapeType.LONG:
                serializer.write_long(self._schema, self.as_integer())
            case ShapeType.BIG_INTEGER:
                serializer.write_big_integer(self._schema, self.as_integer())
            case ShapeType.FLOAT:
                serializer.write_float(self._schema, self.as_float())
            case ShapeType.DOUBLE:
                serializer.write_double(self._schema, self.as_float())
            case ShapeType.BIG_DECIMAL:
                serializer.write_big_decimal(self._schema, self.as_decimal())
            case ShapeType.BLOB:
                serializer.write_blob(self._schema, self.as_blob())
            case ShapeType.TIMESTAMP:
                serializer.write_timestamp(self._schema, self.as_timestamp())
            case ShapeType.DOCUMENT:
                # The shape type is only ever document when the value is null,
                # which is a case we've already handled.
                raise SmithyError(
                    f"Unexpexcted DOCUMENT shape type for document value: {self.as_value()}"
                )
            case _:
                raise SmithyError(
                    f"Unexpected {self._type} shape type for document value: {self.as_value()}"
                )

    def serialize_members(self, serializer: ShapeSerializer) -> None:
        for value in self.as_map().values():
            value.serialize(serializer)

    @classmethod
    def from_shape(cls, shape: SerializeableShape) -> "Document":
        """Constructs a Document from a given shape.

        :param shape: The shape to convert to a document.
        :returns: A Document representation of the given shape.
        """
        serializer = _DocumentSerializer()
        shape.serialize(serializer=serializer)
        serializer.flush()
        return serializer.expect_result()

    def get(self, name: str, default: "Document | None" = None) -> "Document | None":
        """Gets a named member of the document or a default value."""
        return self.as_map().get(name, default)

    def __getitem__(self, key: str | int | slice) -> "Document":
        match key:
            case str():
                return self.as_map()[key]
            case int():
                return self.as_list()[key]
            case slice():
                return Document(self.as_list()[key], schema=self._schema)

    def __setitem__(
        self,
        key: str | int,
        value: "Document | list[Document] | dict[str, Document] | DocumentValue",
    ) -> None:
        if isinstance(key, str):
            schema = self._schema
            if schema.shape_type is ShapeType.STRUCTURE:
                schema = schema.members[key]
            elif schema.shape_type is ShapeType.MAP:
                schema = schema.members["value"]

            if not isinstance(value, Document):
                value = Document(value, schema=schema)
            else:
                value = value._with_schema(schema)

            self.as_map()[key] = value
        else:
            schema = self._schema
            if schema.shape_type is ShapeType.LIST:
                schema = schema.members["member"]

            if not isinstance(value, Document):
                value = Document(value, schema=schema)
            else:
                value = value._with_schema(schema)

            self.as_list()[key] = value
        self._raw_value = None

    def _with_schema(self, schema: Schema) -> "Document":
        if self.shape_type in [ShapeType.STRUCTURE, ShapeType.MAP, ShapeType.UNION]:
            return Document(self.as_map(), schema=schema)
        elif self.shape_type is ShapeType.LIST:
            return Document(self.as_list(), schema=schema)
        else:
            return Document(self._value, schema=schema)

    def __delitem__(self, key: str | int) -> None:
        if isinstance(key, str):
            del self.as_map()[key]
        else:
            del self.as_list()[key]
        self._raw_value = None

    def __repr__(self) -> str:
        value: str
        if self._value is not None:
            value = repr(self._value)
        else:
            value = repr(self._raw_value)

        if self._schema is _DOCUMENT:
            return f"Document(value={value})"
        return f"Document(value={value}, schema={self._schema})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Document):
            return False
        # Note that the schema isn't important to equality. If some subclass, such as a
        # JsonDocument, cares about the schema then it's responsible for transforming
        # the output of `as_value` to the canonical form.
        return self.as_value() == other.as_value()

discriminator property

The shape ID that corresponds to the contents of the document.

shape_type property

The Smithy data model type for the underlying contents of the document.

__init__(value=None, *, schema=_DOCUMENT)

Initializes a document.

:param value: The underlying value of a document. :param schema: A schema defining the document's structure. The default value is a plain document schema with no traits.

Source code in packages/smithy-core/src/smithy_core/documents.py

def __init__(
    self,
    value: DocumentValue | dict[str, "Document"] | list["Document"] = None,
    *,
    schema: Schema = _DOCUMENT,
) -> None:
    """Initializes a document.

    :param value: The underlying value of a document.
    :param schema: A schema defining the document's structure. The default value is
        a plain document schema with no traits.
    """
    self._schema = schema

    # Mappings and Sequences are lazily converted to/from the inner value type.
    if isinstance(value, Mapping):
        if self._is_raw_map(value):
            self._raw_value = value
        else:
            self._value = value  # type: ignore
    elif isinstance(value, Sequence) and not isinstance(value, str | bytes):
        if self._is_raw_list(value):
            self._raw_value = value
        else:
            self._value = value  # type: ignore
    else:
        self._value = value

    if self._schema.shape_type not in (
        ShapeType.DOCUMENT,
        ShapeType.OPERATION,
        ShapeType.SERVICE,
    ):
        self._type = self._schema.shape_type
    else:
        # TODO: set an appropriate schema if one was not provided
        match value:
            case bool():
                self._type = ShapeType.BOOLEAN
            case str():
                self._type = ShapeType.STRING
            case int():
                self._type = ShapeType.LONG
            case float():
                self._type = ShapeType.DOUBLE
            case Decimal():
                self._type = ShapeType.BIG_DECIMAL
            case bytes():
                self._type = ShapeType.BLOB
            case datetime.datetime():
                self._type = ShapeType.TIMESTAMP
            case Sequence():
                self._type = ShapeType.LIST
            case Mapping():
                self._type = ShapeType.MAP
            case _:
                self._type = ShapeType.DOCUMENT

as_blob()

Asserts the document is a blob and returns it as bytes.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_blob(self) -> bytes:
    """Asserts the document is a blob and returns it as bytes."""
    return expect_type(bytes, self._value)

as_boolean()

Asserts the document is a boolean and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_boolean(self) -> bool:
    """Asserts the document is a boolean and returns it."""
    return expect_type(bool, self._value)

as_decimal()

Asserts the document is a Decimal and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_decimal(self) -> Decimal:
    """Asserts the document is a Decimal and returns it."""
    return expect_type(Decimal, self._value)

as_float()

Asserts the document is a float and returns it.

This method is to be used for any type from the Smithy data model that translates to Python's float type. This includes float and double.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_float(self) -> float:
    """Asserts the document is a float and returns it.

    This method is to be used for any type from the Smithy data model that
    translates to Python's float type. This includes float and double.
    """
    return expect_type(float, self._value)

as_integer()

Asserts the document is an integer and returns it.

This method is to be used for any type from the Smithy data model that translates to Python's int type. This includes: byte, short, integer, long, and bigInteger.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_integer(self) -> int:
    """Asserts the document is an integer and returns it.

    This method is to be used for any type from the Smithy data model that
    translates to Python's int type. This includes: byte, short, integer, long, and
    bigInteger.
    """
    if isinstance(self._value, int) and not isinstance(self._value, bool):
        return self._value
    raise ExpectationNotMetError(
        f"Expected int, found {type(self._value)}: {self._value}"
    )

as_list()

Asserts the document is a list and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_list(self) -> list["Document"]:
    """Asserts the document is a list and returns it."""
    if isinstance(self._value, list):
        return self._value
    if (
        self._value is None
        and isinstance(self._raw_value, Sequence)
        and not isinstance(self._raw_value, str | bytes)
    ):
        self._value = self._wrap_list(self._raw_value)
        return self._value
    raise ExpectationNotMetError(
        f"Expected list, found {type(self._value)}: {self._value}"
    )

as_map()

Asserts the document is a map and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_map(self) -> dict[str, "Document"]:
    """Asserts the document is a map and returns it."""
    if isinstance(self._value, Mapping):
        return self._value
    if self._value is None and isinstance(self._raw_value, Mapping):
        self._value = self._wrap_map(self._raw_value)
        return self._value
    raise ExpectationNotMetError(
        f"Expected map, found {type(self._value)}: {self._value}"
    )

as_shape(shape_class)

Converts the document to an instance of the given shape type.

:param shape_class: A class that implements the DeserializeableShape protocol.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_shape[S: DeserializeableShape](self, shape_class: type[S]) -> S:
    """Converts the document to an instance of the given shape type.

    :param shape_class: A class that implements the DeserializeableShape protocol.
    """
    return shape_class.deserialize(_DocumentDeserializer(self))

as_string()

Asserts the document is a string and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_string(self) -> str:
    """Asserts the document is a string and returns it."""
    return expect_type(str, self._value)

as_timestamp()

Asserts the document is a timestamp and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_timestamp(self) -> datetime.datetime:
    """Asserts the document is a timestamp and returns it."""
    return expect_type(datetime.datetime, self._value)

as_value()

Converts the document to a plain, protocol-agnostic DocumentValue and returns it.

Source code in packages/smithy-core/src/smithy_core/documents.py

def as_value(self) -> DocumentValue:
    """Converts the document to a plain, protocol-agnostic DocumentValue and returns
    it."""
    if self._value is not None and self._raw_value is None:
        match self._value:
            case dict():
                self._raw_value = {k: v.as_value() for k, v in self._value.items()}
            case list():
                self._raw_value = [e.as_value() for e in self._value]
            case _:
                return self._value
    return self._raw_value

from_shape(shape) classmethod

Constructs a Document from a given shape.

:param shape: The shape to convert to a document. :returns: A Document representation of the given shape.

Source code in packages/smithy-core/src/smithy_core/documents.py

@classmethod
def from_shape(cls, shape: SerializeableShape) -> "Document":
    """Constructs a Document from a given shape.

    :param shape: The shape to convert to a document.
    :returns: A Document representation of the given shape.
    """
    serializer = _DocumentSerializer()
    shape.serialize(serializer=serializer)
    serializer.flush()
    return serializer.expect_result()

get(name, default=None)

Gets a named member of the document or a default value.

Source code in packages/smithy-core/src/smithy_core/documents.py

def get(self, name: str, default: "Document | None" = None) -> "Document | None":
    """Gets a named member of the document or a default value."""
    return self.as_map().get(name, default)

is_none()

Indicates whether the document contains a null value.

Source code in packages/smithy-core/src/smithy_core/documents.py

def is_none(self) -> bool:
    """Indicates whether the document contains a null value."""
    return self._value is None and self._raw_value is None

TypeRegistry

A registry for on-demand deserialization of types by using a mapping of shape IDs to their deserializers.

Source code in packages/smithy-core/src/smithy_core/documents.py

class TypeRegistry:
    """A registry for on-demand deserialization of types by using a mapping of shape IDs
    to their deserializers."""

    def __init__(
        self,
        types: dict[ShapeID, type[DeserializeableShape]],
        sub_registry: "TypeRegistry | None" = None,
    ):
        """Initialize a TypeRegistry.

        :param types: A mapping of ShapeID to the shapes they deserialize to.
        :param sub_registry: A registry to delegate to if an ID is not found in types.
        """
        self._types = types
        self._sub_registry = sub_registry

    def get(self, shape: ShapeID) -> type[DeserializeableShape]:
        """Get the deserializable shape for the given shape ID.

        :param shape: The shape ID to get from the registry.
        :returns: The corresponding deserializable shape.
        :raises KeyError: If the shape ID is not found in the registry.
        """
        if shape in self._types:
            return self._types[shape]
        if self._sub_registry is not None:
            return self._sub_registry.get(shape)
        raise KeyError(f"Unknown shape: {shape}")

    def __getitem__(self, shape: ShapeID):
        """Get the deserializable shape for the given shape ID.

        :param shape: The shape ID to get from the registry.
        :returns: The corresponding deserializable shape.
        :raises KeyError: If the shape ID is not found in the registry.
        """
        return self.get(shape)

    def __contains__(self, item: object, /):
        """Check if the registry contains the given shape.

        :param item: The shape ID to check for.
        """
        return item in self._types or (
            self._sub_registry is not None and item in self._sub_registry
        )

    def deserialize(self, document: Document) -> DeserializeableShape:
        return document.as_shape(self.get(document.discriminator))

__contains__(item)

Check if the registry contains the given shape.

:param item: The shape ID to check for.

Source code in packages/smithy-core/src/smithy_core/documents.py

def __contains__(self, item: object, /):
    """Check if the registry contains the given shape.

    :param item: The shape ID to check for.
    """
    return item in self._types or (
        self._sub_registry is not None and item in self._sub_registry
    )

__getitem__(shape)

Get the deserializable shape for the given shape ID.

:param shape: The shape ID to get from the registry. :returns: The corresponding deserializable shape. :raises KeyError: If the shape ID is not found in the registry.

Source code in packages/smithy-core/src/smithy_core/documents.py

def __getitem__(self, shape: ShapeID):
    """Get the deserializable shape for the given shape ID.

    :param shape: The shape ID to get from the registry.
    :returns: The corresponding deserializable shape.
    :raises KeyError: If the shape ID is not found in the registry.
    """
    return self.get(shape)

__init__(types, sub_registry=None)

Initialize a TypeRegistry.

:param types: A mapping of ShapeID to the shapes they deserialize to. :param sub_registry: A registry to delegate to if an ID is not found in types.

Source code in packages/smithy-core/src/smithy_core/documents.py

def __init__(
    self,
    types: dict[ShapeID, type[DeserializeableShape]],
    sub_registry: "TypeRegistry | None" = None,
):
    """Initialize a TypeRegistry.

    :param types: A mapping of ShapeID to the shapes they deserialize to.
    :param sub_registry: A registry to delegate to if an ID is not found in types.
    """
    self._types = types
    self._sub_registry = sub_registry

get(shape)

Get the deserializable shape for the given shape ID.

:param shape: The shape ID to get from the registry. :returns: The corresponding deserializable shape. :raises KeyError: If the shape ID is not found in the registry.

Source code in packages/smithy-core/src/smithy_core/documents.py

def get(self, shape: ShapeID) -> type[DeserializeableShape]:
    """Get the deserializable shape for the given shape ID.

    :param shape: The shape ID to get from the registry.
    :returns: The corresponding deserializable shape.
    :raises KeyError: If the shape ID is not found in the registry.
    """
    if shape in self._types:
        return self._types[shape]
    if self._sub_registry is not None:
        return self._sub_registry.get(shape)
    raise KeyError(f"Unknown shape: {shape}")

endpoints

STATIC_ENDPOINT_CONFIG = PropertyKey(key='config', value_type=StaticEndpointConfig) module-attribute

Property containing a config that has a static endpoint.

Endpoint dataclass

Bases: Endpoint

A resolved endpoint.

Source code in packages/smithy-core/src/smithy_core/endpoints.py

@dataclass(kw_only=True)
class Endpoint(_Endpoint):
    """A resolved endpoint."""

    uri: _URI
    """The endpoint URI."""

    properties: _TypedProperties = field(default_factory=TypedProperties)
    """Properties required to interact with the endpoint.

    For example, in some AWS use cases this might contain HTTP headers to add to each
    request.
    """

properties = field(default_factory=TypedProperties) class-attribute instance-attribute

Properties required to interact with the endpoint.

For example, in some AWS use cases this might contain HTTP headers to add to each request.

uri instance-attribute

The endpoint URI.

EndpointResolverParams dataclass

Parameters passed into an Endpoint Resolver's resolve_endpoint method.

Source code in packages/smithy-core/src/smithy_core/endpoints.py

@dataclass(kw_only=True)
class EndpointResolverParams[I: SerializeableShape]:
    """Parameters passed into an Endpoint Resolver's resolve_endpoint method."""

    operation: APIOperation[I, Any] = field(repr=False)
    """The operation to resolve an endpoint for."""

    input: I
    """The input to the operation."""

    context: _TypedProperties
    """The context of the operation invocation."""

context instance-attribute

The context of the operation invocation.

input instance-attribute

The input to the operation.

operation = field(repr=False) class-attribute instance-attribute

The operation to resolve an endpoint for.

StaticEndpointConfig

Bases: Protocol

A config that has a static endpoint.

Source code in packages/smithy-core/src/smithy_core/endpoints.py

class StaticEndpointConfig(Protocol):
    """A config that has a static endpoint."""

    endpoint_uri: str | URI | None
    """A static endpoint to use for the request."""

endpoint_uri instance-attribute

A static endpoint to use for the request.

resolve_static_uri(properties)

Attempt to resolve a static URI from the endpoint resolver params.

:param properties: A TypedProperties bag or EndpointResolverParams to search.

Source code in packages/smithy-core/src/smithy_core/endpoints.py

def resolve_static_uri(
    properties: _TypedProperties | EndpointResolverParams[Any],
) -> _URI | None:
    """Attempt to resolve a static URI from the endpoint resolver params.

    :param properties: A TypedProperties bag or EndpointResolverParams to search.
    """
    properties = (
        properties.context
        if isinstance(properties, EndpointResolverParams)
        else properties
    )
    static_uri_config = properties.get(STATIC_ENDPOINT_CONFIG)
    if static_uri_config is None or static_uri_config.endpoint_uri is None:
        return None

    static_uri = static_uri_config.endpoint_uri

    # If it's not a string, it's already a parsed URI so just pass it along.
    if not isinstance(static_uri, str):
        return static_uri

    parsed = urlparse(static_uri)
    if parsed.hostname is None:
        raise EndpointResolutionError(
            f"Unable to parse hostname from provided URI: {static_uri}"
        )

    return URI(
        host=parsed.hostname,
        path=parsed.path,
        scheme=parsed.scheme,
        query=parsed.query,
        port=parsed.port,
    )

exceptions

Fault = Literal['client', 'server'] | None

Whether the client or server is at fault.

If None, then there was not enough information to determine fault.

AsyncBodyError

Bases: SmithyError

Exception indicating that a request with an async body type was created in a sync context.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class AsyncBodyError(SmithyError):
    """Exception indicating that a request with an async body type was created in a sync
    context."""

CallError dataclass

Bases: SmithyError

Base exception to be used in application-level errors.

Implements :py:class:.interfaces.retries.ErrorRetryInfo.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

@dataclass(kw_only=True)
class CallError(SmithyError):
    """Base exception to be used in application-level errors.

    Implements :py:class:`.interfaces.retries.ErrorRetryInfo`.
    """

    fault: Fault = None
    """Whether the client or server is at fault.

    If None, then there was not enough information to determine fault.
    """

    message: str = field(default="", kw_only=False)
    """The message of the error."""

    is_retry_safe: bool | None = None
    """Whether the exception is safe to retry.

    A value of True does not mean a retry will occur, but rather that a retry is allowed
    to occur.

    A value of None indicates that there is not enough information available to
    determine if a retry is safe.
    """

    retry_after: float | None = None
    """The amount of time that should pass before a retry.

    Retry strategies MAY choose to wait longer.
    """

    is_throttling_error: bool = False
    """Whether the error is a throttling error."""

    def __post_init__(self):
        super().__init__(self.message)

fault = None class-attribute instance-attribute

Whether the client or server is at fault.

If None, then there was not enough information to determine fault.

is_retry_safe = None class-attribute instance-attribute

Whether the exception is safe to retry.

A value of True does not mean a retry will occur, but rather that a retry is allowed to occur.

A value of None indicates that there is not enough information available to determine if a retry is safe.

is_throttling_error = False class-attribute instance-attribute

Whether the error is a throttling error.

message = field(default='', kw_only=False) class-attribute instance-attribute

The message of the error.

retry_after = None class-attribute instance-attribute

The amount of time that should pass before a retry.

Retry strategies MAY choose to wait longer.

DiscriminatorError

Bases: SmithyError

Exception indicating something went wrong when attempting to find the discriminator in a document.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class DiscriminatorError(SmithyError):
    """Exception indicating something went wrong when attempting to find the
    discriminator in a document."""

EndpointResolutionError

Bases: SmithyError

Exception type for all exceptions raised by endpoint resolution.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class EndpointResolutionError(SmithyError):
    """Exception type for all exceptions raised by endpoint resolution."""

ExpectationNotMetError

Bases: SmithyError

Exception type for exceptions thrown by unmet assertions.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class ExpectationNotMetError(SmithyError):
    """Exception type for exceptions thrown by unmet assertions."""

MissingDependencyError

Bases: SmithyError

Exception type raised when a feature that requires a missing optional dependency is called.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class MissingDependencyError(SmithyError):
    """Exception type raised when a feature that requires a missing optional dependency
    is called."""

ModeledError dataclass

Bases: CallError

Base exception to be used for modeled errors.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

@dataclass(kw_only=True)
class ModeledError(CallError):
    """Base exception to be used for modeled errors."""

    fault: Fault = "client"

RetryError

Bases: SmithyError

Base exception type for all exceptions raised in retry strategies.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class RetryError(SmithyError):
    """Base exception type for all exceptions raised in retry strategies."""

SerializationError

Bases: SmithyError

Base exception type for exceptions raised during serialization.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class SerializationError(SmithyError):
    """Base exception type for exceptions raised during serialization."""

SmithyError

Bases: Exception

Base exception type for all exceptions raised by smithy-python.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class SmithyError(Exception):
    """Base exception type for all exceptions raised by smithy-python."""

SmithyIdentityError

Bases: SmithyError

Base exception type for all exceptions raised in identity resolution.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class SmithyIdentityError(SmithyError):
    """Base exception type for all exceptions raised in identity resolution."""

UnsupportedStreamError

Bases: SmithyError

Indicates that a serializer or deserializer's stream method was called, but data streams are not supported.

Source code in packages/smithy-core/src/smithy_core/exceptions.py

class UnsupportedStreamError(SmithyError):
    """Indicates that a serializer or deserializer's stream method was called, but data
    streams are not supported."""

interceptors

InputContext dataclass

Source code in packages/smithy-core/src/smithy_core/interceptors.py

@dataclass(kw_only=True, frozen=True, slots=True)
class InputContext[Request: SerializeableShape]:
    request: Request
    """The modeled request for the operation being invoked."""

    properties: TypedProperties
    """A typed context property bag."""

properties instance-attribute

A typed context property bag.

request instance-attribute

The modeled request for the operation being invoked.

Interceptor

Allows injecting code into the SDK's request execution pipeline.

Terminology:

• execution - An execution is one end-to-end invocation against a client.
• attempt - An attempt is an attempt at performing an execution. By default, executions are retried multiple times based on the client's retry strategy.
• hook - A hook is a single method on the interceptor, allowing injection of code into a specific part of the SDK's request execution pipeline. Hooks are either "read" hooks, which make it possible to read in-flight request or response messages, or "read/write" hooks, which make it possible to modify in-flight requests or responses.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

class Interceptor[
    Request: SerializeableShape,
    Response: DeserializeableShape,
    TransportRequest,
    TransportResponse,
]:
    """Allows injecting code into the SDK's request execution pipeline.

    Terminology:

    * execution - An execution is one end-to-end invocation against a client.
    * attempt - An attempt is an attempt at performing an execution. By default,
        executions are retried multiple times based on the client's retry strategy.
    * hook - A hook is a single method on the interceptor, allowing injection of code
        into a specific part of the SDK's request execution pipeline. Hooks are either
        "read" hooks, which make it possible to read in-flight request or response
        messages, or "read/write" hooks, which make it possible to modify in-flight
        requests or responses.
    """

    def read_before_execution(self, context: InputContext[Request]) -> None:
        """A hook called at the start of an execution, before the SDK does anything
        else.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per execution. The duration between invocation
        of this hook and `read_after_execution` is very close to full duration of the
        execution.

        The `request` of the context will always be available. Other static properties
        will be None.

        Exceptions thrown by this hook will be stored until all interceptors have had
        their `read_before_execution` invoked. Other hooks will then be skipped and
        execution will jump to `modify_before_completion` with the thrown exception as
        the `response`. If multiple `read_before_execution` methods throw exceptions,
        the latest will be used and earlier ones will be logged and dropped.
        """

    def modify_before_serialization(self, context: InputContext[Request]) -> Request:
        """A hook called before the request is serialized into a transport request.

        This method has the ability to modify and return a new request of the same
        type.

        This will ALWAYS be called once per execution, except when a failure occurs
        earlier in the request pipeline.

        The `request` of the context will always be available. This `request` may have
        been modified by earlier `modify_before_serialization` hooks, and may be
        modified further by later hooks. Other static properites will be None.

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_completion` with the thrown exception as the `response`.

        The request returned by this hook MUST be the same type of request
        message passed into this hook. If not, an exception will immediately occur.
        """
        return context.request

    def read_before_serialization(self, context: InputContext[Request]) -> None:
        """A hook called before the input message is serialized into a transport
        request.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per execution, except when a failure occurs
        earlier in the request pipeline. The duration between invocation of this hook
        and `read_after_serialization` is very close to the amount of time spent
        marshalling the request.

        The `request` of the context will always be available. Other static properties
        will be None.

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_completion` with the thrown exception as the `response`.
        """

    def read_after_serialization(
        self, context: RequestContext[Request, TransportRequest]
    ) -> None:
        """A hook called after the input message is serialized into a transport request.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per execution, except when a failure occurs
        earlier in the request pipeline. The duration between
        `read_before_serialization` and the invocation of this hook is very close to
        the amount of time spent serializing the request.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None.

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_completion` with the thrown exception as the `response`.
        """

    def modify_before_retry_loop(
        self, context: RequestContext[Request, TransportRequest]
    ) -> TransportRequest:
        """A hook called before the retry loop is entered.

        This method has the ability to modify and return a new transport request of the
        same type.

        This will always be called once per execution, except when a failure occurs
        earlier in the request pipeline.

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_completion` with the thrown exception as the `response`.

        The transport request returned by this hook MUST be the same type of request
        passed into this hook. If not, an exception will immediately occur.
        """
        return context.transport_request

    def read_before_attempt(
        self, context: RequestContext[Request, TransportRequest]
    ) -> None:
        """A hook called before each attempt at sending the transport request to the
        service.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method will be called multiple times in
        the event of retries.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. In the event of retries, the context will
        not include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        Exceptions thrown by this hook will be stored until all interceptors have had
        their `read_before_attempt` invoked. Other hooks will then be skipped and
        execution will jump to `modify_before_attempt_completion` with the thrown
        exception as the `response` If multiple `read_before_attempt` methods throw
        exceptions, the latest will be used and earlier ones will be logged and dropped.
        """

    def modify_before_signing(
        self, context: RequestContext[Request, TransportRequest]
    ) -> TransportRequest:
        """A hook called before the transport request is signed.

        This method has the ability to modify and return a new transport request of the
        same type.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method will be called multiple times in
        the event of retries.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. The `transport_request` may have been
        modified by earlier `modify_before_signing` hooks, and may be modified further
        by later hooks. In the event of retries, the context will not include changes
        made in previous attempts (e.g. by request signers or other interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.

        The transport request returned by this hook MUST be the same type of request
        passed into this hook. If not, an exception will immediately occur.
        """
        return context.transport_request

    def read_before_signing(
        self, context: RequestContext[Request, TransportRequest]
    ) -> None:
        """A hook called before the transport request is signed.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between invocation of this hook and
        `read_after_signing` is very close to the amount of time spent signing the
        request.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. In the event of retries, the context will
        not include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def read_after_signing(
        self, context: RequestContext[Request, TransportRequest]
    ) -> None:
        """A hook called after the transport request is signed.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between `read_before_signing` and the
        invocation of this hook is very close to the amount of time spent signing the
        request.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. In the event of retries, the context will
        not include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def modify_before_transmit(
        self, context: RequestContext[Request, TransportRequest]
    ) -> TransportRequest:
        """A hook called before the transport request is sent to the service.

        This method has the ability to modify and return a new transport request of the
        same type.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. The `transport_request` may have been
        modified by earlier `modify_before_signing` hooks, and may be modified further
        by later hooks. In the event of retries, the context will not include changes
        made in previous attempts (e.g. by request signers or other interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.

        The transport request returned by this hook MUST be the same type of request
        passed into this hook. If not, an exception will immediately occur.
        """
        return context.transport_request

    def read_before_transmit(
        self, context: RequestContext[Request, TransportRequest]
    ) -> None:
        """A hook called before the transport request is sent to the service.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between invocation of this hook and
        `read_after_transmit` is very close to the amount of time spent communicating
        with the service. Depending on the protocol, the duration may not include the
        time spent reading the response data.

        The `request` and `transport_request` of the context will always be available.
        Other static properties will be None. In the event of retries, the context will
        not include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def read_after_transmit(
        self,
        context: ResponseContext[Request, TransportRequest, TransportResponse],
    ) -> None:
        """A hook called after the transport request is sent to the service and a
        transport response is received.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between `read_before_transmit` and the
        invocation of this hook is very close to the amount of time spent communicating
        with the service. Depending on the protocol, the duration may not include the
        time spent reading the response data.

        The `request`, `transport_request`, and `transport_response` of the context
        will always be available. Other static properties will be None. In the event of
        retries, the context will not include changes made in previous attempts (e.g.
        by request signers or other interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def modify_before_deserialization(
        self,
        context: ResponseContext[Request, TransportRequest, TransportResponse],
    ) -> TransportResponse:
        """A hook called before the transport response is deserialized.

        This method has the ability to modify and return a new transport response of the
        same type.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries.

        The `request`, `transport_request`, and `transport_response` of the context
        will always be available. Other static properties will be None. In the event of
        retries, the context will not include changes made in previous attempts (e.g.
        by request signers or other interceptors). The `transport_response` may have
        been modified by earlier `modify_before_deserialization` hooks, and may be
        modified further by later hooks. In the event of retries, the context will not
        include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.

        The transport response returned by this hook MUST be the same type of
        response passed into this hook. If not, an exception will immediately occur.
        """
        return context.transport_response

    def read_before_deserialization(
        self,
        context: ResponseContext[Request, TransportRequest, TransportResponse],
    ) -> None:
        """A hook called before the transport response is deserialized.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between invocation of this hook and
        `read_after_deserialization` is very close to the amount of time spent
        deserializing the service response. Depending on the protocol and operation,
        the duration may include the time spent downloading the response data.

        The `request`, `transport_request`, and `transport_response` of the context
        will always be available. Other static properties will be None. In the event of
        retries, the context will not include changes made in previous attempts (e.g.
        by request signers or other interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def read_after_deserialization(
        self,
        context: OutputContext[Request, Response, TransportRequest, TransportResponse],
    ) -> None:
        """A hook called after the transport response is deserialized.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per attempt, except when a failure occurs
        earlier in the request pipeline. This method may be called multiple times in
        the event of retries. The duration between `read_before_deserialization`
        and the invocation of this hook is very close to the amount of time spent
        deserializing the service response. Depending on the protocol and operation,
        the duration may include the time spent downloading the response data.

        The `request`, `response`, `transport_request`, and `transport_response` of the
        context will always be available. In the event of retries, the context will not
        include changes made in previous attempts (e.g. by request signers or other
        interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `modify_before_attempt_completion` with the thrown exception as the `response`.
        """

    def modify_before_attempt_completion(
        self,
        context: OutputContext[
            Request, Response, TransportRequest, TransportResponse | None
        ],
    ) -> Response | Exception:
        """A hook called when an attempt is completed.

        This method has the ability to modify and return a new output message or
        exception matching the currently-executing operation.

        This will ALWAYS be called once per attempt, except when a failure occurs
        before `read_before_attempt`. This method may be called multiple times in the
        event of retries.

        The `request`, `response`, and `transport_request` of the context will always
        be available. The `transport_response` will be available if a response was
        received by the service for this attempt. In the event of retries, the context
        will not include changes made in previous attempts (e.g. by request signers or
        other interceptors).

        If exceptions are thrown by this hook, execution will jump to
        `read_after_attempt` with  the thrown exception as the `response`.

        Any output returned by this hook MUST match the operation being invoked. Any
        exception type can be returned, replacing the `response` currently in the
        context.
        """
        return context.response

    def read_after_attempt(
        self,
        context: OutputContext[
            Request, Response, TransportRequest, TransportResponse | None
        ],
    ) -> None:
        """A hook called when an attempt is completed.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will ALWAYS be called once per attempt, as long as `read_before_attempt`
        has been executed.

        The `request`, `response`, and `transport_request` of the context will always
        be available. The `transport_response` will be available if a response was
        received by the service for this attempt. In the event of retries, the context
        will not include changes made in previous attempts (e.g. by request signers or
        other interceptors).

        Exceptions thrown by this hook will be stored until all interceptors have had
        their `read_after_attempt` invoked. If multiple `read_after_attempt` methods
        throw exceptions, the latest will be used and earlier ones will be logged and
        dropped. If the retry strategy determines that the execution is retryable,
        execution will then jump to `read_before_attempt`. Otherwise, execution will
        jump to `modify_before_completion` with the thrown exception as the `response`.
        """

    def modify_before_completion(
        self,
        context: OutputContext[
            Request, Response, TransportRequest | None, TransportResponse | None
        ],
    ) -> Response | Exception:
        """A hook called when an execution is completed.

        This method has the ability to modify and return a new output message or
        exception matching the currently-executing operation.

        This will always be called once per execution.

        The `request` and `response` of the context will always be available. The
        `transport_request` and `transport_response` will be available if the execution
        proceeded far enough for them to be generated.

        If exceptions are thrown by this hook, execution will jump to
        `read_after_execution` with the thrown exception as the `response`.

        Any output returned by this hook MUST match the operation being invoked. Any
        exception type can be returned, replacing the `response` currently in the context.
        """
        return context.response

    def read_after_execution(
        self,
        context: OutputContext[
            Request, Response, TransportRequest | None, TransportResponse | None
        ],
    ) -> None:
        """A hook called when an execution is completed.

        Implementations MUST NOT modify the `request`, `response`, `transport_request`,
        or `transport_response` in this hook.

        This will always be called once per execution. The duration between
        `read_before_execution` and the invocation of this hook is very close to the
        full duration of the execution.

        The `request` and `response` of the context will always be available. The
        `transport_request` and `transport_response` will be available if the execution
        proceeded far enough for them to be generated.

        Exceptions thrown by this hook will be stored until all interceptors have had
        their `read_after_execution` invoked. The exception will then be treated as the
        final response. If multiple `read_after_execution` methods throw exceptions,
        the latest will be used and earlier ones will be logged and dropped.
        """

modify_before_attempt_completion(context)

A hook called when an attempt is completed.

This method has the ability to modify and return a new output message or exception matching the currently-executing operation.

This will ALWAYS be called once per attempt, except when a failure occurs before read_before_attempt. This method may be called multiple times in the event of retries.

The request, response, and transport_request of the context will always be available. The transport_response will be available if a response was received by the service for this attempt. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to read_after_attempt with the thrown exception as the response.

Any output returned by this hook MUST match the operation being invoked. Any exception type can be returned, replacing the response currently in the context.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_attempt_completion(
    self,
    context: OutputContext[
        Request, Response, TransportRequest, TransportResponse | None
    ],
) -> Response | Exception:
    """A hook called when an attempt is completed.

    This method has the ability to modify and return a new output message or
    exception matching the currently-executing operation.

    This will ALWAYS be called once per attempt, except when a failure occurs
    before `read_before_attempt`. This method may be called multiple times in the
    event of retries.

    The `request`, `response`, and `transport_request` of the context will always
    be available. The `transport_response` will be available if a response was
    received by the service for this attempt. In the event of retries, the context
    will not include changes made in previous attempts (e.g. by request signers or
    other interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `read_after_attempt` with  the thrown exception as the `response`.

    Any output returned by this hook MUST match the operation being invoked. Any
    exception type can be returned, replacing the `response` currently in the
    context.
    """
    return context.response

modify_before_completion(context)

A hook called when an execution is completed.

This method has the ability to modify and return a new output message or exception matching the currently-executing operation.

This will always be called once per execution.

The request and response of the context will always be available. The transport_request and transport_response will be available if the execution proceeded far enough for them to be generated.

If exceptions are thrown by this hook, execution will jump to read_after_execution with the thrown exception as the response.

Any output returned by this hook MUST match the operation being invoked. Any exception type can be returned, replacing the response currently in the context.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_completion(
    self,
    context: OutputContext[
        Request, Response, TransportRequest | None, TransportResponse | None
    ],
) -> Response | Exception:
    """A hook called when an execution is completed.

    This method has the ability to modify and return a new output message or
    exception matching the currently-executing operation.

    This will always be called once per execution.

    The `request` and `response` of the context will always be available. The
    `transport_request` and `transport_response` will be available if the execution
    proceeded far enough for them to be generated.

    If exceptions are thrown by this hook, execution will jump to
    `read_after_execution` with the thrown exception as the `response`.

    Any output returned by this hook MUST match the operation being invoked. Any
    exception type can be returned, replacing the `response` currently in the context.
    """
    return context.response

modify_before_deserialization(context)

A hook called before the transport response is deserialized.

This method has the ability to modify and return a new transport response of the same type.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries.

The request, transport_request, and transport_response of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors). The transport_response may have been modified by earlier modify_before_deserialization hooks, and may be modified further by later hooks. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

The transport response returned by this hook MUST be the same type of response passed into this hook. If not, an exception will immediately occur.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_deserialization(
    self,
    context: ResponseContext[Request, TransportRequest, TransportResponse],
) -> TransportResponse:
    """A hook called before the transport response is deserialized.

    This method has the ability to modify and return a new transport response of the
    same type.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries.

    The `request`, `transport_request`, and `transport_response` of the context
    will always be available. Other static properties will be None. In the event of
    retries, the context will not include changes made in previous attempts (e.g.
    by request signers or other interceptors). The `transport_response` may have
    been modified by earlier `modify_before_deserialization` hooks, and may be
    modified further by later hooks. In the event of retries, the context will not
    include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.

    The transport response returned by this hook MUST be the same type of
    response passed into this hook. If not, an exception will immediately occur.
    """
    return context.transport_response

modify_before_retry_loop(context)

A hook called before the retry loop is entered.

This method has the ability to modify and return a new transport request of the same type.

This will always be called once per execution, except when a failure occurs earlier in the request pipeline.

If exceptions are thrown by this hook, execution will jump to modify_before_completion with the thrown exception as the response.

The transport request returned by this hook MUST be the same type of request passed into this hook. If not, an exception will immediately occur.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_retry_loop(
    self, context: RequestContext[Request, TransportRequest]
) -> TransportRequest:
    """A hook called before the retry loop is entered.

    This method has the ability to modify and return a new transport request of the
    same type.

    This will always be called once per execution, except when a failure occurs
    earlier in the request pipeline.

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_completion` with the thrown exception as the `response`.

    The transport request returned by this hook MUST be the same type of request
    passed into this hook. If not, an exception will immediately occur.
    """
    return context.transport_request

modify_before_serialization(context)

A hook called before the request is serialized into a transport request.

This method has the ability to modify and return a new request of the same type.

This will ALWAYS be called once per execution, except when a failure occurs earlier in the request pipeline.

The request of the context will always be available. This request may have been modified by earlier modify_before_serialization hooks, and may be modified further by later hooks. Other static properites will be None.

If exceptions are thrown by this hook, execution will jump to modify_before_completion with the thrown exception as the response.

The request returned by this hook MUST be the same type of request message passed into this hook. If not, an exception will immediately occur.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_serialization(self, context: InputContext[Request]) -> Request:
    """A hook called before the request is serialized into a transport request.

    This method has the ability to modify and return a new request of the same
    type.

    This will ALWAYS be called once per execution, except when a failure occurs
    earlier in the request pipeline.

    The `request` of the context will always be available. This `request` may have
    been modified by earlier `modify_before_serialization` hooks, and may be
    modified further by later hooks. Other static properites will be None.

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_completion` with the thrown exception as the `response`.

    The request returned by this hook MUST be the same type of request
    message passed into this hook. If not, an exception will immediately occur.
    """
    return context.request

modify_before_signing(context)

A hook called before the transport request is signed.

This method has the ability to modify and return a new transport request of the same type.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method will be called multiple times in the event of retries.

The request and transport_request of the context will always be available. Other static properties will be None. The transport_request may have been modified by earlier modify_before_signing hooks, and may be modified further by later hooks. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

The transport request returned by this hook MUST be the same type of request passed into this hook. If not, an exception will immediately occur.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_signing(
    self, context: RequestContext[Request, TransportRequest]
) -> TransportRequest:
    """A hook called before the transport request is signed.

    This method has the ability to modify and return a new transport request of the
    same type.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method will be called multiple times in
    the event of retries.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. The `transport_request` may have been
    modified by earlier `modify_before_signing` hooks, and may be modified further
    by later hooks. In the event of retries, the context will not include changes
    made in previous attempts (e.g. by request signers or other interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.

    The transport request returned by this hook MUST be the same type of request
    passed into this hook. If not, an exception will immediately occur.
    """
    return context.transport_request

modify_before_transmit(context)

A hook called before the transport request is sent to the service.

This method has the ability to modify and return a new transport request of the same type.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries.

The request and transport_request of the context will always be available. Other static properties will be None. The transport_request may have been modified by earlier modify_before_signing hooks, and may be modified further by later hooks. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

The transport request returned by this hook MUST be the same type of request passed into this hook. If not, an exception will immediately occur.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def modify_before_transmit(
    self, context: RequestContext[Request, TransportRequest]
) -> TransportRequest:
    """A hook called before the transport request is sent to the service.

    This method has the ability to modify and return a new transport request of the
    same type.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. The `transport_request` may have been
    modified by earlier `modify_before_signing` hooks, and may be modified further
    by later hooks. In the event of retries, the context will not include changes
    made in previous attempts (e.g. by request signers or other interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.

    The transport request returned by this hook MUST be the same type of request
    passed into this hook. If not, an exception will immediately occur.
    """
    return context.transport_request

read_after_attempt(context)

A hook called when an attempt is completed.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will ALWAYS be called once per attempt, as long as read_before_attempt has been executed.

The request, response, and transport_request of the context will always be available. The transport_response will be available if a response was received by the service for this attempt. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

Exceptions thrown by this hook will be stored until all interceptors have had their read_after_attempt invoked. If multiple read_after_attempt methods throw exceptions, the latest will be used and earlier ones will be logged and dropped. If the retry strategy determines that the execution is retryable, execution will then jump to read_before_attempt. Otherwise, execution will jump to modify_before_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_attempt(
    self,
    context: OutputContext[
        Request, Response, TransportRequest, TransportResponse | None
    ],
) -> None:
    """A hook called when an attempt is completed.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will ALWAYS be called once per attempt, as long as `read_before_attempt`
    has been executed.

    The `request`, `response`, and `transport_request` of the context will always
    be available. The `transport_response` will be available if a response was
    received by the service for this attempt. In the event of retries, the context
    will not include changes made in previous attempts (e.g. by request signers or
    other interceptors).

    Exceptions thrown by this hook will be stored until all interceptors have had
    their `read_after_attempt` invoked. If multiple `read_after_attempt` methods
    throw exceptions, the latest will be used and earlier ones will be logged and
    dropped. If the retry strategy determines that the execution is retryable,
    execution will then jump to `read_before_attempt`. Otherwise, execution will
    jump to `modify_before_completion` with the thrown exception as the `response`.
    """

read_after_deserialization(context)

A hook called after the transport response is deserialized.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between read_before_deserialization and the invocation of this hook is very close to the amount of time spent deserializing the service response. Depending on the protocol and operation, the duration may include the time spent downloading the response data.

The request, response, transport_request, and transport_response of the context will always be available. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_deserialization(
    self,
    context: OutputContext[Request, Response, TransportRequest, TransportResponse],
) -> None:
    """A hook called after the transport response is deserialized.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between `read_before_deserialization`
    and the invocation of this hook is very close to the amount of time spent
    deserializing the service response. Depending on the protocol and operation,
    the duration may include the time spent downloading the response data.

    The `request`, `response`, `transport_request`, and `transport_response` of the
    context will always be available. In the event of retries, the context will not
    include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

read_after_execution(context)

A hook called when an execution is completed.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per execution. The duration between read_before_execution and the invocation of this hook is very close to the full duration of the execution.

The request and response of the context will always be available. The transport_request and transport_response will be available if the execution proceeded far enough for them to be generated.

Exceptions thrown by this hook will be stored until all interceptors have had their read_after_execution invoked. The exception will then be treated as the final response. If multiple read_after_execution methods throw exceptions, the latest will be used and earlier ones will be logged and dropped.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_execution(
    self,
    context: OutputContext[
        Request, Response, TransportRequest | None, TransportResponse | None
    ],
) -> None:
    """A hook called when an execution is completed.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per execution. The duration between
    `read_before_execution` and the invocation of this hook is very close to the
    full duration of the execution.

    The `request` and `response` of the context will always be available. The
    `transport_request` and `transport_response` will be available if the execution
    proceeded far enough for them to be generated.

    Exceptions thrown by this hook will be stored until all interceptors have had
    their `read_after_execution` invoked. The exception will then be treated as the
    final response. If multiple `read_after_execution` methods throw exceptions,
    the latest will be used and earlier ones will be logged and dropped.
    """

read_after_serialization(context)

A hook called after the input message is serialized into a transport request.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per execution, except when a failure occurs earlier in the request pipeline. The duration between read_before_serialization and the invocation of this hook is very close to the amount of time spent serializing the request.

The request and transport_request of the context will always be available. Other static properties will be None.

If exceptions are thrown by this hook, execution will jump to modify_before_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_serialization(
    self, context: RequestContext[Request, TransportRequest]
) -> None:
    """A hook called after the input message is serialized into a transport request.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per execution, except when a failure occurs
    earlier in the request pipeline. The duration between
    `read_before_serialization` and the invocation of this hook is very close to
    the amount of time spent serializing the request.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None.

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_completion` with the thrown exception as the `response`.
    """

read_after_signing(context)

A hook called after the transport request is signed.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between read_before_signing and the invocation of this hook is very close to the amount of time spent signing the request.

The request and transport_request of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_signing(
    self, context: RequestContext[Request, TransportRequest]
) -> None:
    """A hook called after the transport request is signed.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between `read_before_signing` and the
    invocation of this hook is very close to the amount of time spent signing the
    request.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. In the event of retries, the context will
    not include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

read_after_transmit(context)

A hook called after the transport request is sent to the service and a transport response is received.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between read_before_transmit and the invocation of this hook is very close to the amount of time spent communicating with the service. Depending on the protocol, the duration may not include the time spent reading the response data.

The request, transport_request, and transport_response of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_after_transmit(
    self,
    context: ResponseContext[Request, TransportRequest, TransportResponse],
) -> None:
    """A hook called after the transport request is sent to the service and a
    transport response is received.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between `read_before_transmit` and the
    invocation of this hook is very close to the amount of time spent communicating
    with the service. Depending on the protocol, the duration may not include the
    time spent reading the response data.

    The `request`, `transport_request`, and `transport_response` of the context
    will always be available. Other static properties will be None. In the event of
    retries, the context will not include changes made in previous attempts (e.g.
    by request signers or other interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

read_before_attempt(context)

A hook called before each attempt at sending the transport request to the service.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method will be called multiple times in the event of retries.

The request and transport_request of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

Exceptions thrown by this hook will be stored until all interceptors have had their read_before_attempt invoked. Other hooks will then be skipped and execution will jump to modify_before_attempt_completion with the thrown exception as the response If multiple read_before_attempt methods throw exceptions, the latest will be used and earlier ones will be logged and dropped.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_attempt(
    self, context: RequestContext[Request, TransportRequest]
) -> None:
    """A hook called before each attempt at sending the transport request to the
    service.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method will be called multiple times in
    the event of retries.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. In the event of retries, the context will
    not include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    Exceptions thrown by this hook will be stored until all interceptors have had
    their `read_before_attempt` invoked. Other hooks will then be skipped and
    execution will jump to `modify_before_attempt_completion` with the thrown
    exception as the `response` If multiple `read_before_attempt` methods throw
    exceptions, the latest will be used and earlier ones will be logged and dropped.
    """

read_before_deserialization(context)

A hook called before the transport response is deserialized.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between invocation of this hook and read_after_deserialization is very close to the amount of time spent deserializing the service response. Depending on the protocol and operation, the duration may include the time spent downloading the response data.

The request, transport_request, and transport_response of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_deserialization(
    self,
    context: ResponseContext[Request, TransportRequest, TransportResponse],
) -> None:
    """A hook called before the transport response is deserialized.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between invocation of this hook and
    `read_after_deserialization` is very close to the amount of time spent
    deserializing the service response. Depending on the protocol and operation,
    the duration may include the time spent downloading the response data.

    The `request`, `transport_request`, and `transport_response` of the context
    will always be available. Other static properties will be None. In the event of
    retries, the context will not include changes made in previous attempts (e.g.
    by request signers or other interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

read_before_execution(context)

A hook called at the start of an execution, before the SDK does anything else.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per execution. The duration between invocation of this hook and read_after_execution is very close to full duration of the execution.

The request of the context will always be available. Other static properties will be None.

Exceptions thrown by this hook will be stored until all interceptors have had their read_before_execution invoked. Other hooks will then be skipped and execution will jump to modify_before_completion with the thrown exception as the response. If multiple read_before_execution methods throw exceptions, the latest will be used and earlier ones will be logged and dropped.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_execution(self, context: InputContext[Request]) -> None:
    """A hook called at the start of an execution, before the SDK does anything
    else.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per execution. The duration between invocation
    of this hook and `read_after_execution` is very close to full duration of the
    execution.

    The `request` of the context will always be available. Other static properties
    will be None.

    Exceptions thrown by this hook will be stored until all interceptors have had
    their `read_before_execution` invoked. Other hooks will then be skipped and
    execution will jump to `modify_before_completion` with the thrown exception as
    the `response`. If multiple `read_before_execution` methods throw exceptions,
    the latest will be used and earlier ones will be logged and dropped.
    """

read_before_serialization(context)

A hook called before the input message is serialized into a transport request.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per execution, except when a failure occurs earlier in the request pipeline. The duration between invocation of this hook and read_after_serialization is very close to the amount of time spent marshalling the request.

The request of the context will always be available. Other static properties will be None.

If exceptions are thrown by this hook, execution will jump to modify_before_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_serialization(self, context: InputContext[Request]) -> None:
    """A hook called before the input message is serialized into a transport
    request.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per execution, except when a failure occurs
    earlier in the request pipeline. The duration between invocation of this hook
    and `read_after_serialization` is very close to the amount of time spent
    marshalling the request.

    The `request` of the context will always be available. Other static properties
    will be None.

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_completion` with the thrown exception as the `response`.
    """

read_before_signing(context)

A hook called before the transport request is signed.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between invocation of this hook and read_after_signing is very close to the amount of time spent signing the request.

The request and transport_request of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_signing(
    self, context: RequestContext[Request, TransportRequest]
) -> None:
    """A hook called before the transport request is signed.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between invocation of this hook and
    `read_after_signing` is very close to the amount of time spent signing the
    request.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. In the event of retries, the context will
    not include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

read_before_transmit(context)

A hook called before the transport request is sent to the service.

Implementations MUST NOT modify the request, response, transport_request, or transport_response in this hook.

This will always be called once per attempt, except when a failure occurs earlier in the request pipeline. This method may be called multiple times in the event of retries. The duration between invocation of this hook and read_after_transmit is very close to the amount of time spent communicating with the service. Depending on the protocol, the duration may not include the time spent reading the response data.

The request and transport_request of the context will always be available. Other static properties will be None. In the event of retries, the context will not include changes made in previous attempts (e.g. by request signers or other interceptors).

If exceptions are thrown by this hook, execution will jump to modify_before_attempt_completion with the thrown exception as the response.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def read_before_transmit(
    self, context: RequestContext[Request, TransportRequest]
) -> None:
    """A hook called before the transport request is sent to the service.

    Implementations MUST NOT modify the `request`, `response`, `transport_request`,
    or `transport_response` in this hook.

    This will always be called once per attempt, except when a failure occurs
    earlier in the request pipeline. This method may be called multiple times in
    the event of retries. The duration between invocation of this hook and
    `read_after_transmit` is very close to the amount of time spent communicating
    with the service. Depending on the protocol, the duration may not include the
    time spent reading the response data.

    The `request` and `transport_request` of the context will always be available.
    Other static properties will be None. In the event of retries, the context will
    not include changes made in previous attempts (e.g. by request signers or other
    interceptors).

    If exceptions are thrown by this hook, execution will jump to
    `modify_before_attempt_completion` with the thrown exception as the `response`.
    """

InterceptorChain

Bases: AnyInterceptor

An interceptor that contains an ordered list of delegate interceptors.

This is primarily intended for use within the client itself.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

class InterceptorChain(AnyInterceptor):
    """An interceptor that contains an ordered list of delegate interceptors.

    This is primarily intended for use within the client itself.
    """

    def __init__(self, chain: Sequence[AnyInterceptor]) -> None:
        """Construct an InterceptorChain.

        :param chain: The ordered interceptors to chain together.
        """
        self._chain = tuple(chain)

    def __repr__(self) -> str:
        return f"InterceptorChain(chain={self._chain!r})"

    def read_before_execution(self, context: InputContext[Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_before_execution(context)

    def modify_before_serialization(self, context: InputContext[Any]) -> Any:
        request = context.request
        for interceptor in self._chain:
            request = interceptor.modify_before_serialization(context)
        return request

    def read_before_serialization(self, context: InputContext[Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_before_serialization(context)

    def read_after_serialization(self, context: RequestContext[Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_after_serialization(context)

    def modify_before_retry_loop(self, context: RequestContext[Any, Any]) -> Any:
        transport_request = context.transport_request
        for interceptor in self._chain:
            transport_request = interceptor.modify_before_retry_loop(context)
        return transport_request

    def read_before_attempt(self, context: RequestContext[Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_before_attempt(context)

    def modify_before_signing(self, context: RequestContext[Any, Any]) -> Any:
        transport_request = context.transport_request
        for interceptor in self._chain:
            transport_request = interceptor.modify_before_signing(context)
        return transport_request

    def read_before_signing(self, context: RequestContext[Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_before_signing(context)

    def read_after_signing(self, context: RequestContext[Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_after_signing(context)

    def modify_before_transmit(self, context: RequestContext[Any, Any]) -> Any:
        transport_request = context.transport_request
        for interceptor in self._chain:
            transport_request = interceptor.modify_before_transmit(context)
        return transport_request

    def read_before_transmit(self, context: RequestContext[Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_before_transmit(context)

    def read_after_transmit(self, context: ResponseContext[Any, Any, Any]) -> None:
        for interceptor in self._chain:
            interceptor.read_after_transmit(context)

    def modify_before_deserialization(
        self, context: ResponseContext[Any, Any, Any]
    ) -> Any:
        transport_response = context.transport_response
        for interceptor in self._chain:
            transport_response = interceptor.modify_before_deserialization(context)
        return transport_response

    def read_before_deserialization(
        self, context: ResponseContext[Any, Any, Any]
    ) -> None:
        for interceptor in self._chain:
            interceptor.read_before_deserialization(context)

    def read_after_deserialization(
        self, context: OutputContext[Any, Any, Any, Any]
    ) -> None:
        for interceptor in self._chain:
            interceptor.read_after_deserialization(context)

    def modify_before_attempt_completion(
        self, context: OutputContext[Any, Any, Any, Any | None]
    ) -> Any | Exception:
        response = context.response
        for interceptor in self._chain:
            response = interceptor.modify_before_attempt_completion(context)
        return response

    def read_after_attempt(
        self, context: OutputContext[Any, Any, Any, Any | None]
    ) -> None:
        for interceptor in self._chain:
            interceptor.read_after_attempt(context)

    def modify_before_completion(
        self, context: OutputContext[Any, Any, Any | None, Any | None]
    ) -> Any | Exception:
        response = context.response
        for interceptor in self._chain:
            response = interceptor.modify_before_attempt_completion(context)
        return response

    def read_after_execution(
        self, context: OutputContext[Any, Any, Any | None, Any | None]
    ) -> None:
        exception: Exception | None = None
        for interceptor in self._chain:
            # Every one of these is supposed to be guaranteed to be called.
            try:
                interceptor.read_after_execution(context)
            except Exception as e:
                context = replace(context, response=e)
                exception = e
        if exception is not None:
            raise exception

__init__(chain)

Construct an InterceptorChain.

:param chain: The ordered interceptors to chain together.

Source code in packages/smithy-core/src/smithy_core/interceptors.py

def __init__(self, chain: Sequence[AnyInterceptor]) -> None:
    """Construct an InterceptorChain.

    :param chain: The ordered interceptors to chain together.
    """
    self._chain = tuple(chain)

OutputContext dataclass

Bases: ResponseContext[Request, TransportRequest, TransportResponse]

Source code in packages/smithy-core/src/smithy_core/interceptors.py

@dataclass(kw_only=True, frozen=True, slots=True)
class OutputContext[
    Request: SerializeableShape,
    Response: DeserializeableShape,
    TransportRequest,
    TransportResponse,
](ResponseContext[Request, TransportRequest, TransportResponse]):
    response: Response | Exception
    """The modeled response for the operation being invoked."""

response instance-attribute

The modeled response for the operation being invoked.

RequestContext dataclass

Bases: InputContext[Request]

Source code in packages/smithy-core/src/smithy_core/interceptors.py

@dataclass(kw_only=True, frozen=True, slots=True)
class RequestContext[Request: SerializeableShape, TransportRequest](
    InputContext[Request]
):
    transport_request: TransportRequest
    """The transmittable request for the operation being invoked."""

transport_request instance-attribute

The transmittable request for the operation being invoked.

ResponseContext dataclass

Bases: RequestContext[Request, TransportRequest]

Source code in packages/smithy-core/src/smithy_core/interceptors.py

@dataclass(kw_only=True, frozen=True, slots=True)
class ResponseContext[Request: SerializeableShape, TransportRequest, TransportResponse](
    RequestContext[Request, TransportRequest]
):
    transport_response: TransportResponse
    """The transmitted response for the operation being invoked."""

transport_response instance-attribute

The transmitted response for the operation being invoked.

interfaces

BytesReader

Bases: Protocol

A protocol for objects that support reading bytes from them.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

@runtime_checkable
class BytesReader(Protocol):
    """A protocol for objects that support reading bytes from them."""

    def read(self, size: int = -1, /) -> bytes: ...

BytesWriter

Bases: Protocol

A protocol for objects that support writing bytes to them.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

@runtime_checkable
class BytesWriter(Protocol):
    """A protocol for objects that support writing bytes to them."""

    def write(self, b: bytes, /) -> int: ...

Endpoint

Bases: Protocol

A resolved endpoint.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

class Endpoint(Protocol):
    """A resolved endpoint."""

    uri: URI
    """The endpoint URI."""

    properties: "TypedProperties"
    """Properties required to interact with the endpoint.

    For example, in some AWS use cases this might contain HTTP headers to add to each
    request.
    """

properties instance-attribute

Properties required to interact with the endpoint.

For example, in some AWS use cases this might contain HTTP headers to add to each request.

uri instance-attribute

The endpoint URI.

PropertyKey

Bases: Protocol

A typed properties key.

Used with :py:class:Context to set and get typed values.

For a concrete implementation, see :py:class:smithy_core.types.PropertyKey.

.. code-block:: python

UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
    key="union",
    value_type=str | int,
)

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

@runtime_checkable
class PropertyKey[T](Protocol):
    """A typed properties key.

    Used with :py:class:`Context` to set and get typed values.

    For a concrete implementation, see :py:class:`smithy_core.types.PropertyKey`.

    .. code-block:: python

        UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
            key="union",
            value_type=str | int,
        )
    """

    key: str
    """The string key used to access the value."""

    value_type: "TypeForm[T]"
    """The type of the associated value in the properties bag."""

    def __str__(self) -> str:
        return self.key

key instance-attribute

The string key used to access the value.

value_type instance-attribute

The type of the associated value in the properties bag.

SeekableBytesReader

Bases: BytesReader, Protocol

A synchronous bytes reader with seek and tell methods.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

@runtime_checkable
class SeekableBytesReader(BytesReader, Protocol):
    """A synchronous bytes reader with seek and tell methods."""

    def tell(self) -> int: ...
    def seek(self, offset: int, whence: int = 0, /) -> int: ...

TypedProperties

Bases: Protocol

A properties map with typed setters and getters.

Keys can be either a string or a :py:class:PropertyKey. Using a PropertyKey instead of a string enables type checkers to narrow to the associated value type rather than having to use Any.

Letting the value be either a string or PropertyKey allows consumers who care about typing to get it, and those who don't care about typing to not have to think about it.

..code-block:: python

foo = PropertyKey(key="foo", value_type=str)
properties = TypedProperties()
properties[foo] = "bar"

assert assert_type(properties[foo], str) == "bar"
assert assert_type(properties["foo"], Any) == "bar"

For a concrete implementation, see :py:class:smithy_core.types.TypedProperties.

Note that unions and other special types cannot easily be used here due to being incompatible with type[T]. PEP747 proposes a fix to this case, but it has not yet been accepted. In the meantime, there is a workaround. The PropertyKey must be assigned to an explicitly typed variable, and the value_type parameter of the constructor must have a # type: ignore comment, like so:

.. code-block:: python

UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
    key="union",
    value_type=str | int,  # type: ignore
)

properties = TypedProperties()
properties[UNION_PROPERTY] = "foo"

assert assert_type(properties[UNION_PROPERTY], str | int) == "foo"

Type checkers will be able to use such a property as expected.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

@runtime_checkable
class TypedProperties(Protocol):
    """A properties map with typed setters and getters.

    Keys can be either a string or a :py:class:`PropertyKey`. Using a PropertyKey instead
    of a string enables type checkers to narrow to the associated value type rather
    than having to use Any.

    Letting the value be either a string or PropertyKey allows consumers who care about
    typing to get it, and those who don't care about typing to not have to think about
    it.

    ..code-block:: python

        foo = PropertyKey(key="foo", value_type=str)
        properties = TypedProperties()
        properties[foo] = "bar"

        assert assert_type(properties[foo], str) == "bar"
        assert assert_type(properties["foo"], Any) == "bar"

    For a concrete implementation, see :py:class:`smithy_core.types.TypedProperties`.

    Note that unions and other special types cannot easily be used here due to being
    incompatible with ``type[T]``. PEP747 proposes a fix to this case, but it has not
    yet been accepted. In the meantime, there is a workaround. The PropertyKey must
    be assigned to an explicitly typed variable, and the ``value_type`` parameter of
    the constructor must have a ``# type: ignore`` comment, like so:

    .. code-block:: python

        UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
            key="union",
            value_type=str | int,  # type: ignore
        )

        properties = TypedProperties()
        properties[UNION_PROPERTY] = "foo"

        assert assert_type(properties[UNION_PROPERTY], str | int) == "foo"

    Type checkers will be able to use such a property as expected.
    """

    @overload
    def __getitem__[T](self, key: PropertyKey[T]) -> T: ...
    @overload
    def __getitem__(self, key: str) -> Any: ...

    @overload
    def __setitem__[T](self, key: PropertyKey[T], value: T) -> None: ...
    @overload
    def __setitem__(self, key: str, value: Any) -> None: ...

    def __delitem__(self, key: str | PropertyKey[Any]) -> None: ...

    @overload
    def get[T](self, key: PropertyKey[T], default: None = None) -> T | None: ...
    @overload
    def get[T](self, key: PropertyKey[T], default: T) -> T: ...
    @overload
    def get[T, DT](self, key: PropertyKey[T], default: DT) -> T | DT: ...
    @overload
    def get(self, key: str, default: None = None) -> Any | None: ...
    @overload
    def get[T](self, key: str, default: T) -> Any | T: ...

    @overload
    def pop[T](self, key: PropertyKey[T], default: None = None) -> T | None: ...
    @overload
    def pop[T](self, key: PropertyKey[T], default: T) -> T: ...
    @overload
    def pop[T, DT](self, key: PropertyKey[T], default: DT) -> T | DT: ...
    @overload
    def pop(self, key: str, default: None = None) -> Any | None: ...
    @overload
    def pop[T](self, key: str, default: T) -> Any | T: ...

    def __iter__(self) -> Iterator[str]: ...
    def items(self) -> ItemsView[str, Any]: ...
    def keys(self) -> KeysView[str]: ...
    def values(self) -> ValuesView[Any]: ...
    def __contains__(self, key: object) -> bool: ...

URI

Bases: Protocol

Universal Resource Identifier, target location for a :py:class:Request.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

class URI(Protocol):
    """Universal Resource Identifier, target location for a :py:class:`Request`."""

    scheme: str
    """For example ``http`` or ``mqtts``."""

    username: str | None
    """Username part of the userinfo URI component."""

    password: str | None
    """Password part of the userinfo URI component."""

    host: str
    """The hostname, for example ``amazonaws.com``."""

    port: int | None
    """An explicit port number."""

    path: str | None
    """Path component of the URI."""

    query: str | None
    """Query component of the URI as string."""

    fragment: str | None
    """Part of the URI specification, but may not be transmitted by a client."""

    def build(self) -> str:
        """Construct URI string representation.

        Returns a string of the form
        ``{scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}``
        """
        ...

    @property
    def netloc(self) -> str:
        """Construct netloc string in format ``{username}:{password}@{host}:{port}``"""
        ...

fragment instance-attribute

Part of the URI specification, but may not be transmitted by a client.

host instance-attribute

The hostname, for example amazonaws.com.

netloc property

Construct netloc string in format {username}:{password}@{host}:{port}

password instance-attribute

Password part of the userinfo URI component.

path instance-attribute

Path component of the URI.

port instance-attribute

An explicit port number.

query instance-attribute

Query component of the URI as string.

scheme instance-attribute

For example http or mqtts.

username instance-attribute

Username part of the userinfo URI component.

build()

Construct URI string representation.

Returns a string of the form {scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

def build(self) -> str:
    """Construct URI string representation.

    Returns a string of the form
    ``{scheme}://{username}:{password}@{host}:{port}{path}?{query}#{fragment}``
    """
    ...

is_bytes_reader(obj)

Determines whether the given object conforms to the BytesReader protocol.

This is necessary to distinguish this from an async reader, since runtime_checkable doesn't make that distinction.

:param obj: The object to inspect.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

def is_bytes_reader(obj: Any) -> TypeGuard[BytesReader]:
    """Determines whether the given object conforms to the BytesReader protocol.

    This is necessary to distinguish this from an async reader, since runtime_checkable
    doesn't make that distinction.

    :param obj: The object to inspect.
    """
    return isinstance(obj, BytesReader) and not iscoroutinefunction(
        getattr(obj, "read")
    )

is_streaming_blob(obj)

Determines whether the given object is a StreamingBlob.

:param obj: The object to inspect.

Source code in packages/smithy-core/src/smithy_core/interfaces/__init__.py

def is_streaming_blob(obj: Any) -> TypeGuard[StreamingBlob]:
    """Determines whether the given object is a StreamingBlob.

    :param obj: The object to inspect.
    """
    return isinstance(obj, bytes | bytearray) or is_bytes_reader(obj)

auth

AuthOption

Bases: Protocol

Auth scheme used for signing and identity resolution.

Source code in packages/smithy-core/src/smithy_core/interfaces/auth.py

class AuthOption(Protocol):
    """Auth scheme used for signing and identity resolution."""

    scheme_id: ShapeID
    """The ID of the auth scheme to use."""

    identity_properties: TypedProperties
    """Paramters to pass to the identity resolver method."""

    signer_properties: TypedProperties
    """Paramters to pass to the signing method."""

identity_properties instance-attribute

Paramters to pass to the identity resolver method.

scheme_id instance-attribute

The ID of the auth scheme to use.

signer_properties instance-attribute

Paramters to pass to the signing method.

AuthSchemeResolver

Bases: Protocol

Determines which authentication scheme to use for a given service.

Source code in packages/smithy-core/src/smithy_core/interfaces/auth.py

class AuthSchemeResolver(Protocol):
    """Determines which authentication scheme to use for a given service."""

    def resolve_auth_scheme(
        self, *, auth_parameters: "AuthParams[Any, Any]"
    ) -> Sequence[AuthOption]:
        """Resolve an ordered list of applicable auth schemes.

        :param auth_parameters: The parameters required for determining which
            authentication schemes to potentially use.
        """
        ...

resolve_auth_scheme(*, auth_parameters)

Resolve an ordered list of applicable auth schemes.

:param auth_parameters: The parameters required for determining which authentication schemes to potentially use.

Source code in packages/smithy-core/src/smithy_core/interfaces/auth.py

def resolve_auth_scheme(
    self, *, auth_parameters: "AuthParams[Any, Any]"
) -> Sequence[AuthOption]:
    """Resolve an ordered list of applicable auth schemes.

    :param auth_parameters: The parameters required for determining which
        authentication schemes to potentially use.
    """
    ...

exceptions

HasFault

Bases: Protocol

A protocol for a modeled error.

All modeled errors will have a fault that is either "client" or "server".

Source code in packages/smithy-core/src/smithy_core/interfaces/exceptions.py

@runtime_checkable
class HasFault(Protocol):
    """A protocol for a modeled error.

    All modeled errors will have a fault that is either "client" or "server".
    """

    fault: Fault

identity

Identity

Bases: Protocol

An entity available to the client representing who the user is.

Source code in packages/smithy-core/src/smithy_core/interfaces/identity.py

@runtime_checkable
class Identity(Protocol):
    """An entity available to the client representing who the user is."""

    expiration: datetime | None = None
    """The expiration time of the identity.

    If time zone is provided, it is updated to UTC. The value must always be in UTC.
    """

    def __post_init__(self) -> None:
        if self.expiration is not None:
            self.expiration = ensure_utc(self.expiration)

    @property
    def is_expired(self) -> bool:
        """Whether the identity is expired."""
        if self.expiration is None:
            return False
        return datetime.now(tz=UTC) >= self.expiration

expiration = None class-attribute instance-attribute

The expiration time of the identity.

If time zone is provided, it is updated to UTC. The value must always be in UTC.

is_expired property

Whether the identity is expired.

retries

ErrorRetryInfo

Bases: Protocol

A protocol for errors that have retry information embedded.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

@runtime_checkable
class ErrorRetryInfo(Protocol):
    """A protocol for errors that have retry information embedded."""

    is_retry_safe: bool | None = None
    """Whether the error is safe to retry.

    A value of True does not mean a retry will occur, but rather that a retry is allowed
    to occur.

    A value of None indicates that there is not enough information available to
    determine if a retry is safe.
    """

    retry_after: float | None = None
    """The amount of time that should pass before a retry.

    Retry strategies MAY choose to wait longer.
    """

    is_throttling_error: bool = False
    """Whether the error is a throttling error."""

is_retry_safe = None class-attribute instance-attribute

Whether the error is safe to retry.

A value of True does not mean a retry will occur, but rather that a retry is allowed to occur.

A value of None indicates that there is not enough information available to determine if a retry is safe.

is_throttling_error = False class-attribute instance-attribute

Whether the error is a throttling error.

retry_after = None class-attribute instance-attribute

The amount of time that should pass before a retry.

Retry strategies MAY choose to wait longer.

RetryBackoffStrategy

Bases: Protocol

Stateless strategy for computing retry delays based on retry attempt account.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

class RetryBackoffStrategy(Protocol):
    """Stateless strategy for computing retry delays based on retry attempt account."""

    def compute_next_backoff_delay(self, retry_attempt: int) -> float:
        """Calculate timespan in seconds to delay before next retry.

        :param retry_attempt: The index of the retry attempt that is about to be made
        after the delay. The initial attempt, before any retries, is index ``0``, the
        first retry attempt after the initial attempt failed is index ``1``, and so on.
        """
        ...

compute_next_backoff_delay(retry_attempt)

Calculate timespan in seconds to delay before next retry.

:param retry_attempt: The index of the retry attempt that is about to be made after the delay. The initial attempt, before any retries, is index 0, the first retry attempt after the initial attempt failed is index 1, and so on.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

def compute_next_backoff_delay(self, retry_attempt: int) -> float:
    """Calculate timespan in seconds to delay before next retry.

    :param retry_attempt: The index of the retry attempt that is about to be made
    after the delay. The initial attempt, before any retries, is index ``0``, the
    first retry attempt after the initial attempt failed is index ``1``, and so on.
    """
    ...

RetryStrategy

Bases: Protocol

Issuer of :py:class:RetryTokens.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

class RetryStrategy(Protocol):
    """Issuer of :py:class:`RetryToken`s."""

    backoff_strategy: RetryBackoffStrategy
    """The strategy used by returned tokens to compute delay duration values."""

    max_attempts: int
    """Upper limit on total attempt count (initial attempt plus retries)."""

    def acquire_initial_retry_token(
        self, *, token_scope: str | None = None
    ) -> RetryToken:
        """Called before any retries (for the first attempt at the operation).

        :param token_scope: An arbitrary string accepted by the retry strategy to
            separate tokens into scopes.
        :returns: A retry token, to be used for determining the retry delay, refreshing
            the token after a failure, and recording success after success.
        :raises RetryError: If the retry strategy has no available tokens.
        """
        ...

    def refresh_retry_token_for_retry(
        self, *, token_to_renew: RetryToken, error: Exception
    ) -> RetryToken:
        """Replace an existing retry token from a failed attempt with a new token.

        After a failed operation call, this method is called to exchange a retry token
        that was previously obtained by calling :py:func:`acquire_initial_retry_token`
        or this method with a new retry token for the next attempt. This method can
        either choose to allow another retry and send a new or updated token, or reject
        the retry attempt and raise the error.

        :param token_to_renew: The token used for the previous failed attempt.
        :param error: The error that triggered the need for a retry.
        :raises RetryError: If no further retry attempts are allowed.
        """
        ...

    def record_success(self, *, token: RetryToken) -> None:
        """Return token after successful completion of an operation.

        Upon successful completion of the operation, a user calls this function to
        record that the operation was successful.

        :param token: The token used for the previous successful attempt.
        """
        ...

backoff_strategy instance-attribute

The strategy used by returned tokens to compute delay duration values.

max_attempts instance-attribute

Upper limit on total attempt count (initial attempt plus retries).

acquire_initial_retry_token(*, token_scope=None)

Called before any retries (for the first attempt at the operation).

:param token_scope: An arbitrary string accepted by the retry strategy to separate tokens into scopes. :returns: A retry token, to be used for determining the retry delay, refreshing the token after a failure, and recording success after success. :raises RetryError: If the retry strategy has no available tokens.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

def acquire_initial_retry_token(
    self, *, token_scope: str | None = None
) -> RetryToken:
    """Called before any retries (for the first attempt at the operation).

    :param token_scope: An arbitrary string accepted by the retry strategy to
        separate tokens into scopes.
    :returns: A retry token, to be used for determining the retry delay, refreshing
        the token after a failure, and recording success after success.
    :raises RetryError: If the retry strategy has no available tokens.
    """
    ...

record_success(*, token)

Return token after successful completion of an operation.

Upon successful completion of the operation, a user calls this function to record that the operation was successful.

:param token: The token used for the previous successful attempt.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

def record_success(self, *, token: RetryToken) -> None:
    """Return token after successful completion of an operation.

    Upon successful completion of the operation, a user calls this function to
    record that the operation was successful.

    :param token: The token used for the previous successful attempt.
    """
    ...

refresh_retry_token_for_retry(*, token_to_renew, error)

Replace an existing retry token from a failed attempt with a new token.

After a failed operation call, this method is called to exchange a retry token that was previously obtained by calling :py:func:acquire_initial_retry_token or this method with a new retry token for the next attempt. This method can either choose to allow another retry and send a new or updated token, or reject the retry attempt and raise the error.

:param token_to_renew: The token used for the previous failed attempt. :param error: The error that triggered the need for a retry. :raises RetryError: If no further retry attempts are allowed.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

def refresh_retry_token_for_retry(
    self, *, token_to_renew: RetryToken, error: Exception
) -> RetryToken:
    """Replace an existing retry token from a failed attempt with a new token.

    After a failed operation call, this method is called to exchange a retry token
    that was previously obtained by calling :py:func:`acquire_initial_retry_token`
    or this method with a new retry token for the next attempt. This method can
    either choose to allow another retry and send a new or updated token, or reject
    the retry attempt and raise the error.

    :param token_to_renew: The token used for the previous failed attempt.
    :param error: The error that triggered the need for a retry.
    :raises RetryError: If no further retry attempts are allowed.
    """
    ...

RetryToken dataclass

Bases: Protocol

Token issued by a :py:class:RetryStrategy for the next attempt.

Source code in packages/smithy-core/src/smithy_core/interfaces/retries.py

@dataclass(kw_only=True)
class RetryToken(Protocol):
    """Token issued by a :py:class:`RetryStrategy` for the next attempt."""

    retry_count: int
    """Retry count is the total number of attempts minus the initial attempt."""

    retry_delay: float
    """Delay in seconds to wait before the retry attempt."""

retry_count instance-attribute

Retry count is the total number of attempts minus the initial attempt.

retry_delay instance-attribute

Delay in seconds to wait before the retry attempt.

prelude

Shared schemas for shapes built into Smithy's prelude.

retries

ExponentialBackoffJitterType

Bases: Enum

Jitter mode for exponential backoff.

For use with :py:class:ExponentialRetryBackoffStrategy.

Source code in packages/smithy-core/src/smithy_core/retries.py

class ExponentialBackoffJitterType(Enum):
    """Jitter mode for exponential backoff.

    For use with :py:class:`ExponentialRetryBackoffStrategy`.
    """

    DEFAULT = 1
    """Truncated binary exponential backoff delay with equal jitter:

    .. code-block:: python

        capped = min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))
        (capped / 2) + random_between(0, capped / 2)

    Also known as "Equal Jitter". Similar to :py:var:`FULL` but always keep some of the
    backoff and jitters by a smaller amount.
    """

    NONE = 2
    """Truncated binary exponential backoff delay without jitter:

    .. code-block:: python

        min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))
    """

    FULL = 3
    """Truncated binary exponential backoff delay with full jitter:

    .. code-block:: python

        random_between(
            max_backoff,
            min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))
        )
    """

    DECORRELATED = 4
    """Truncated binary exponential backoff delay with decorrelated jitter:

    .. code-block:: python

        min(max_backoff, random_between(backoff_scale_value, t_(i-1) * 3))

    Similar to :py:var:`FULL`, but also increases the maximum jitter at each retry.
    """

DECORRELATED = 4 class-attribute instance-attribute

Truncated binary exponential backoff delay with decorrelated jitter:

.. code-block:: python

min(max_backoff, random_between(backoff_scale_value, t_(i-1) * 3))

Similar to :py:var:FULL, but also increases the maximum jitter at each retry.

DEFAULT = 1 class-attribute instance-attribute

Truncated binary exponential backoff delay with equal jitter:

.. code-block:: python

capped = min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))
(capped / 2) + random_between(0, capped / 2)

Also known as "Equal Jitter". Similar to :py:var:FULL but always keep some of the backoff and jitters by a smaller amount.

FULL = 3 class-attribute instance-attribute

Truncated binary exponential backoff delay with full jitter:

.. code-block:: python

random_between(
    max_backoff,
    min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))
)

NONE = 2 class-attribute instance-attribute

Truncated binary exponential backoff delay without jitter:

.. code-block:: python

min(max_backoff, backoff_scale_value * 2 ** (retry_attempt - 1))

ExponentialRetryBackoffStrategy

Bases: RetryBackoffStrategy

Source code in packages/smithy-core/src/smithy_core/retries.py

class ExponentialRetryBackoffStrategy(retries_interface.RetryBackoffStrategy):
    def __init__(
        self,
        *,
        backoff_scale_value: float = 0.025,
        max_backoff: float = 20,
        jitter_type: ExponentialBackoffJitterType = ExponentialBackoffJitterType.DEFAULT,
        random: Callable[[], float] = random.random,
    ):
        """Exponential backoff with optional jitter.

        .. seealso:: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

        :param backoff_scale_value: Factor that linearly adjusts returned backoff delay
        values. See the methods ``_next_delay_*`` for the formula used to calculate the
        delay for each jitter type. If set to ``None`` (the default), :py:attr:`random`
        will be called to generate a value.

        :param max_backoff: Upper limit for backoff delay values returned, in seconds.

        :param jitter_type: Determines the formula used to apply jitter to the backoff
        delay.

        :param random: A callable that returns random numbers between ``0`` and ``1``.
        Use the default ``random.random`` unless you require an alternate source of
        randomness or a non-uniform distribution.
        """
        self._backoff_scale_value = backoff_scale_value
        self._max_backoff = max_backoff
        self._jitter_type = jitter_type
        self._random = random
        self._previous_delay_seconds = self._backoff_scale_value

    def compute_next_backoff_delay(self, retry_attempt: int) -> float:
        """Calculate timespan in seconds to delay before next retry.

        See the methods ``_next_delay_*`` for the formula used to calculate the delay
        for each jitter type for values of ``retry_attempt > 0``.

        :param retry_attempt: The index of the retry attempt that is about to be made
        after the delay. The initial attempt, before any retries, is index ``0``, and
        will return a delay of ``0``. The first retry attempt after a failed initial
        attempt is index ``1``, and so on.
        """
        if retry_attempt == 0:
            return 0

        match self._jitter_type:
            case ExponentialBackoffJitterType.NONE:
                seconds = self._next_delay_no_jitter(retry_attempt=retry_attempt)
            case ExponentialBackoffJitterType.DEFAULT:
                seconds = self._next_delay_equal_jitter(retry_attempt=retry_attempt)
            case ExponentialBackoffJitterType.FULL:
                seconds = self._next_delay_full_jitter(retry_attempt=retry_attempt)
            case ExponentialBackoffJitterType.DECORRELATED:
                seconds = self._next_delay_decorrelated_jitter(
                    previous_delay=self._previous_delay_seconds
                )

        self._previous_delay_seconds = seconds
        return seconds

    def _jitter_free_uncapped_delay(self, retry_attempt: int) -> float:
        """The basic exponential delay without jitter or upper bound:

        .. code-block:: python

            backoff_scale_value * 2 ** (retry_attempt - 1)
        """
        return self._backoff_scale_value * (2.0 ** (retry_attempt - 1))

    def _next_delay_no_jitter(self, retry_attempt: int) -> float:
        """Calculates truncated binary exponential backoff delay without jitter.

        Used when :py:var:`jitter_type` is :py:attr:`ExponentialBackoffJitterType.NONE`.
        """
        no_jitter_delay = self._jitter_free_uncapped_delay(retry_attempt)
        return min(no_jitter_delay, self._max_backoff)

    def _next_delay_full_jitter(self, retry_attempt: int) -> float:
        """Calculates truncated binary exponential backoff delay with full jitter.

        Used when :py:var:`jitter_type` is :py:attr:`ExponentialBackoffJitterType.FULL`.
        """

        no_jitter_delay = self._jitter_free_uncapped_delay(retry_attempt)
        return self._random() * min(no_jitter_delay, self._max_backoff)

    def _next_delay_equal_jitter(self, retry_attempt: int) -> float:
        """Calculates truncated binary exponential backoff delay with equal jitter:

        Used when :py:var:`jitter_type` is
        :py:attr:`ExponentialBackoffJitterType.DEFAULT`.
        """
        no_jitter_delay = self._jitter_free_uncapped_delay(retry_attempt)
        return (self._random() * 0.5 + 0.5) * min(no_jitter_delay, self._max_backoff)

    def _next_delay_decorrelated_jitter(self, previous_delay: float) -> float:
        """Calculates truncated binary exp. backoff delay with decorrelated jitter:

        Used when :py:var:`jitter_type` is
        :py:attr:`ExponentialBackoffJitterType.DECORRELATED`.
        """
        return min(
            self._backoff_scale_value + self._random() * previous_delay * 3,
            self._max_backoff,
        )

__init__(*, backoff_scale_value=0.025, max_backoff=20, jitter_type=ExponentialBackoffJitterType.DEFAULT, random=random.random)

Exponential backoff with optional jitter.

.. seealso:: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

:param backoff_scale_value: Factor that linearly adjusts returned backoff delay values. See the methods _next_delay_* for the formula used to calculate the delay for each jitter type. If set to None (the default), :py:attr:random will be called to generate a value.

:param max_backoff: Upper limit for backoff delay values returned, in seconds.

:param jitter_type: Determines the formula used to apply jitter to the backoff delay.

:param random: A callable that returns random numbers between 0 and 1. Use the default random.random unless you require an alternate source of randomness or a non-uniform distribution.

Source code in packages/smithy-core/src/smithy_core/retries.py

def __init__(
    self,
    *,
    backoff_scale_value: float = 0.025,
    max_backoff: float = 20,
    jitter_type: ExponentialBackoffJitterType = ExponentialBackoffJitterType.DEFAULT,
    random: Callable[[], float] = random.random,
):
    """Exponential backoff with optional jitter.

    .. seealso:: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

    :param backoff_scale_value: Factor that linearly adjusts returned backoff delay
    values. See the methods ``_next_delay_*`` for the formula used to calculate the
    delay for each jitter type. If set to ``None`` (the default), :py:attr:`random`
    will be called to generate a value.

    :param max_backoff: Upper limit for backoff delay values returned, in seconds.

    :param jitter_type: Determines the formula used to apply jitter to the backoff
    delay.

    :param random: A callable that returns random numbers between ``0`` and ``1``.
    Use the default ``random.random`` unless you require an alternate source of
    randomness or a non-uniform distribution.
    """
    self._backoff_scale_value = backoff_scale_value
    self._max_backoff = max_backoff
    self._jitter_type = jitter_type
    self._random = random
    self._previous_delay_seconds = self._backoff_scale_value

compute_next_backoff_delay(retry_attempt)

Calculate timespan in seconds to delay before next retry.

See the methods _next_delay_* for the formula used to calculate the delay for each jitter type for values of retry_attempt > 0.

:param retry_attempt: The index of the retry attempt that is about to be made after the delay. The initial attempt, before any retries, is index 0, and will return a delay of 0. The first retry attempt after a failed initial attempt is index 1, and so on.

Source code in packages/smithy-core/src/smithy_core/retries.py

def compute_next_backoff_delay(self, retry_attempt: int) -> float:
    """Calculate timespan in seconds to delay before next retry.

    See the methods ``_next_delay_*`` for the formula used to calculate the delay
    for each jitter type for values of ``retry_attempt > 0``.

    :param retry_attempt: The index of the retry attempt that is about to be made
    after the delay. The initial attempt, before any retries, is index ``0``, and
    will return a delay of ``0``. The first retry attempt after a failed initial
    attempt is index ``1``, and so on.
    """
    if retry_attempt == 0:
        return 0

    match self._jitter_type:
        case ExponentialBackoffJitterType.NONE:
            seconds = self._next_delay_no_jitter(retry_attempt=retry_attempt)
        case ExponentialBackoffJitterType.DEFAULT:
            seconds = self._next_delay_equal_jitter(retry_attempt=retry_attempt)
        case ExponentialBackoffJitterType.FULL:
            seconds = self._next_delay_full_jitter(retry_attempt=retry_attempt)
        case ExponentialBackoffJitterType.DECORRELATED:
            seconds = self._next_delay_decorrelated_jitter(
                previous_delay=self._previous_delay_seconds
            )

    self._previous_delay_seconds = seconds
    return seconds

SimpleRetryStrategy

Bases: RetryStrategy

Source code in packages/smithy-core/src/smithy_core/retries.py

class SimpleRetryStrategy(retries_interface.RetryStrategy):
    def __init__(
        self,
        *,
        backoff_strategy: retries_interface.RetryBackoffStrategy | None = None,
        max_attempts: int = 5,
    ):
        """Basic retry strategy that simply invokes the given backoff strategy.

        :param backoff_strategy: The backoff strategy used by returned tokens to compute
        the retry delay. Defaults to :py:class:`ExponentialRetryBackoffStrategy`.

        :param max_attempts: Upper limit on total number of attempts made, including
        initial attempt and retries.
        """
        self.backoff_strategy = backoff_strategy or ExponentialRetryBackoffStrategy()
        self.max_attempts = max_attempts

    def acquire_initial_retry_token(
        self, *, token_scope: str | None = None
    ) -> SimpleRetryToken:
        """Called before any retries (for the first attempt at the operation).

        :param token_scope: This argument is ignored by this retry strategy.
        """
        retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
        return SimpleRetryToken(retry_count=0, retry_delay=retry_delay)

    def refresh_retry_token_for_retry(
        self,
        *,
        token_to_renew: retries_interface.RetryToken,
        error: Exception,
    ) -> SimpleRetryToken:
        """Replace an existing retry token from a failed attempt with a new token.

        This retry strategy always returns a token until the attempt count stored in
        the new token exceeds the ``max_attempts`` value.

        :param token_to_renew: The token used for the previous failed attempt.
        :param error: The error that triggered the need for a retry.
        :raises RetryError: If no further retry attempts are allowed.
        """
        if isinstance(error, retries_interface.ErrorRetryInfo) and error.is_retry_safe:
            retry_count = token_to_renew.retry_count + 1
            if retry_count >= self.max_attempts:
                raise RetryError(
                    f"Reached maximum number of allowed attempts: {self.max_attempts}"
                ) from error
            retry_delay = self.backoff_strategy.compute_next_backoff_delay(retry_count)
            return SimpleRetryToken(retry_count=retry_count, retry_delay=retry_delay)
        else:
            raise RetryError(f"Error is not retryable: {error}") from error

    def record_success(self, *, token: retries_interface.RetryToken) -> None:
        """Not used by this retry strategy."""

__init__(*, backoff_strategy=None, max_attempts=5)

Basic retry strategy that simply invokes the given backoff strategy.

:param backoff_strategy: The backoff strategy used by returned tokens to compute the retry delay. Defaults to :py:class:ExponentialRetryBackoffStrategy.

:param max_attempts: Upper limit on total number of attempts made, including initial attempt and retries.

Source code in packages/smithy-core/src/smithy_core/retries.py

def __init__(
    self,
    *,
    backoff_strategy: retries_interface.RetryBackoffStrategy | None = None,
    max_attempts: int = 5,
):
    """Basic retry strategy that simply invokes the given backoff strategy.

    :param backoff_strategy: The backoff strategy used by returned tokens to compute
    the retry delay. Defaults to :py:class:`ExponentialRetryBackoffStrategy`.

    :param max_attempts: Upper limit on total number of attempts made, including
    initial attempt and retries.
    """
    self.backoff_strategy = backoff_strategy or ExponentialRetryBackoffStrategy()
    self.max_attempts = max_attempts

acquire_initial_retry_token(*, token_scope=None)

Called before any retries (for the first attempt at the operation).

:param token_scope: This argument is ignored by this retry strategy.

Source code in packages/smithy-core/src/smithy_core/retries.py

def acquire_initial_retry_token(
    self, *, token_scope: str | None = None
) -> SimpleRetryToken:
    """Called before any retries (for the first attempt at the operation).

    :param token_scope: This argument is ignored by this retry strategy.
    """
    retry_delay = self.backoff_strategy.compute_next_backoff_delay(0)
    return SimpleRetryToken(retry_count=0, retry_delay=retry_delay)

record_success(*, token)

Not used by this retry strategy.

Source code in packages/smithy-core/src/smithy_core/retries.py

def record_success(self, *, token: retries_interface.RetryToken) -> None:
    """Not used by this retry strategy."""

refresh_retry_token_for_retry(*, token_to_renew, error)

Replace an existing retry token from a failed attempt with a new token.

This retry strategy always returns a token until the attempt count stored in the new token exceeds the max_attempts value.

:param token_to_renew: The token used for the previous failed attempt. :param error: The error that triggered the need for a retry. :raises RetryError: If no further retry attempts are allowed.

Source code in packages/smithy-core/src/smithy_core/retries.py

def refresh_retry_token_for_retry(
    self,
    *,
    token_to_renew: retries_interface.RetryToken,
    error: Exception,
) -> SimpleRetryToken:
    """Replace an existing retry token from a failed attempt with a new token.

    This retry strategy always returns a token until the attempt count stored in
    the new token exceeds the ``max_attempts`` value.

    :param token_to_renew: The token used for the previous failed attempt.
    :param error: The error that triggered the need for a retry.
    :raises RetryError: If no further retry attempts are allowed.
    """
    if isinstance(error, retries_interface.ErrorRetryInfo) and error.is_retry_safe:
        retry_count = token_to_renew.retry_count + 1
        if retry_count >= self.max_attempts:
            raise RetryError(
                f"Reached maximum number of allowed attempts: {self.max_attempts}"
            ) from error
        retry_delay = self.backoff_strategy.compute_next_backoff_delay(retry_count)
        return SimpleRetryToken(retry_count=retry_count, retry_delay=retry_delay)
    else:
        raise RetryError(f"Error is not retryable: {error}") from error

SimpleRetryToken dataclass

Basic retry token that stores only the attempt count and backoff strategy.

Retry tokens should always be obtained from an implementation of :py:class:retries_interface.RetryStrategy.

Source code in packages/smithy-core/src/smithy_core/retries.py

@dataclass(kw_only=True)
class SimpleRetryToken:
    """Basic retry token that stores only the attempt count and backoff strategy.

    Retry tokens should always be obtained from an implementation of
    :py:class:`retries_interface.RetryStrategy`.
    """

    retry_count: int
    """Retry count is the total number of attempts minus the initial attempt."""

    retry_delay: float
    """Delay in seconds to wait before the retry attempt."""

    @property
    def attempt_count(self) -> int:
        """The total number of attempts including the initial attempt and retries."""
        return self.retry_count + 1

attempt_count property

The total number of attempts including the initial attempt and retries.

retry_count instance-attribute

Retry count is the total number of attempts minus the initial attempt.

retry_delay instance-attribute

Delay in seconds to wait before the retry attempt.

rfc3986

Module vended from rfc3986 abnf_rexexp.py and misc.py.

https://github.com/python-hyper/rfc3986/blob/main/src/rfc3986/abnf_regexp.py https://github.com/python-hyper/rfc3986/blob/main/src/rfc3986/misc.py

schemas

APIOperation dataclass

A modeled Smithy operation.

Source code in packages/smithy-core/src/smithy_core/schemas.py

@dataclass(kw_only=True, frozen=True)
class APIOperation[I: "SerializeableShape", O: "DeserializeableShape"]:
    """A modeled Smithy operation."""

    input: type[I]
    """The input type of the operation."""

    output: type[O]
    """The output type of the operation."""

    schema: Schema = field(repr=False)
    """The schema of the operation."""

    input_schema: Schema = field(repr=False)
    """The schema of the operation's input shape."""

    output_schema: Schema = field(repr=False)
    """The schema of the operation's output shape."""

    error_registry: "TypeRegistry"
    """A TypeRegistry used to create errors."""

    effective_auth_schemes: Sequence[ShapeID]
    """A list of effective auth schemes for the operation."""

    @property
    def idempotency_token_member(self) -> Schema | None:
        """The input schema member that serves as the idempotency token."""
        for member in self.input_schema.members.values():
            if member.get_trait(IdempotencyTokenTrait) is not None:
                return member
        return None

    @property
    def input_stream_member(self) -> Schema | None:
        """The input schema member that contains an event stream or data stream."""
        for member in self.input_schema.members.values():
            if member.get_trait(StreamingTrait) is not None:
                return member
        return None

    @property
    def output_stream_member(self) -> Schema | None:
        """The output schema member that contains an event stream or data stream."""
        for member in self.output_schema.members.values():
            if member.get_trait(StreamingTrait) is not None:
                return member
        return None

effective_auth_schemes instance-attribute

A list of effective auth schemes for the operation.

error_registry instance-attribute

A TypeRegistry used to create errors.

idempotency_token_member property

The input schema member that serves as the idempotency token.

input instance-attribute

The input type of the operation.

input_schema = field(repr=False) class-attribute instance-attribute

The schema of the operation's input shape.

input_stream_member property

The input schema member that contains an event stream or data stream.

output instance-attribute

The output type of the operation.

output_schema = field(repr=False) class-attribute instance-attribute

The schema of the operation's output shape.

output_stream_member property

The output schema member that contains an event stream or data stream.

schema = field(repr=False) class-attribute instance-attribute

The schema of the operation.

MemberSchema

Bases: TypedDict

A simplified schema for members.

This is only used for :ref:Schema.collection.

:param target: The target of the member. :param traits: An optional list of traits for the member.

Source code in packages/smithy-core/src/smithy_core/schemas.py

class MemberSchema(TypedDict):
    """A simplified schema for members.

    This is only used for :ref:`Schema.collection`.

    :param target: The target of the member.
    :param traits: An optional list of traits for the member.
    """

    target: Required[Schema]
    traits: NotRequired[list["Trait | DynamicTrait"]]

Schema dataclass

Describes a shape, its traits, and its members.

Source code in packages/smithy-core/src/smithy_core/schemas.py

@dataclass(kw_only=True, frozen=True, init=False)
class Schema:
    """Describes a shape, its traits, and its members."""

    id: ShapeID
    shape_type: ShapeType
    traits: dict[ShapeID, "Trait | DynamicTrait"] = field(
        default_factory=dict[ShapeID, "Trait | DynamicTrait"]
    )
    members: dict[str, "Schema"] = field(default_factory=dict[str, "Schema"])
    member_target: "Schema | None" = None
    member_index: int | None = None

    def __init__(
        self,
        *,
        id: ShapeID,
        shape_type: ShapeType,
        traits: list["Trait | DynamicTrait"]
        | dict[ShapeID, "Trait | DynamicTrait"]
        | None = None,
        members: list["Schema"] | dict[str, "Schema"] | None = None,
        member_target: "Schema | None" = None,
        member_index: int | None = None,
    ) -> None:
        """Initialize a schema.

        :param id: The ID of the shape.
        :param shape_type: The type of the shape.
        :param traits: Traits applied to the shape, which describe additional metadata.
        :param members: Members of the shape. These correspond to list contents,
            dataclass properties, map keys/values, and union variants.
        :param member_target: The schema that the member points to, if the shape is the
            MEMBER type.
        :param member_index: The index of the member, if the shape is the MEMBER type.
            This is used for faster match checks when all members of a shape are known.
        """
        _member_props = [
            id.member is not None,
            member_target is not None,
            member_index is not None,
        ]
        if any(_member_props) and not all(_member_props):
            raise SmithyError(
                "If any member property is set, all member properties must be set. "
                f"member_name: {id.member!r}, member_target: "
                f"{member_target!r}, member_index: {member_index!r}"
            )

        # setattr is required because the class is frozen
        object.__setattr__(self, "id", id)
        object.__setattr__(self, "shape_type", shape_type)

        if traits:
            if isinstance(traits, list):
                traits = {t.id: t for t in traits}
        else:
            traits = {}
        object.__setattr__(self, "traits", traits)

        if members is None:
            members = {}
        elif isinstance(members, list):
            m: dict[str, Schema] = {}
            for member in sorted(members, key=lambda m: m.expect_member_index()):
                m[member.expect_member_name()] = member
            members = m
        object.__setattr__(self, "members", members)

        if member_target is not None:
            object.__setattr__(self, "member_target", member_target)

        if member_index is not None:
            object.__setattr__(self, "member_index", member_index)

    @property
    def member_name(self) -> str | None:
        """The name of the member, if the shape is the MEMBER type."""
        return self.id.member

    def expect_member_name(self) -> str:
        """Assert the schema is a member schema and return its member name.

        :raises ExpectationNotMetError: If member_name wasn't set.
        :returns: Returns the member name.
        """
        return self.id.expect_member()

    def expect_member_target(self) -> "Schema":
        """Assert the schema is a member schema and return its target.

        If the target is a class containing a schema, the schema is extracted and
        returned.

        :raises ExpectationNotMetError: If member_target wasn't set.
        :returns: Returns the target schema.
        """
        if self.member_target is None:
            raise ExpectationNotMetError(
                "Expected member_target to be set, but was None."
            )
        return self.member_target

    def expect_member_index(self) -> int:
        """Assert the schema is a member schema and return its member index.

        :raises ExpectationNotMetError: If member_index wasn't set.
        :returns: Returns the member index.
        """
        if self.member_index is None:
            raise ExpectationNotMetError(
                "Expected member_index to be set, but was None."
            )
        return self.member_index

    @overload
    def get_trait[T: "Trait"](self, t: type[T]) -> T | None: ...

    @overload
    def get_trait(self, t: ShapeID) -> "Trait | DynamicTrait | None": ...

    def get_trait(self, t: "type[Trait] | ShapeID") -> "Trait | DynamicTrait | None":
        """Get a trait based on its ShapeID or class.

        :returns: A Trait if the trait class is known, a DynamicTrait if it isn't, or
            None if the trait is not present on the Schema.
        """
        if isinstance(t, ShapeID):
            return self.traits.get(t)

        result = self.traits.get(t.id)

        # If the trait wasn't known when the schema was created, but is known now, go
        # ahead and convert it.
        if isinstance(result, DynamicTrait):
            result = t(result)
            self.traits[t.id] = result

        return result

    @overload
    def expect_trait[T: "Trait"](self, t: type[T]) -> T: ...

    @overload
    def expect_trait(self, t: ShapeID) -> "Trait | DynamicTrait": ...

    def expect_trait(self, t: "type[Trait] | ShapeID") -> "Trait | DynamicTrait":
        """Get a trait based on its ShapeID or class.

        :returns: A Trait if the trait class is known, a DynamicTrait if it isn't.
        """
        id = t if isinstance(t, ShapeID) else t.id
        return self.traits[id]

    def __contains__(self, item: Any):
        """Returns whether the schema has the given member or trait."""
        match item:
            case type():
                if issubclass(item, Trait):
                    return item.id in self.traits
                return False
            case ShapeID():
                if (member := item.member) is not None:
                    if self.id.with_member(member) == item:
                        return member in self.members
                    return False
                return item in self.traits
            case str():
                return item in self.members
            case _:
                return False

    @classmethod
    def collection(
        cls,
        *,
        id: ShapeID,
        shape_type: ShapeType = ShapeType.STRUCTURE,
        traits: list["Trait | DynamicTrait"] | None = None,
        members: Mapping[str, "MemberSchema | None"] | None = None,
    ) -> Self:
        """Create a schema for a collection shape.

        :param id: The ID of the shape.
        :param shape_type: The type of the shape. Defaults to STRUCTURE.
        :param traits: Traits applied to the shape, which describe additional metadata.
        :param members: Members of the shape. These correspond to list contents, dataclass
            properties, map keys/values, and union variants. In contrast to the main
            constructor, this is a dict of member names to a simplified dict containing
            only ``traits`` and a ``target``. Member schemas will be generated from this.

            If the value is None, it MUST be populated later. This is to allow a preservation
            of modeled order without having to explicitly provide it and therefore generate
            a ton of boilerplate.
        """
        struct_members: dict[str, Schema] = {}
        if members:
            for i, (k, member) in enumerate(members.items()):
                if member is None:
                    struct_members[k] = None  # type: ignore
                    continue

                struct_members[k] = cls.member(
                    id=id.with_member(k),
                    target=member["target"],
                    index=i,
                    member_traits=member.get("traits"),
                )

        result = cls(
            id=id,
            shape_type=shape_type,
            traits=traits,
            members=struct_members,
        )
        return result

    @classmethod
    def member(
        cls,
        id: ShapeID,
        target: "Schema",
        index: int,
        member_traits: list["Trait | DynamicTrait"] | None = None,
    ) -> "Schema":
        """Create a schema for a member shape.

        Member schemas are largely copies of the schemas they target to make it easier
        to use them. They contain all the members of the target and all of the traits of
        the target. Any traits provided to this method, will override traits of the same
        type on the member schema.

        :param id: The member's id.
        :param target: The schema the member is targeting.
        :param index: The member's index.
        """
        id.expect_member()
        if target.member_target is not None:
            raise ExpectationNotMetError("Member targets must not be members.")
        resolved_traits = target.traits.copy()
        if member_traits:
            resolved_traits.update({t.id: t for t in member_traits})
        return replace(
            target,
            id=id,
            traits=resolved_traits,
            member_target=target,
            member_index=index,
        )

member_name property

The name of the member, if the shape is the MEMBER type.

__contains__(item)

Returns whether the schema has the given member or trait.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def __contains__(self, item: Any):
    """Returns whether the schema has the given member or trait."""
    match item:
        case type():
            if issubclass(item, Trait):
                return item.id in self.traits
            return False
        case ShapeID():
            if (member := item.member) is not None:
                if self.id.with_member(member) == item:
                    return member in self.members
                return False
            return item in self.traits
        case str():
            return item in self.members
        case _:
            return False

__init__(*, id, shape_type, traits=None, members=None, member_target=None, member_index=None)

Initialize a schema.

:param id: The ID of the shape. :param shape_type: The type of the shape. :param traits: Traits applied to the shape, which describe additional metadata. :param members: Members of the shape. These correspond to list contents, dataclass properties, map keys/values, and union variants. :param member_target: The schema that the member points to, if the shape is the MEMBER type. :param member_index: The index of the member, if the shape is the MEMBER type. This is used for faster match checks when all members of a shape are known.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def __init__(
    self,
    *,
    id: ShapeID,
    shape_type: ShapeType,
    traits: list["Trait | DynamicTrait"]
    | dict[ShapeID, "Trait | DynamicTrait"]
    | None = None,
    members: list["Schema"] | dict[str, "Schema"] | None = None,
    member_target: "Schema | None" = None,
    member_index: int | None = None,
) -> None:
    """Initialize a schema.

    :param id: The ID of the shape.
    :param shape_type: The type of the shape.
    :param traits: Traits applied to the shape, which describe additional metadata.
    :param members: Members of the shape. These correspond to list contents,
        dataclass properties, map keys/values, and union variants.
    :param member_target: The schema that the member points to, if the shape is the
        MEMBER type.
    :param member_index: The index of the member, if the shape is the MEMBER type.
        This is used for faster match checks when all members of a shape are known.
    """
    _member_props = [
        id.member is not None,
        member_target is not None,
        member_index is not None,
    ]
    if any(_member_props) and not all(_member_props):
        raise SmithyError(
            "If any member property is set, all member properties must be set. "
            f"member_name: {id.member!r}, member_target: "
            f"{member_target!r}, member_index: {member_index!r}"
        )

    # setattr is required because the class is frozen
    object.__setattr__(self, "id", id)
    object.__setattr__(self, "shape_type", shape_type)

    if traits:
        if isinstance(traits, list):
            traits = {t.id: t for t in traits}
    else:
        traits = {}
    object.__setattr__(self, "traits", traits)

    if members is None:
        members = {}
    elif isinstance(members, list):
        m: dict[str, Schema] = {}
        for member in sorted(members, key=lambda m: m.expect_member_index()):
            m[member.expect_member_name()] = member
        members = m
    object.__setattr__(self, "members", members)

    if member_target is not None:
        object.__setattr__(self, "member_target", member_target)

    if member_index is not None:
        object.__setattr__(self, "member_index", member_index)

collection(*, id, shape_type=ShapeType.STRUCTURE, traits=None, members=None) classmethod

Create a schema for a collection shape.

:param id: The ID of the shape. :param shape_type: The type of the shape. Defaults to STRUCTURE. :param traits: Traits applied to the shape, which describe additional metadata. :param members: Members of the shape. These correspond to list contents, dataclass properties, map keys/values, and union variants. In contrast to the main constructor, this is a dict of member names to a simplified dict containing only traits and a target. Member schemas will be generated from this.

If the value is None, it MUST be populated later. This is to allow a preservation
of modeled order without having to explicitly provide it and therefore generate
a ton of boilerplate.

Source code in packages/smithy-core/src/smithy_core/schemas.py

@classmethod
def collection(
    cls,
    *,
    id: ShapeID,
    shape_type: ShapeType = ShapeType.STRUCTURE,
    traits: list["Trait | DynamicTrait"] | None = None,
    members: Mapping[str, "MemberSchema | None"] | None = None,
) -> Self:
    """Create a schema for a collection shape.

    :param id: The ID of the shape.
    :param shape_type: The type of the shape. Defaults to STRUCTURE.
    :param traits: Traits applied to the shape, which describe additional metadata.
    :param members: Members of the shape. These correspond to list contents, dataclass
        properties, map keys/values, and union variants. In contrast to the main
        constructor, this is a dict of member names to a simplified dict containing
        only ``traits`` and a ``target``. Member schemas will be generated from this.

        If the value is None, it MUST be populated later. This is to allow a preservation
        of modeled order without having to explicitly provide it and therefore generate
        a ton of boilerplate.
    """
    struct_members: dict[str, Schema] = {}
    if members:
        for i, (k, member) in enumerate(members.items()):
            if member is None:
                struct_members[k] = None  # type: ignore
                continue

            struct_members[k] = cls.member(
                id=id.with_member(k),
                target=member["target"],
                index=i,
                member_traits=member.get("traits"),
            )

    result = cls(
        id=id,
        shape_type=shape_type,
        traits=traits,
        members=struct_members,
    )
    return result

expect_member_index()

Assert the schema is a member schema and return its member index.

:raises ExpectationNotMetError: If member_index wasn't set. :returns: Returns the member index.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def expect_member_index(self) -> int:
    """Assert the schema is a member schema and return its member index.

    :raises ExpectationNotMetError: If member_index wasn't set.
    :returns: Returns the member index.
    """
    if self.member_index is None:
        raise ExpectationNotMetError(
            "Expected member_index to be set, but was None."
        )
    return self.member_index

expect_member_name()

Assert the schema is a member schema and return its member name.

:raises ExpectationNotMetError: If member_name wasn't set. :returns: Returns the member name.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def expect_member_name(self) -> str:
    """Assert the schema is a member schema and return its member name.

    :raises ExpectationNotMetError: If member_name wasn't set.
    :returns: Returns the member name.
    """
    return self.id.expect_member()

expect_member_target()

Assert the schema is a member schema and return its target.

If the target is a class containing a schema, the schema is extracted and returned.

:raises ExpectationNotMetError: If member_target wasn't set. :returns: Returns the target schema.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def expect_member_target(self) -> "Schema":
    """Assert the schema is a member schema and return its target.

    If the target is a class containing a schema, the schema is extracted and
    returned.

    :raises ExpectationNotMetError: If member_target wasn't set.
    :returns: Returns the target schema.
    """
    if self.member_target is None:
        raise ExpectationNotMetError(
            "Expected member_target to be set, but was None."
        )
    return self.member_target

expect_trait(t)

expect_trait(t: type[T]) -> T

expect_trait(t: ShapeID) -> Trait | DynamicTrait

Get a trait based on its ShapeID or class.

:returns: A Trait if the trait class is known, a DynamicTrait if it isn't.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def expect_trait(self, t: "type[Trait] | ShapeID") -> "Trait | DynamicTrait":
    """Get a trait based on its ShapeID or class.

    :returns: A Trait if the trait class is known, a DynamicTrait if it isn't.
    """
    id = t if isinstance(t, ShapeID) else t.id
    return self.traits[id]

get_trait(t)

get_trait(t: type[T]) -> T | None

get_trait(t: ShapeID) -> Trait | DynamicTrait | None

Get a trait based on its ShapeID or class.

:returns: A Trait if the trait class is known, a DynamicTrait if it isn't, or None if the trait is not present on the Schema.

Source code in packages/smithy-core/src/smithy_core/schemas.py

def get_trait(self, t: "type[Trait] | ShapeID") -> "Trait | DynamicTrait | None":
    """Get a trait based on its ShapeID or class.

    :returns: A Trait if the trait class is known, a DynamicTrait if it isn't, or
        None if the trait is not present on the Schema.
    """
    if isinstance(t, ShapeID):
        return self.traits.get(t)

    result = self.traits.get(t.id)

    # If the trait wasn't known when the schema was created, but is known now, go
    # ahead and convert it.
    if isinstance(result, DynamicTrait):
        result = t(result)
        self.traits[t.id] = result

    return result

member(id, target, index, member_traits=None) classmethod

Create a schema for a member shape.

Member schemas are largely copies of the schemas they target to make it easier to use them. They contain all the members of the target and all of the traits of the target. Any traits provided to this method, will override traits of the same type on the member schema.

:param id: The member's id. :param target: The schema the member is targeting. :param index: The member's index.

Source code in packages/smithy-core/src/smithy_core/schemas.py

@classmethod
def member(
    cls,
    id: ShapeID,
    target: "Schema",
    index: int,
    member_traits: list["Trait | DynamicTrait"] | None = None,
) -> "Schema":
    """Create a schema for a member shape.

    Member schemas are largely copies of the schemas they target to make it easier
    to use them. They contain all the members of the target and all of the traits of
    the target. Any traits provided to this method, will override traits of the same
    type on the member schema.

    :param id: The member's id.
    :param target: The schema the member is targeting.
    :param index: The member's index.
    """
    id.expect_member()
    if target.member_target is not None:
        raise ExpectationNotMetError("Member targets must not be members.")
    resolved_traits = target.traits.copy()
    if member_traits:
        resolved_traits.update({t.id: t for t in member_traits})
    return replace(
        target,
        id=id,
        traits=resolved_traits,
        member_target=target,
        member_index=index,
    )

serializers

InterceptingSerializer

Bases: ShapeSerializer

A shape serializer capable of injecting data before writing.

This can, for example, be used to add in structure member keys, commas between structure members, or commas between lists.

Source code in packages/smithy-core/src/smithy_core/serializers.py

class InterceptingSerializer(ShapeSerializer, metaclass=ABCMeta):
    """A shape serializer capable of injecting data before writing.

    This can, for example, be used to add in structure member keys, commas between
    structure members, or commas between lists.
    """

    @abstractmethod
    def before(self, schema: "Schema") -> ShapeSerializer: ...

    @abstractmethod
    def after(self, schema: "Schema") -> None: ...

    @contextmanager
    def begin_struct(self, schema: "Schema") -> Iterator[ShapeSerializer]:
        delegate = self.before(schema)

        try:
            with delegate.begin_struct(schema) as s:
                yield s
        except Exception:
            raise
        else:
            self.after(schema)

    @contextmanager
    def begin_list(self, schema: "Schema", size: int) -> Iterator[ShapeSerializer]:
        delegate = self.before(schema)

        try:
            with delegate.begin_list(schema, size) as s:
                yield s
        except Exception:
            raise
        else:
            self.after(schema)

    @contextmanager
    def begin_map(self, schema: "Schema", size: int) -> Iterator[MapSerializer]:
        delegate = self.before(schema)

        try:
            with delegate.begin_map(schema, size) as s:
                yield s
        except Exception:
            raise
        else:
            self.after(schema)

    def write_null(self, schema: "Schema") -> None:
        self.before(schema).write_null(schema)
        self.after(schema)

    def write_boolean(self, schema: "Schema", value: bool) -> None:
        self.before(schema).write_boolean(schema, value)
        self.after(schema)

    def write_byte(self, schema: "Schema", value: int) -> None:
        self.before(schema).write_byte(schema, value)
        self.after(schema)

    def write_short(self, schema: "Schema", value: int) -> None:
        self.before(schema).write_short(schema, value)
        self.after(schema)

    def write_integer(self, schema: "Schema", value: int) -> None:
        self.before(schema).write_integer(schema, value)
        self.after(schema)

    def write_long(self, schema: "Schema", value: int) -> None:
        self.before(schema).write_long(schema, value)
        self.after(schema)

    def write_float(self, schema: "Schema", value: float) -> None:
        self.before(schema).write_float(schema, value)
        self.after(schema)

    def write_double(self, schema: "Schema", value: float) -> None:
        self.before(schema).write_double(schema, value)
        self.after(schema)

    def write_big_integer(self, schema: "Schema", value: int) -> None:
        self.before(schema).write_big_integer(schema, value)
        self.after(schema)

    def write_big_decimal(self, schema: "Schema", value: Decimal) -> None:
        self.before(schema).write_big_decimal(schema, value)
        self.after(schema)

    def write_string(self, schema: "Schema", value: str) -> None:
        self.before(schema).write_string(schema, value)
        self.after(schema)

    def write_blob(self, schema: "Schema", value: bytes) -> None:
        self.before(schema).write_blob(schema, value)
        self.after(schema)

    def write_timestamp(self, schema: "Schema", value: datetime.datetime) -> None:
        self.before(schema).write_timestamp(schema, value)
        self.after(schema)

    def write_document(self, schema: "Schema", value: "Document") -> None:
        self.before(schema).write_document(schema, value)
        self.after(schema)

    def write_data_stream(self, schema: "Schema", value: "_Stream") -> None:
        self.before(schema).write_data_stream(schema, value)
        self.after(schema)

MapSerializer

Bases: Protocol

Protocol for serializing maps.

These are responsible for writing any data needed between keys and values as well as any data needed between entries.

Source code in packages/smithy-core/src/smithy_core/serializers.py

@runtime_checkable
class MapSerializer(Protocol):
    """Protocol for serializing maps.

    These are responsible for writing any data needed between keys and values as well as
    any data needed between entries.
    """

    def entry(self, key: str, value_writer: Callable[[ShapeSerializer], None]):
        """Write a map entry.

        :param key: The entry's key.
        :param value_writer: A callable that accepts a shape serializer to write values.
        """
        ...

entry(key, value_writer)

Write a map entry.

:param key: The entry's key. :param value_writer: A callable that accepts a shape serializer to write values.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def entry(self, key: str, value_writer: Callable[[ShapeSerializer], None]):
    """Write a map entry.

    :param key: The entry's key.
    :param value_writer: A callable that accepts a shape serializer to write values.
    """
    ...

SerializeableShape

Bases: Protocol

Protocol for shapes that are serializeable using a ShapeSerializer.

Source code in packages/smithy-core/src/smithy_core/serializers.py

@runtime_checkable
class SerializeableShape(Protocol):
    """Protocol for shapes that are serializeable using a ShapeSerializer."""

    def serialize(self, serializer: ShapeSerializer) -> None:
        """Serialize the shape using the given serializer.

        :param serializer: The serializer to write shape data to.
        """
        ...

serialize(serializer)

Serialize the shape using the given serializer.

:param serializer: The serializer to write shape data to.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def serialize(self, serializer: ShapeSerializer) -> None:
    """Serialize the shape using the given serializer.

    :param serializer: The serializer to write shape data to.
    """
    ...

SerializeableStruct

Bases: SerializeableShape, Protocol

Protocol for structures that are serializeable using a ShapeSerializer.

Source code in packages/smithy-core/src/smithy_core/serializers.py

@runtime_checkable
class SerializeableStruct(SerializeableShape, Protocol):
    """Protocol for structures that are serializeable using a ShapeSerializer."""

    def serialize_members(self, serializer: ShapeSerializer) -> None:
        """Serialize structure members using the given serializer.

        :param serializer: The serializer to write member data to.
        """
        ...

serialize_members(serializer)

Serialize structure members using the given serializer.

:param serializer: The serializer to write member data to.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def serialize_members(self, serializer: ShapeSerializer) -> None:
    """Serialize structure members using the given serializer.

    :param serializer: The serializer to write member data to.
    """
    ...

ShapeSerializer

Bases: Protocol

Protocol for serializing shapes based on the Smithy data model.

If used as a base class, all non-float number methods default to calling write_integer and write_double defaults to calling write_float. These extra numeric methods are for types in the Smithy data model that don't have Python equivalents, but may have equivalents in the format being written.

Source code in packages/smithy-core/src/smithy_core/serializers.py

@runtime_checkable
class ShapeSerializer(Protocol):
    """Protocol for serializing shapes based on the Smithy data model.

    If used as a base class, all non-float number methods default to calling
    ``write_integer`` and ``write_double`` defaults to calling ``write_float``.
    These extra numeric methods are for types in the Smithy data model that
    don't have Python equivalents, but may have equivalents in the format
    being written.
    """

    def begin_struct(
        self, schema: "Schema"
    ) -> AbstractContextManager["ShapeSerializer"]:
        """Open a structure for writing.

        The returned context manager is responsible for closing the structure when the
        caller has finished writing members.

        The shape serializer contained in the context manager is responsible for writing
        out the member name as well as any additional data needed between the member
        name and value and between members.

        :param schema: The schema of the structure.
        :returns: A context manager containing a member serializer.
        """
        ...

    def write_struct(self, schema: "Schema", struct: "SerializeableStruct") -> None:
        """Write a structured shape to the output.

        This method is primarily intended to be used to serialize members that target
        structure or union shapes.

        :param schema: The member schema of the structure.
        :param struct: The structure to serialize.
        """
        with self.begin_struct(schema=schema) as struct_serializer:
            struct.serialize_members(struct_serializer)

    def begin_list(
        self, schema: "Schema", size: int
    ) -> AbstractContextManager["ShapeSerializer"]:
        """Open a list for writing.

        The returned context manager is responsible for closing the list when the caller
        has finished writing elements.

        The shape serializer contained in the context manager is responsible for
        inserting any data needed between elements.

        :param schema: The schema of the list.
        :param size: The size of the list.
        :returns: A context manager containing an element serializer.
        """
        ...

    def begin_map(
        self, schema: "Schema", size: int
    ) -> AbstractContextManager["MapSerializer"]:
        """Open a map for writing.

        The returned context manager is responsible for closing the map when the caller
        has finished writing members.

        The MapSerializer contained in the context manager is responsible for writing
        out any additional data needed between the entry name and value as well as any
        data needed between entries.

        :param schema: The schema of the map.
        :param size: The size of the map.
        :returns: A context manager containing a map serializer.
        """
        ...

    def write_null(self, schema: "Schema") -> None:
        """Write a null value to the output.

        :param schema: The shape's schema.
        """
        ...

    def write_boolean(self, schema: "Schema", value: bool) -> None:
        """Write a boolean value to the output.

        :param schema: The shape's schema.
        :param value: The boolean value to write.
        """
        ...

    def write_byte(self, schema: "Schema", value: int) -> None:
        """Write a byte (8-bit integer) value to the output.

        :param schema: The shape's schema.
        :param value: The byte value to write.
        """
        self.write_integer(schema, value)

    def write_short(self, schema: "Schema", value: int) -> None:
        """Write a short (16-bit integer) value to the output.

        :param schema: The shape's schema.
        :param value: The short value to write.
        """
        self.write_integer(schema, value)

    def write_integer(self, schema: "Schema", value: int) -> None:
        """Write an integer (32-bit) value to the output.

        :param schema: The shape's schema.
        :param value: The integer value to write.
        """
        ...

    def write_long(self, schema: "Schema", value: int) -> None:
        """Write a long (64-bit integer) value to the output.

        :param schema: The shape's schema.
        :param value: The long value to write.
        """
        self.write_integer(schema, value)

    def write_float(self, schema: "Schema", value: float) -> None:
        """Write a float (32-bit) value to the output.

        :param schema: The shape's schema.
        :param value: The float value to write.
        """
        ...

    def write_double(self, schema: "Schema", value: float) -> None:
        """Write a double (64-bit float) value to the output.

        :param schema: The shape's schema.
        :param value: The double value to write.
        """
        self.write_float(schema, value)

    def write_big_integer(self, schema: "Schema", value: int) -> None:
        """Write a big integer (arbirtrarily large integer) value to the output.

        :param schema: The shape's schema.
        :param value: The big integer value to write.
        """
        self.write_integer(schema, value)

    def write_big_decimal(self, schema: "Schema", value: Decimal) -> None:
        """Write a big decimal (arbitrarily large float) value to the output.

        :param schema: The shape's schema.
        :param value: The big decimal value to write.
        """
        ...

    def write_string(self, schema: "Schema", value: str) -> None:
        """Write a string value to the output.

        :param schema: The shape's schema.
        :param value: The string value to write.
        """
        ...

    def write_blob(self, schema: "Schema", value: bytes) -> None:
        """Write a blob value to the output.

        :param schema: The shape's schema.
        :param value: The blob value to write.
        """
        ...

    def write_timestamp(self, schema: "Schema", value: datetime.datetime) -> None:
        """Write a timestamp value to the output.

        :param schema: The shape's schema.
        :param value: The timestamp value to write.
        """
        ...

    def write_document(self, schema: "Schema", value: "Document") -> None:
        """Write a document value to the output.

        :param schema: The shape's schema.
        :param value: The document value to write.
        """
        ...

    def write_data_stream(self, schema: "Schema", value: "_Stream") -> None:
        """Write a data stream to the output.

        If the value is a stream (i.e. not bytes or bytearray) it MUST NOT be read
        directly by this method. Such values are intended to only be read as needed when
        sending a message, and so should be bound directly to the request / response
        type and then read by the transport.

        Data streams are only supported at the top-level input and output for
        operations.

        :param schema: The shape's schema.
        :param value: The streaming value to write.
        """
        if isinstance(value, bytes | bytearray):
            self.write_blob(schema, bytes(value))
        raise UnsupportedStreamError()

    def flush(self) -> None:
        """Flush the underlying data."""

begin_list(schema, size)

Open a list for writing.

The returned context manager is responsible for closing the list when the caller has finished writing elements.

The shape serializer contained in the context manager is responsible for inserting any data needed between elements.

:param schema: The schema of the list. :param size: The size of the list. :returns: A context manager containing an element serializer.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def begin_list(
    self, schema: "Schema", size: int
) -> AbstractContextManager["ShapeSerializer"]:
    """Open a list for writing.

    The returned context manager is responsible for closing the list when the caller
    has finished writing elements.

    The shape serializer contained in the context manager is responsible for
    inserting any data needed between elements.

    :param schema: The schema of the list.
    :param size: The size of the list.
    :returns: A context manager containing an element serializer.
    """
    ...

begin_map(schema, size)

Open a map for writing.

The returned context manager is responsible for closing the map when the caller has finished writing members.

The MapSerializer contained in the context manager is responsible for writing out any additional data needed between the entry name and value as well as any data needed between entries.

:param schema: The schema of the map. :param size: The size of the map. :returns: A context manager containing a map serializer.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def begin_map(
    self, schema: "Schema", size: int
) -> AbstractContextManager["MapSerializer"]:
    """Open a map for writing.

    The returned context manager is responsible for closing the map when the caller
    has finished writing members.

    The MapSerializer contained in the context manager is responsible for writing
    out any additional data needed between the entry name and value as well as any
    data needed between entries.

    :param schema: The schema of the map.
    :param size: The size of the map.
    :returns: A context manager containing a map serializer.
    """
    ...

begin_struct(schema)

Open a structure for writing.

The returned context manager is responsible for closing the structure when the caller has finished writing members.

The shape serializer contained in the context manager is responsible for writing out the member name as well as any additional data needed between the member name and value and between members.

:param schema: The schema of the structure. :returns: A context manager containing a member serializer.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def begin_struct(
    self, schema: "Schema"
) -> AbstractContextManager["ShapeSerializer"]:
    """Open a structure for writing.

    The returned context manager is responsible for closing the structure when the
    caller has finished writing members.

    The shape serializer contained in the context manager is responsible for writing
    out the member name as well as any additional data needed between the member
    name and value and between members.

    :param schema: The schema of the structure.
    :returns: A context manager containing a member serializer.
    """
    ...

flush()

Flush the underlying data.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def flush(self) -> None:
    """Flush the underlying data."""

write_big_decimal(schema, value)

Write a big decimal (arbitrarily large float) value to the output.

:param schema: The shape's schema. :param value: The big decimal value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_big_decimal(self, schema: "Schema", value: Decimal) -> None:
    """Write a big decimal (arbitrarily large float) value to the output.

    :param schema: The shape's schema.
    :param value: The big decimal value to write.
    """
    ...

write_big_integer(schema, value)

Write a big integer (arbirtrarily large integer) value to the output.

:param schema: The shape's schema. :param value: The big integer value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_big_integer(self, schema: "Schema", value: int) -> None:
    """Write a big integer (arbirtrarily large integer) value to the output.

    :param schema: The shape's schema.
    :param value: The big integer value to write.
    """
    self.write_integer(schema, value)

write_blob(schema, value)

Write a blob value to the output.

:param schema: The shape's schema. :param value: The blob value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_blob(self, schema: "Schema", value: bytes) -> None:
    """Write a blob value to the output.

    :param schema: The shape's schema.
    :param value: The blob value to write.
    """
    ...

write_boolean(schema, value)

Write a boolean value to the output.

:param schema: The shape's schema. :param value: The boolean value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_boolean(self, schema: "Schema", value: bool) -> None:
    """Write a boolean value to the output.

    :param schema: The shape's schema.
    :param value: The boolean value to write.
    """
    ...

write_byte(schema, value)

Write a byte (8-bit integer) value to the output.

:param schema: The shape's schema. :param value: The byte value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_byte(self, schema: "Schema", value: int) -> None:
    """Write a byte (8-bit integer) value to the output.

    :param schema: The shape's schema.
    :param value: The byte value to write.
    """
    self.write_integer(schema, value)

write_data_stream(schema, value)

Write a data stream to the output.

If the value is a stream (i.e. not bytes or bytearray) it MUST NOT be read directly by this method. Such values are intended to only be read as needed when sending a message, and so should be bound directly to the request / response type and then read by the transport.

Data streams are only supported at the top-level input and output for operations.

:param schema: The shape's schema. :param value: The streaming value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_data_stream(self, schema: "Schema", value: "_Stream") -> None:
    """Write a data stream to the output.

    If the value is a stream (i.e. not bytes or bytearray) it MUST NOT be read
    directly by this method. Such values are intended to only be read as needed when
    sending a message, and so should be bound directly to the request / response
    type and then read by the transport.

    Data streams are only supported at the top-level input and output for
    operations.

    :param schema: The shape's schema.
    :param value: The streaming value to write.
    """
    if isinstance(value, bytes | bytearray):
        self.write_blob(schema, bytes(value))
    raise UnsupportedStreamError()

write_document(schema, value)

Write a document value to the output.

:param schema: The shape's schema. :param value: The document value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_document(self, schema: "Schema", value: "Document") -> None:
    """Write a document value to the output.

    :param schema: The shape's schema.
    :param value: The document value to write.
    """
    ...

write_double(schema, value)

Write a double (64-bit float) value to the output.

:param schema: The shape's schema. :param value: The double value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_double(self, schema: "Schema", value: float) -> None:
    """Write a double (64-bit float) value to the output.

    :param schema: The shape's schema.
    :param value: The double value to write.
    """
    self.write_float(schema, value)

write_float(schema, value)

Write a float (32-bit) value to the output.

:param schema: The shape's schema. :param value: The float value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_float(self, schema: "Schema", value: float) -> None:
    """Write a float (32-bit) value to the output.

    :param schema: The shape's schema.
    :param value: The float value to write.
    """
    ...

write_integer(schema, value)

Write an integer (32-bit) value to the output.

:param schema: The shape's schema. :param value: The integer value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_integer(self, schema: "Schema", value: int) -> None:
    """Write an integer (32-bit) value to the output.

    :param schema: The shape's schema.
    :param value: The integer value to write.
    """
    ...

write_long(schema, value)

Write a long (64-bit integer) value to the output.

:param schema: The shape's schema. :param value: The long value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_long(self, schema: "Schema", value: int) -> None:
    """Write a long (64-bit integer) value to the output.

    :param schema: The shape's schema.
    :param value: The long value to write.
    """
    self.write_integer(schema, value)

write_null(schema)

Write a null value to the output.

:param schema: The shape's schema.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_null(self, schema: "Schema") -> None:
    """Write a null value to the output.

    :param schema: The shape's schema.
    """
    ...

write_short(schema, value)

Write a short (16-bit integer) value to the output.

:param schema: The shape's schema. :param value: The short value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_short(self, schema: "Schema", value: int) -> None:
    """Write a short (16-bit integer) value to the output.

    :param schema: The shape's schema.
    :param value: The short value to write.
    """
    self.write_integer(schema, value)

write_string(schema, value)

Write a string value to the output.

:param schema: The shape's schema. :param value: The string value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_string(self, schema: "Schema", value: str) -> None:
    """Write a string value to the output.

    :param schema: The shape's schema.
    :param value: The string value to write.
    """
    ...

write_struct(schema, struct)

Write a structured shape to the output.

This method is primarily intended to be used to serialize members that target structure or union shapes.

:param schema: The member schema of the structure. :param struct: The structure to serialize.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_struct(self, schema: "Schema", struct: "SerializeableStruct") -> None:
    """Write a structured shape to the output.

    This method is primarily intended to be used to serialize members that target
    structure or union shapes.

    :param schema: The member schema of the structure.
    :param struct: The structure to serialize.
    """
    with self.begin_struct(schema=schema) as struct_serializer:
        struct.serialize_members(struct_serializer)

write_timestamp(schema, value)

Write a timestamp value to the output.

:param schema: The shape's schema. :param value: The timestamp value to write.

Source code in packages/smithy-core/src/smithy_core/serializers.py

def write_timestamp(self, schema: "Schema", value: datetime.datetime) -> None:
    """Write a timestamp value to the output.

    :param schema: The shape's schema.
    :param value: The timestamp value to write.
    """
    ...

SpecificShapeSerializer

Bases: ShapeSerializer

Expects to serialize a specific kind of shape, failing if other shapes are serialized.

Source code in packages/smithy-core/src/smithy_core/serializers.py

class SpecificShapeSerializer(ShapeSerializer):
    """Expects to serialize a specific kind of shape, failing if other shapes are
    serialized."""

    def _invalid_state(
        self, schema: "Schema | None" = None, message: str | None = None
    ) -> Never:
        if message is None:
            message = f"Unexpected schema type: {schema}"
        raise SmithyError(message)

    def begin_struct(
        self, schema: "Schema"
    ) -> AbstractContextManager["ShapeSerializer"]:
        self._invalid_state(schema)

    def begin_list(
        self, schema: "Schema", size: int
    ) -> AbstractContextManager["ShapeSerializer"]:
        self._invalid_state(schema)

    def begin_map(
        self, schema: "Schema", size: int
    ) -> AbstractContextManager["MapSerializer"]:
        self._invalid_state(schema)

    def write_null(self, schema: "Schema") -> None:
        self._invalid_state(schema)

    def write_boolean(self, schema: "Schema", value: bool) -> None:
        self._invalid_state(schema)

    def write_byte(self, schema: "Schema", value: int) -> None:
        self._invalid_state(schema)

    def write_short(self, schema: "Schema", value: int) -> None:
        self._invalid_state(schema)

    def write_integer(self, schema: "Schema", value: int) -> None:
        self._invalid_state(schema)

    def write_long(self, schema: "Schema", value: int) -> None:
        self._invalid_state(schema)

    def write_float(self, schema: "Schema", value: float) -> None:
        self._invalid_state(schema)

    def write_double(self, schema: "Schema", value: float) -> None:
        self._invalid_state(schema)

    def write_big_integer(self, schema: "Schema", value: int) -> None:
        self._invalid_state(schema)

    def write_big_decimal(self, schema: "Schema", value: Decimal) -> None:
        self._invalid_state(schema)

    def write_string(self, schema: "Schema", value: str) -> None:
        self._invalid_state(schema)

    def write_blob(self, schema: "Schema", value: bytes) -> None:
        self._invalid_state(schema)

    def write_timestamp(self, schema: "Schema", value: datetime.datetime) -> None:
        self._invalid_state(schema)

    def write_document(self, schema: "Schema", value: "Document") -> None:
        self._invalid_state(schema)

    def write_data_stream(self, schema: "Schema", value: "_Stream") -> None:
        self._invalid_state(schema)

shapes

ShapeID

An identifier for a Smithy shape.

Source code in packages/smithy-core/src/smithy_core/shapes.py

class ShapeID:
    """An identifier for a Smithy shape."""

    _id: str
    _namespace: str
    _name: str
    _member: str | None = None

    def __init__(self, id: str) -> None:
        """Initialize a ShapeID.

        :param id: The string representation of the ID.
        """
        self._id = id
        if "#" not in id:
            raise SmithyError(f"Invalid shape id: {id}")
        self._namespace, self._name = id.split("#", 1)
        if not self.namespace or not self._name:
            raise SmithyError(f"Invalid shape id: {id}")

        if len(split_name := self._name.split("$", 1)) > 1:
            self._name, self._member = split_name
            if not self._name or not self._member:
                raise SmithyError(f"Invalid shape id: {id}")

    @property
    def namespace(self) -> str:
        """The namespace of the shape."""
        return self._namespace

    @property
    def name(self) -> str:
        """The name of the shape, or the name of the containing shape if the shape is a
        member."""
        return self._name

    @property
    def member(self) -> str | None:
        """The member name of the shape.

        This is only set for member shapes.
        """
        return self._member

    def expect_member(self) -> str:
        """Assert the member name is set and get it.

        :raises ExpectationNotMetException: If member wasn't set.
        :returns: Returns the member name.
        """
        if self.member is None:
            raise ExpectationNotMetError("Expected member to be set, but was None.")
        return self.member

    def with_member(self, member: str) -> "ShapeID":
        """Create a new shape id from the current id with the given member name.

        :param member: The member name to use on the new shape id.
        """
        return ShapeID.from_parts(
            namespace=self.namespace,
            name=self.name,
            member=member,
        )

    def __str__(self) -> str:
        return self._id

    def __repr__(self) -> str:
        return f"ShapeId({self._id})"

    def __eq__(self, other: object) -> bool:
        return self._id == str(other)

    def __hash__(self) -> int:
        return hash(self._id)

    @classmethod
    def from_parts(
        cls, *, namespace: str, name: str, member: str | None = None
    ) -> Self:
        """Initialize a ShapeID from component parts instead of a string whole.

        :param namesapce: The shape's namespace.
        :param name: The shape's individual name.
        :param member: The shape member's name. Only set for member shapes.
        """
        if member is not None:
            return cls(f"{namespace}#{name}${member}")
        return cls(f"{namespace}#{name}")

member property

The member name of the shape.

This is only set for member shapes.

name property

The name of the shape, or the name of the containing shape if the shape is a member.

namespace property

The namespace of the shape.

__init__(id)

Initialize a ShapeID.

:param id: The string representation of the ID.

Source code in packages/smithy-core/src/smithy_core/shapes.py

def __init__(self, id: str) -> None:
    """Initialize a ShapeID.

    :param id: The string representation of the ID.
    """
    self._id = id
    if "#" not in id:
        raise SmithyError(f"Invalid shape id: {id}")
    self._namespace, self._name = id.split("#", 1)
    if not self.namespace or not self._name:
        raise SmithyError(f"Invalid shape id: {id}")

    if len(split_name := self._name.split("$", 1)) > 1:
        self._name, self._member = split_name
        if not self._name or not self._member:
            raise SmithyError(f"Invalid shape id: {id}")

expect_member()

Assert the member name is set and get it.

:raises ExpectationNotMetException: If member wasn't set. :returns: Returns the member name.

Source code in packages/smithy-core/src/smithy_core/shapes.py

def expect_member(self) -> str:
    """Assert the member name is set and get it.

    :raises ExpectationNotMetException: If member wasn't set.
    :returns: Returns the member name.
    """
    if self.member is None:
        raise ExpectationNotMetError("Expected member to be set, but was None.")
    return self.member

from_parts(*, namespace, name, member=None) classmethod

Initialize a ShapeID from component parts instead of a string whole.

:param namesapce: The shape's namespace. :param name: The shape's individual name. :param member: The shape member's name. Only set for member shapes.

Source code in packages/smithy-core/src/smithy_core/shapes.py

@classmethod
def from_parts(
    cls, *, namespace: str, name: str, member: str | None = None
) -> Self:
    """Initialize a ShapeID from component parts instead of a string whole.

    :param namesapce: The shape's namespace.
    :param name: The shape's individual name.
    :param member: The shape member's name. Only set for member shapes.
    """
    if member is not None:
        return cls(f"{namespace}#{name}${member}")
    return cls(f"{namespace}#{name}")

with_member(member)

Create a new shape id from the current id with the given member name.

:param member: The member name to use on the new shape id.

Source code in packages/smithy-core/src/smithy_core/shapes.py

def with_member(self, member: str) -> "ShapeID":
    """Create a new shape id from the current id with the given member name.

    :param member: The member name to use on the new shape id.
    """
    return ShapeID.from_parts(
        namespace=self.namespace,
        name=self.name,
        member=member,
    )

ShapeType

Bases: Enum

The type of data that a shape represents.

Source code in packages/smithy-core/src/smithy_core/shapes.py

class ShapeType(Enum):
    """The type of data that a shape represents."""

    BLOB = 1
    BOOLEAN = 2
    STRING = 3
    TIMESTAMP = 4
    BYTE = 5
    SHORT = 6
    INTEGER = 7
    LONG = 8
    FLOAT = 9
    DOUBLE = 10
    BIG_INTEGER = 11
    BIG_DECIMAL = 12
    DOCUMENT = 13
    ENUM = 14
    INT_ENUM = 15

    LIST = 16
    MAP = 17
    STRUCTURE = 18
    UNION = 19

    # We won't acutally be using these, probably
    # MEMBER = 20
    SERVICE = 21
    # RESOURCE = 22
    OPERATION = 23

traits

APIKeyLocation

Bases: Enum

The locations that the api key could be placed in the signed request.

Source code in packages/smithy-core/src/smithy_core/traits.py

class APIKeyLocation(Enum):
    """The locations that the api key could be placed in the signed request."""

    HEADER = "header"
    QUERY = "query"

DynamicTrait dataclass

A component that can be attached to a schema to describe additional information about it.

Typed traits can be used by creating a :py:class:Trait subclass.

Source code in packages/smithy-core/src/smithy_core/traits.py

@dataclass(kw_only=True, frozen=True, slots=True)
class DynamicTrait:
    """A component that can be attached to a schema to describe additional information
    about it.

    Typed traits can be used by creating a :py:class:`Trait` subclass.
    """

    id: ShapeID
    """The ID of the trait."""

    document_value: "DocumentValue" = None
    """The value of the trait."""

document_value = None class-attribute instance-attribute

The value of the trait.

id instance-attribute

The ID of the trait.

Trait dataclass

A component that can be attached to a schema to describe additional information about it.

This is a base class that registers subclasses. Any known subclasses will automatically be used when constructing schemas. Any unknown traits may instead be created as a :py:class:DynamicTrait.

The id property of subclasses is set during subclass creation by __init_subclass__, so it is not necessary for subclasses to set it manually.

Source code in packages/smithy-core/src/smithy_core/traits.py

@dataclass(init=False, frozen=True)
class Trait:
    """A component that can be attached to a schema to describe additional information
    about it.

    This is a base class that registers subclasses. Any known subclasses will
    automatically be used when constructing schemas. Any unknown traits may instead be
    created as a :py:class:`DynamicTrait`.

    The `id` property of subclasses is set during subclass creation by
    `__init_subclass__`, so it is not necessary for subclasses to set it manually.
    """

    _REGISTRY: ClassVar[dict[ShapeID, type["Trait"]]] = {}

    id: ClassVar[ShapeID]
    """The ID of the trait."""

    document_value: "DocumentValue" = None
    """The value of the trait as a DocumentValue."""

    def __init_subclass__(cls, id: ShapeID) -> None:
        cls.id = id
        Trait._REGISTRY[id] = cls

    def __init__(self, value: "DocumentValue | DynamicTrait" = None):
        if type(self) is Trait:
            raise TypeError(
                "Only subclasses of Trait may be directly instantiated. "
                "Use DynamicTrait for traits without a concrete class."
            )

        if isinstance(value, DynamicTrait):
            if value.id != self.id:
                raise ValueError(
                    f"Attempted to instantiate an instance of {type(self)} from an "
                    f"invalid ID. Expected {self.id} but found {value.id}."
                )
            # Note that setattr is needed because it's a frozen (read-only) dataclass
            object.__setattr__(self, "document_value", value.document_value)
        else:
            object.__setattr__(self, "document_value", value)

    # Dynamically creates a subclass instance based on the trait id
    @staticmethod
    def new(id: ShapeID, value: "DocumentValue" = None) -> "Trait | DynamicTrait":
        """Dynamically create a new trait of the given ID.

        If the ID corresponds to a known Trait class, that class will be instantiated
        and returned. Otherwise, a :py:class:`DynamicTrait` will be returned.

        :returns: A trait of the given ID with the given value.
        """
        if (cls := Trait._REGISTRY.get(id, None)) is not None:
            return cls(value)
        return DynamicTrait(id=id, document_value=value)

document_value = None class-attribute instance-attribute

The value of the trait as a DocumentValue.

id class-attribute

The ID of the trait.

new(id, value=None) staticmethod

Dynamically create a new trait of the given ID.

If the ID corresponds to a known Trait class, that class will be instantiated and returned. Otherwise, a :py:class:DynamicTrait will be returned.

:returns: A trait of the given ID with the given value.

Source code in packages/smithy-core/src/smithy_core/traits.py

@staticmethod
def new(id: ShapeID, value: "DocumentValue" = None) -> "Trait | DynamicTrait":
    """Dynamically create a new trait of the given ID.

    If the ID corresponds to a known Trait class, that class will be instantiated
    and returned. Otherwise, a :py:class:`DynamicTrait` will be returned.

    :returns: A trait of the given ID with the given value.
    """
    if (cls := Trait._REGISTRY.get(id, None)) is not None:
        return cls(value)
    return DynamicTrait(id=id, document_value=value)

types

JsonBlob

Bases: bytes

Bytes that contain json data which can be lazily loaded.

Source code in packages/smithy-core/src/smithy_core/types.py

class JsonBlob(bytes):
    """Bytes that contain json data which can be lazily loaded."""

    _json = None

    def as_json(self) -> Any:
        if not self._json:
            self._json = json.loads(self.decode(encoding="utf-8"))
        return self._json

    @staticmethod
    def from_json(j: Any) -> "JsonBlob":
        json_string = JsonBlob(json.dumps(j).encode(encoding="utf-8"))
        json_string._json = j
        return json_string

JsonString

Bases: str

A string that contains json data which can be lazily loaded.

Source code in packages/smithy-core/src/smithy_core/types.py

class JsonString(str):
    """A string that contains json data which can be lazily loaded."""

    _json = None

    def as_json(self) -> Any:
        if not self._json:
            self._json = json.loads(self)
        return self._json

    @staticmethod
    def from_json(j: Any) -> "JsonString":
        json_string = JsonString(json.dumps(j))
        json_string._json = j
        return json_string

PathPattern dataclass

A formattable URI path pattern.

The pattern may contain formattable labels, which may be normal labels or greedy labels. Normal labels forbid path separators, greedy labels allow them.

Source code in packages/smithy-core/src/smithy_core/types.py

@dataclass(init=False, frozen=True)
class PathPattern:
    """A formattable URI path pattern.

    The pattern may contain formattable labels, which may be normal labels or greedy
    labels. Normal labels forbid path separators, greedy labels allow them.
    """

    pattern: str
    """The path component of the URI which is a formattable string."""

    greedy_labels: set[str]
    """The pattern labels whose values may contain path separators."""

    def __init__(self, pattern: str) -> None:
        object.__setattr__(self, "pattern", pattern)
        object.__setattr__(
            self, "greedy_labels", set(_GREEDY_LABEL_RE.findall(pattern))
        )

    def format(self, *args: object, **kwargs: str) -> str:
        if args:
            raise ValueError("PathPattern formatting requires only keyword arguments.")

        for key, value in kwargs.items():
            if "/" in value and key not in self.greedy_labels:
                raise ValueError(
                    'Non-greedy labels must not contain path separators ("/").'
                )

        result = self.pattern.replace("+}", "}").format(**kwargs)
        if "//" in result:
            raise ValueError(
                f'Path must not contain empty segments, but was "{result}".'
            )
        return result

greedy_labels instance-attribute

The pattern labels whose values may contain path separators.

pattern instance-attribute

The path component of the URI which is a formattable string.

PropertyKey dataclass

Bases: PropertyKey[T]

A typed property key.

.. code-block:: python

UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
    key="union",
    value_type=str | int,
)

Source code in packages/smithy-core/src/smithy_core/types.py

@dataclass(kw_only=True, frozen=True, slots=True, init=False)
class PropertyKey[T](_PropertyKey[T]):
    """A typed property key.

    .. code-block:: python

        UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
            key="union",
            value_type=str | int,
        )
    """

    key: str
    """The string key used to access the value."""

    value_type: "TypeForm[T]"
    """The type of the associated value in the property bag."""

    def __init__(self, *, key: str, value_type: "TypeForm[T]") -> None:
        # Intern the key to speed up dict access
        object.__setattr__(self, "key", sys.intern(key))
        object.__setattr__(self, "value_type", value_type)

key instance-attribute

The string key used to access the value.

value_type instance-attribute

The type of the associated value in the property bag.

TimestampFormat

Bases: Enum

Smithy-defined timestamp formats with serialization and deserialization helpers.

See Smithy's docs <https://smithy.io/2.0/spec/protocol-traits.html#smithy-api-timestampformat-trait>_ for more details.

Source code in packages/smithy-core/src/smithy_core/types.py

class TimestampFormat(Enum):
    """Smithy-defined timestamp formats with serialization and deserialization helpers.

    See `Smithy's docs <https://smithy.io/2.0/spec/protocol-traits.html#smithy-api-timestampformat-trait>`_
    for more details.
    """

    DATE_TIME = "date-time"
    """RFC3339 section 5.6 datetime with optional millisecond precision but no UTC
    offset."""

    HTTP_DATE = "http-date"
    """An HTTP date as defined by the IMF-fixdate production in RFC 9110 section
    5.6.7."""

    EPOCH_SECONDS = "epoch-seconds"
    """Also known as Unix time, the number of seconds that have elapsed since 00:00:00
    Coordinated Universal Time (UTC), Thursday, 1 January 1970, with optional
    millisecond precision."""

    def serialize(self, value: datetime) -> str | float:
        """Serializes a datetime into the timestamp format.

        :param value: The timestamp to serialize.
        :returns: A formatted timestamp. This will be a float for EPOCH_SECONDS, or a
            string otherwise.
        """
        value = ensure_utc(value)
        match self:
            case TimestampFormat.EPOCH_SECONDS:
                return serialize_epoch_seconds(value)
            case TimestampFormat.HTTP_DATE:
                return format_datetime(value, usegmt=True)
            case TimestampFormat.DATE_TIME:
                return serialize_rfc3339(value)

    def deserialize(self, value: str | float) -> datetime:
        """Deserializes a datetime from a value of the format.

        :param value: The timestamp value to deserialize. If the format is
            EPOCH_SECONDS, the value must be an int or float, or a string containing an
            int or float. Otherwise, it must be a string.
        :returns: The provided value as a datetime instance.
        """
        match self:
            case TimestampFormat.EPOCH_SECONDS:
                if isinstance(value, str):
                    try:
                        value = float(value)
                    except ValueError as e:
                        raise ExpectationNotMetError from e
                return epoch_seconds_to_datetime(value=value)
            case TimestampFormat.HTTP_DATE:
                return ensure_utc(parsedate_to_datetime(expect_type(str, value)))
            case TimestampFormat.DATE_TIME:
                return ensure_utc(datetime.fromisoformat(expect_type(str, value)))

DATE_TIME = 'date-time' class-attribute instance-attribute

RFC3339 section 5.6 datetime with optional millisecond precision but no UTC offset.

EPOCH_SECONDS = 'epoch-seconds' class-attribute instance-attribute

Also known as Unix time, the number of seconds that have elapsed since 00:00:00 Coordinated Universal Time (UTC), Thursday, 1 January 1970, with optional millisecond precision.

HTTP_DATE = 'http-date' class-attribute instance-attribute

An HTTP date as defined by the IMF-fixdate production in RFC 9110 section 5.6.7.

deserialize(value)

Deserializes a datetime from a value of the format.

:param value: The timestamp value to deserialize. If the format is EPOCH_SECONDS, the value must be an int or float, or a string containing an int or float. Otherwise, it must be a string. :returns: The provided value as a datetime instance.

Source code in packages/smithy-core/src/smithy_core/types.py

def deserialize(self, value: str | float) -> datetime:
    """Deserializes a datetime from a value of the format.

    :param value: The timestamp value to deserialize. If the format is
        EPOCH_SECONDS, the value must be an int or float, or a string containing an
        int or float. Otherwise, it must be a string.
    :returns: The provided value as a datetime instance.
    """
    match self:
        case TimestampFormat.EPOCH_SECONDS:
            if isinstance(value, str):
                try:
                    value = float(value)
                except ValueError as e:
                    raise ExpectationNotMetError from e
            return epoch_seconds_to_datetime(value=value)
        case TimestampFormat.HTTP_DATE:
            return ensure_utc(parsedate_to_datetime(expect_type(str, value)))
        case TimestampFormat.DATE_TIME:
            return ensure_utc(datetime.fromisoformat(expect_type(str, value)))

serialize(value)

Serializes a datetime into the timestamp format.

:param value: The timestamp to serialize. :returns: A formatted timestamp. This will be a float for EPOCH_SECONDS, or a string otherwise.

Source code in packages/smithy-core/src/smithy_core/types.py

def serialize(self, value: datetime) -> str | float:
    """Serializes a datetime into the timestamp format.

    :param value: The timestamp to serialize.
    :returns: A formatted timestamp. This will be a float for EPOCH_SECONDS, or a
        string otherwise.
    """
    value = ensure_utc(value)
    match self:
        case TimestampFormat.EPOCH_SECONDS:
            return serialize_epoch_seconds(value)
        case TimestampFormat.HTTP_DATE:
            return format_datetime(value, usegmt=True)
        case TimestampFormat.DATE_TIME:
            return serialize_rfc3339(value)

TypedProperties

Bases: UserDict[str, Any], TypedProperties

A map with typed setters and getters.

Keys can be either a string or a :py:class:smithy_core.interfaces.PropertyKey. Using a PropertyKey instead of a string enables type checkers to narrow to the associated value type rather than having to use Any.

Letting the value be either a string or PropertyKey allows consumers who care about typing to get it, and those who don't care about typing to not have to think about it.

No runtime type assertion is performed.

..code-block:: python

foo = PropertyKey(key="foo", value_type=str)
properties = TypedProperties()
properties[foo] = "bar"

assert assert_type(properties[foo], str) == "bar"
assert assert_type(properties["foo"], Any) == "bar"

Note that unions and other special types cannot easily be used here due to being incompatible with type[T]. PEP747 proposes a fix to this case, but it has not yet been accepted. In the meantime, there is a workaround. The PropertyKey must be assigned to an explicitly typed variable, and the value_type parameter of the constructor must have a # type: ignore comment, like so:

.. code-block:: python

UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
    key="union",
    value_type=str | int,  # type: ignore
)

properties = TypedProperties()
properties[UNION_PROPERTY] = "foo"

assert assert_type(properties[UNION_PROPERTY], str | int) == "foo"

Type checkers will be able to use such a property as expected.

Source code in packages/smithy-core/src/smithy_core/types.py

class TypedProperties(UserDict[str, Any], _TypedProperties):
    """A map with typed setters and getters.

    Keys can be either a string or a :py:class:`smithy_core.interfaces.PropertyKey`.
    Using a PropertyKey instead of a string enables type checkers to narrow to the
    associated value type rather than having to use Any.

    Letting the value be either a string or PropertyKey allows consumers who care about
    typing to get it, and those who don't care about typing to not have to think about
    it.

    No runtime type assertion is performed.

    ..code-block:: python

        foo = PropertyKey(key="foo", value_type=str)
        properties = TypedProperties()
        properties[foo] = "bar"

        assert assert_type(properties[foo], str) == "bar"
        assert assert_type(properties["foo"], Any) == "bar"

    Note that unions and other special types cannot easily be used here due to being
    incompatible with ``type[T]``. PEP747 proposes a fix to this case, but it has not
    yet been accepted. In the meantime, there is a workaround. The PropertyKey must
    be assigned to an explicitly typed variable, and the ``value_type`` parameter of
    the constructor must have a ``# type: ignore`` comment, like so:

    .. code-block:: python

        UNION_PROPERTY: PropertyKey[str | int] = PropertyKey(
            key="union",
            value_type=str | int,  # type: ignore
        )

        properties = TypedProperties()
        properties[UNION_PROPERTY] = "foo"

        assert assert_type(properties[UNION_PROPERTY], str | int) == "foo"

    Type checkers will be able to use such a property as expected.
    """

    @overload
    def __getitem__[T](self, key: _PropertyKey[T]) -> T: ...
    @overload
    def __getitem__(self, key: str) -> Any: ...
    def __getitem__(self, key: str | _PropertyKey[Any]) -> Any:
        return self.data[key if isinstance(key, str) else key.key]

    @overload
    def __setitem__[T](self, key: _PropertyKey[T], value: T) -> None: ...
    @overload
    def __setitem__(self, key: str, value: Any) -> None: ...
    def __setitem__(self, key: str | _PropertyKey[Any], value: Any) -> None:
        self.data[key if isinstance(key, str) else key.key] = value

    def __delitem__(self, key: str | _PropertyKey[Any]) -> None:
        del self.data[key if isinstance(key, str) else key.key]

    def __contains__(self, key: object) -> bool:
        return super().__contains__(key.key if isinstance(key, _PropertyKey) else key)

    @overload
    def get[T](self, key: _PropertyKey[T], default: None = None) -> T | None: ...
    @overload
    def get[T](self, key: _PropertyKey[T], default: T) -> T: ...
    @overload
    def get[T, DT](self, key: _PropertyKey[T], default: DT) -> T | DT: ...
    @overload
    def get(self, key: str, default: None = None) -> Any | None: ...
    @overload
    def get[T](self, key: str, default: T) -> Any | T: ...

    # pyright has trouble detecting compatible overrides when both the superclass
    # and subclass have overloads.
    def get(self, key: str | _PropertyKey[Any], default: Any = None) -> Any:  # type: ignore
        return self.data.get(key if isinstance(key, str) else key.key, default)

    @overload
    def pop[T](self, key: _PropertyKey[T], default: None = None) -> T | None: ...
    @overload
    def pop[T](self, key: _PropertyKey[T], default: T) -> T: ...
    @overload
    def pop[T, DT](self, key: _PropertyKey[T], default: DT) -> T | DT: ...
    @overload
    def pop(self, key: str, default: None = None) -> Any | None: ...
    @overload
    def pop[T](self, key: str, default: T) -> Any | T: ...

    # pyright has trouble detecting compatible overrides when both the superclass
    # and subclass have overloads.
    def pop(self, key: str | _PropertyKey[Any], default: Any = None) -> Any:  # type: ignore
        return self.data.pop(key if isinstance(key, str) else key.key, default)

utils

ensure_utc(value)

Ensures that the given datetime is a UTC timezone-aware datetime.

If the datetime isn't timezone-aware, its timezone is set to UTC. If it is aware, it's replaced with the equivalent datetime under UTC.

:param value: A datetime object that may or may not be timezone-aware. :returns: A UTC timezone-aware equivalent datetime.

Source code in packages/smithy-core/src/smithy_core/utils.py

def ensure_utc(value: datetime) -> datetime:
    """Ensures that the given datetime is a UTC timezone-aware datetime.

    If the datetime isn't timezone-aware, its timezone is set to UTC. If it is aware,
    it's replaced with the equivalent datetime under UTC.

    :param value: A datetime object that may or may not be timezone-aware.
    :returns: A UTC timezone-aware equivalent datetime.
    """
    if value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    else:
        return value.astimezone(UTC)

epoch_seconds_to_datetime(value)

Parse numerical epoch timestamps (seconds since 1970) into a datetime in UTC.

Falls back to using timedelta when fromtimestamp raises OverflowError. From Python's fromtimestamp documentation: "This may raise OverflowError, if the timestamp is out of the range of values supported by the platform C localtime() function, and OSError on localtime() failure. It's common for this to be restricted to years from 1970 through 2038." This affects 32-bit systems.

Source code in packages/smithy-core/src/smithy_core/utils.py

def epoch_seconds_to_datetime(value: int | float) -> datetime:
    """Parse numerical epoch timestamps (seconds since 1970) into a datetime in UTC.

    Falls back to using ``timedelta`` when ``fromtimestamp`` raises ``OverflowError``.
    From Python's ``fromtimestamp`` documentation: "This may raise OverflowError, if the
    timestamp is out of the range of values supported by the platform C localtime()
    function, and OSError on localtime() failure. It's common for this to be restricted
    to years from 1970 through 2038." This affects 32-bit systems.
    """
    try:
        return datetime.fromtimestamp(value, tz=UTC)
    except OverflowError:
        epoch_zero = datetime(1970, 1, 1, 0, 0, 0, tzinfo=UTC)
        return epoch_zero + timedelta(seconds=value)

expect_type(typ, value)

expect_type(typ: type[_T], value: Any) -> _T

expect_type(typ: UnionType, value: Any) -> Any

Asserts a value is of the given type and returns it unchanged.

This performs both a runtime assertion and type narrowing during type checking similar to typing.cast. If the runtime assertion is not needed, typing.cast should be preferred.

:param typ: The expected type. :param value: The value which is expected to be the given type. :returns: The given value cast as the given type. :raises SmithyError: If the value does not match the type.

Source code in packages/smithy-core/src/smithy_core/utils.py

def expect_type(typ: UnionType | type, value: Any) -> Any:
    """Asserts a value is of the given type and returns it unchanged.

    This performs both a runtime assertion and type narrowing during type checking
    similar to ``typing.cast``. If the runtime assertion is not needed, ``typing.cast``
    should be preferred.

    :param typ: The expected type.
    :param value: The value which is expected to be the given type.
    :returns: The given value cast as the given type.
    :raises SmithyError: If the value does not match the type.
    """
    if not isinstance(value, typ):
        raise ExpectationNotMetError(f"Expected {typ}, found {type(value)}: {value}")
    return value

limited_parse_float(value)

Asserts a value is a float or a limited set of non-numerical strings and returns it as a float.

:param value: An object that is expected to be a float. :returns: The given value as a float. :raises SmithyError: If the value is not a float or one of the strings NaN, Infinity, or -Infinity.

Source code in packages/smithy-core/src/smithy_core/utils.py

def limited_parse_float(value: Any) -> float:
    """Asserts a value is a float or a limited set of non-numerical strings and returns
    it as a float.

    :param value: An object that is expected to be a float.
    :returns: The given value as a float.
    :raises SmithyError: If the value is not a float or one of the strings ``NaN``,
        ``Infinity``, or ``-Infinity``.
    """
    # TODO: add limited bounds checking
    if isinstance(value, str) and value in _NON_NUMERICAL_FLOATS:
        return float(value)

    return expect_type(float, value)

limited_serialize_float(given)

Serializes non-numeric floats to strings.

Numeric floats are returned without alteration.

:param given: A float to be conditionally serialized. :returns: The given float as a float or string.

Source code in packages/smithy-core/src/smithy_core/utils.py

def limited_serialize_float(given: float) -> str | float:
    """Serializes non-numeric floats to strings.

    Numeric floats are returned without alteration.

    :param given: A float to be conditionally serialized.
    :returns: The given float as a float or string.
    """
    if isnan(given):
        return "NaN"
    if isinf(given):
        return "-Infinity" if given < 0 else "Infinity"

    return given

remove_dot_segments(path, remove_consecutive_slashes=False)

Removes dot segments from a path per :rfc:3986#section-5.2.4.

Optionally removes consecutive slashes.

:param path: The path to modify. :param remove_consecutive_slashes: Whether to remove consecutive slashes. :returns: The path with dot segments removed.

Source code in packages/smithy-core/src/smithy_core/utils.py

def remove_dot_segments(path: str, remove_consecutive_slashes: bool = False) -> str:
    """Removes dot segments from a path per :rfc:`3986#section-5.2.4`.

    Optionally removes consecutive slashes.

    :param path: The path to modify.
    :param remove_consecutive_slashes: Whether to remove consecutive slashes.
    :returns: The path with dot segments removed.
    """
    output: list[str] = []
    for segment in path.split("/"):
        if segment == ".":
            continue
        elif segment != "..":
            output.append(segment)
        elif output:
            output.pop()
    if path.startswith("/") and (not output or output[0]):
        output.insert(0, "")
    if output and path.endswith(("/.", "/..")):
        output.append("")
    result = "/".join(output)
    if remove_consecutive_slashes:
        result = result.replace("//", "/")
    return result

serialize_epoch_seconds(given)

Serializes a datetime into a string containing the epoch seconds.

If microseconds is 0, no fractional part is serialized.

:param given: The datetime to serialize. :returns: A string containing the seconds since the UNIX epoch.

Source code in packages/smithy-core/src/smithy_core/utils.py

def serialize_epoch_seconds(given: datetime) -> float:
    """Serializes a datetime into a string containing the epoch seconds.

    If ``microseconds`` is 0, no fractional part is serialized.

    :param given: The datetime to serialize.
    :returns: A string containing the seconds since the UNIX epoch.
    """
    result = given.timestamp()
    if given.microsecond == 0:
        result = int(result)
    return result

serialize_float(given)

Serializes a float to a string.

This ensures non-numeric floats are serialized correctly, and ensures that there is a fractional part.

:param given: A float or Decimal to be serialized. :returns: The string representation of the given float.

Source code in packages/smithy-core/src/smithy_core/utils.py

def serialize_float(given: float | Decimal) -> str:
    """Serializes a float to a string.

    This ensures non-numeric floats are serialized correctly, and ensures that there is
    a fractional part.

    :param given: A float or Decimal to be serialized.
    :returns: The string representation of the given float.
    """
    if isnan(given):
        return "NaN"
    if isinf(given):
        return "-Infinity" if given < 0 else "Infinity"

    if isinstance(given, Decimal):
        given = given.normalize()

    result = str(given)
    if result.isnumeric():
        result += ".0"
    return result

serialize_rfc3339(given)

Serializes a datetime into an RFC3339 string representation.

If microseconds is 0, no fractional part is serialized.

:param given: The datetime to serialize. :returns: An RFC3339 formatted timestamp.

Source code in packages/smithy-core/src/smithy_core/utils.py

def serialize_rfc3339(given: datetime) -> str:
    """Serializes a datetime into an RFC3339 string representation.

    If ``microseconds`` is 0, no fractional part is serialized.

    :param given: The datetime to serialize.
    :returns: An RFC3339 formatted timestamp.
    """
    if given.microsecond != 0:
        return given.strftime(RFC3339_MICRO)
    else:
        return given.strftime(RFC3339)

split_every(given, split_char, n)

Splits a string every nth instance of the given character.

:param given: The string to split. :param split_char: The character to split on. :param n: The number of instances of split_char to see before each split. :returns: A list of strings.

Source code in packages/smithy-core/src/smithy_core/utils.py

def split_every(given: str, split_char: str, n: int) -> list[str]:
    """Splits a string every nth instance of the given character.

    :param given: The string to split.
    :param split_char: The character to split on.
    :param n: The number of instances of split_char to see before each split.
    :returns: A list of strings.
    """
    split = given.split(split_char)
    return [split_char.join(split[i : i + n]) for i in range(0, len(split), n)]

strict_parse_bool(given)

Strictly parses a boolean from string.

:param given: A string that is expected to contain either "true" or "false". :returns: The given string parsed to a boolean. :raises ExpectationNotMetError: if the given string is neither "true" nor "false".

Source code in packages/smithy-core/src/smithy_core/utils.py

def strict_parse_bool(given: str) -> bool:
    """Strictly parses a boolean from string.

    :param given: A string that is expected to contain either "true" or "false".
    :returns: The given string parsed to a boolean.
    :raises ExpectationNotMetError: if the given string is neither "true" nor "false".
    """
    match given:
        case "true":
            return True
        case "false":
            return False
        case _:
            raise ExpectationNotMetError(f"Expected 'true' or 'false', found: {given}")

strict_parse_float(given)

Strictly parses a float from a string.

Unlike float(), this forbids the use of "inf" and case-sensitively matches Infinity and NaN.

:param given: A string that is expected to contain a float. :returns: The given string parsed to a float. :raises ExpectationNotMetError: If the given string isn't a float.

Source code in packages/smithy-core/src/smithy_core/utils.py

def strict_parse_float(given: str) -> float:
    """Strictly parses a float from a string.

    Unlike float(), this forbids the use of "inf" and case-sensitively matches Infinity
    and NaN.

    :param given: A string that is expected to contain a float.
    :returns: The given string parsed to a float.
    :raises ExpectationNotMetError: If the given string isn't a float.
    """
    if _FLOAT_REGEX.fullmatch(given):
        return float(given)
    raise ExpectationNotMetError(f"Expected float, found: {given}")