Server Test Python

import pathlib

import typing

class TracingHandler:

    """

    Modifies the Python `logging` module to deliver its log messages using [tracing::Subscriber] events.

    To achieve this goal, the following changes are made to the module:

    - A new builtin function `logging.py_tracing_event` transcodes `logging.LogRecord`s to `tracing::Event`s. This function

      is not exported in `logging.__all__`, as it is not intended to be called directly.

    - A new class `logging.TracingHandler` provides a `logging.Handler` that delivers all records to `python_tracing`.

    """

    def handler(self) -> typing.Any:

        ...

    def __init__(self, level: typing.Optional[int] = ..., logfile: typing.Optional[pathlib.Path] = ..., format: typing.Optional[typing.Literal['compact', 'pretty', 'json']] = ...) -> None:

        ...

def py_tracing_event() -> None:

    """

    Consumes a Python `logging.LogRecord` and emits a Rust [tracing::Event] instead.

    """

    ...

import typing

class MiddlewareException(Exception):

    """

    Exception that can be thrown from a Python middleware.

    It allows to specify a message and HTTP status code and implementing protocol specific capabilities

    to build a [aws_smithy_http_server::response::Response] from it.

    """

    message: str

    status_code: int

    def __init__(self, message: str, status_code: typing.Optional[int] = ...) -> None:

        ...

class Request:

    """

    Python-compatible [Request] object.

    """

    body: typing.Awaitable[bytes]

    """

    Return the HTTP body of this request.

    Note that this is a costly operation because the whole request body is cloned.

    """

    headers: typing.MutableMapping[str, str]

    """

    Return the HTTP headers of this request.

    """

    method: str

    """

    Return the HTTP method of this request.

    """

    uri: str

    """

    Return the URI of this request.

    """

    version: str

    """

    Return the HTTP version of this request.

    """

class Response:

    """

    Python-compatible [Response] object.

    """

    body: typing.Awaitable[bytes]

    """

    Return the HTTP body of this response.

    Note that this is a costly operation because the whole response body is cloned.

    """

    headers: typing.MutableMapping[str, str]

    """

    Return the HTTP headers of this response.

    """

    status: int

    """

    Return the HTTP status of this response.

    """

    version: str

    """

    Return the HTTP version of this response.

    """

    def __init__(self, status: int, headers: typing.Optional[typing.Dict[str, str]] = ..., body: typing.Optional[bytes] = ...) -> None:

        ...

import naming_test_structs.model

import typing

class Option:

    value: typing.Optional[str]

    def __init__(self, value: typing.Optional[str] = ...) -> None:

        ...

class Result:

    value: typing.Optional[str]

    def __init__(self, value: typing.Optional[str] = ...) -> None:

        ...

class SomeUnion:

    def as_option(self) -> naming_test_structs.model.Option:

        """

        Tries to convert the enum instance into [`Option`](crate::model::SomeUnion::Option), extracting the inner [`Option`](crate::model::Option).

        """

        ...

    def as_result(self) -> naming_test_structs.model.Result:

        """

        Tries to convert the enum instance into [`Result`](crate::model::SomeUnion::Result), extracting the inner [`Result`](crate::model::Result).

        """

        ...

    def is_option(self) -> bool:

        """

        Returns true if this is a [`Option`](crate::model::SomeUnion::Option).

        """

        ...

    def is_result(self) -> bool:

        """

        Returns true if this is a [`Result`](crate::model::SomeUnion::Result).

        """

        ...

    @staticmethod

    def option(data: naming_test_structs.model.Option) -> SomeUnion:

        """

        Creates a new union instance of [`Option`](crate::model::SomeUnion::Option)

        """

        ...

    @staticmethod

    def result(data: naming_test_structs.model.Result) -> SomeUnion:

        """

        Creates a new union instance of [`Result`](crate::model::SomeUnion::Result)

        """

        ...

class SomethingElse:

    option: typing.Optional[naming_test_structs.model.Option]

    option_list: typing.Optional[typing.List[naming_test_structs.model.Result]]

    result: typing.Optional[naming_test_structs.model.Result]

    result_list: typing.Optional[typing.List[naming_test_structs.model.Result]]

    some_union: typing.Optional[naming_test_structs.model.SomeUnion]

    def __init__(self, result: typing.Optional[naming_test_structs.model.Result] = ..., result_list: typing.Optional[typing.List[naming_test_structs.model.Result]] = ..., option: typing.Optional[naming_test_structs.model.Option] = ..., option_list: typing.Optional[typing.List[naming_test_structs.model.Result]] = ..., some_union: typing.Optional[naming_test_structs.model.SomeUnion] = ...) -> None:

        ...

import naming_test_structs.model

import typing

class StructsOutput:

    option: typing.Optional[naming_test_structs.model.Option]

    option_list: typing.Optional[typing.List[naming_test_structs.model.Result]]

    result: typing.Optional[naming_test_structs.model.Result]

    result_list: typing.Optional[typing.List[naming_test_structs.model.Result]]

    something_else: typing.Optional[naming_test_structs.model.SomethingElse]

    def __init__(self, result: typing.Optional[naming_test_structs.model.Result] = ..., result_list: typing.Optional[typing.List[naming_test_structs.model.Result]] = ..., option: typing.Optional[naming_test_structs.model.Option] = ..., option_list: typing.Optional[typing.List[naming_test_structs.model.Result]] = ..., something_else: typing.Optional[naming_test_structs.model.SomethingElse] = ...) -> None:

        ...

import typing

class PySocket:

    """

    Socket implementation that can be shared between multiple Python processes.

    Python cannot handle true multi-threaded applications due to the [GIL],

    often resulting in reduced performance and only one core used by the application.

    To work around this, Python web applications usually create a socket with

    SO_REUSEADDR and SO_REUSEPORT enabled that can be shared between multiple

    Python processes, allowing you to maximize performance and use all available

    computing capacity of the host.

    [GIL]: https://wiki.python.org/moin/GlobalInterpreterLock

    """

    def try_clone(self) -> PySocket:

        """

        Clone the inner socket allowing it to be shared between multiple

        Python processes.

        """

        ...

    def __init__(self, address: str, port: int, backlog: typing.Optional[int] = ...) -> None:

        ...

import pathlib

class TlsConfig:

    """

    PyTlsConfig represents TLS configuration created from Python.

    """

    def __init__(self, key_path: pathlib.Path, cert_path: pathlib.Path, reload_secs: int = ...) -> None:

        ...

import typing

class Blob:

    """

    Python Wrapper for [aws_smithy_types::Blob].

    """

    data: bytes

    """

    Python getter for the `Blob` byte array.

    """

    def __init__(self, input: bytes) -> None:

        ...

class ByteStream:

    """

    Python Wrapper for [aws_smithy_types::byte_stream::ByteStream].

    ByteStream provides misuse-resistant primitives to make it easier to handle common patterns with streaming data.

    On the Rust side, The Python implementation wraps the original [ByteStream](aws_smithy_types::byte_stream::ByteStream)

    in a clonable structure and implements the [Stream](futures::stream::Stream) trait for it to

    allow Rust to handle the type transparently.

    On the Python side both sync and async iterators are exposed by implementing `__iter__()` and `__aiter__()` magic methods,

    which allows to just loop over the stream chunks.

    ### Example of async streaming:

    ```python

        stream = await ByteStream.from_path("/tmp/music.mp3")

        async for chunk in stream:

            print(chunk)

    ```

    ### Example of sync streaming:

    ```python

        stream = ByteStream.from_stream_blocking("/tmp/music.mp3")

        for chunk in stream:

            print(chunk)

    ```

    The main difference between the two implementations is that the async one is scheduling the Python coroutines as Rust futures,

    effectively maintaining the asyncronous behavior that Rust exposes, while the sync one is blocking the Tokio runtime to be able

    to await one chunk at a time.

    The original Rust [ByteStream](aws_smithy_types::byte_stream::ByteStream) is wrapped inside a `Arc<Mutex>` to allow the type to be

    [Clone] (required by PyO3) and to allow internal mutability, required to fetch the next chunk of data.

    """

    @staticmethod

    def from_path(path: str) -> typing.Awaitable[ByteStream]:

        """

        Create a new [ByteStream](aws_smithy_types::byte_stream::ByteStream) from a path, forcing

        Python to await this coroutine.

        """

        ...

    @staticmethod

    def from_path_blocking(path: str) -> ByteStream:

        """

        Create a new [ByteStream](aws_smithy_types::byte_stream::ByteStream) from a path, without

        requiring Python to await this method.

        **NOTE:** This method will block the Rust event loop when it is running.

        """

        ...

    def __init__(self, input: bytes) -> None:

        ...

class DateTime:

    """

    Python Wrapper for [aws_smithy_types::date_time::DateTime].

    """

    def as_nanos(self) -> int:

        """

        Returns the number of nanoseconds since the Unix epoch that this `DateTime` represents.

        """

        ...

    def as_secs_f64(self) -> float:

        """

        Returns the `DateTime` value as an `f64` representing the seconds since the Unix epoch.

        """

        ...

    @staticmethod

    def from_fractional_secs(epoch_seconds: int, fraction: float) -> DateTime:

        """

        Creates a `DateTime` from a number of seconds and a fractional second since the Unix epoch.

        """

        ...

    @staticmethod

    def from_millis(epoch_millis: int) -> DateTime:

        """

        Creates a `DateTime` from a number of milliseconds since the Unix epoch.

        """

        ...

    @staticmethod

    def from_nanos(epoch_nanos: int) -> DateTime:

        """

        Creates a `DateTime` from a number of nanoseconds since the Unix epoch.

        """

        ...

    @staticmethod

    def from_secs(epoch_seconds: int) -> DateTime:

        """

        Creates a `DateTime` from a number of seconds since the Unix epoch.

        """

        ...

    @staticmethod

    def from_secs_and_nanos(seconds: int, subsecond_nanos: int) -> DateTime:

        """

        Creates a `DateTime` from a number of seconds and sub-second nanos since the Unix epoch.

        """

        ...

    @staticmethod

    def from_secs_f64(epoch_seconds: float) -> DateTime:

        """

        Creates a `DateTime` from an `f64` representing the number of seconds since the Unix epoch.

        """

        ...

    @staticmethod

    def from_str(s: str, format: Format) -> DateTime:

        """

        Parses a `DateTime` from a string using the given `format`.

        """

        ...

    def has_subsec_nanos(self) -> bool:

        """

        Returns true if sub-second nanos is greater than zero.

        """

        ...

    @staticmethod

    def read(format: Format, delim: str) -> typing.Tuple[DateTime, str]:

        """

        Read 1 date of `format` from `s`, expecting either `delim` or EOF.

        TODO(PythonTyping): How do we represent `char` in Python?

        """

        ...

    def secs(self) -> int:

        """

        Returns the epoch seconds component of the `DateTime`.

        """

        ...

    def subsec_nanos(self) -> int:

        """

        Returns the sub-second nanos component of the `DateTime`.

        """

        ...

    def to_millis(self) -> int:

        """

        Converts the `DateTime` to the number of milliseconds since the Unix epoch.

        """

        ...

class Format:

    """

    Formats for representing a `DateTime` in the Smithy protocols.

    """

    DateTime: Format

    """

    Formats for representing a `DateTime` in the Smithy protocols.

    """

    EpochSeconds: Format

    """

    Formats for representing a `DateTime` in the Smithy protocols.

    """

    HttpDate: Format

    """

    Formats for representing a `DateTime` in the Smithy protocols.

    """

        crate::model::SomeUnion::Result(inner) => {

        crate::model::SomeUnion::Option(inner) => {

import rest_json.input

import rest_json.middleware

import rest_json.output

import rest_json.tls

import typing

Ctx = typing.TypeVar('Ctx')

class App(typing.Generic[Ctx]):

    """

    Main Python application, used to register operations and context and start multiple

    workers on the same shared socket.

    Operations can be registered using the application object as a decorator (`@app.operation_name`).

    Here's a full example to get you started:

    ```python

    from rest_json import input

    from rest_json import output

    from rest_json import error

    from rest_json import middleware

    from rest_json import App

    @dataclass

    class Context:

        counter: int = 0

    app = App()

    app.context(Context())

    @app.request_middleware

    def request_middleware(request: middleware::Request):

        if request.get_header("x-amzn-id") != "secret":

            raise middleware.MiddlewareException("Unsupported `x-amz-id` header", 401)

    # This example uses all query string types.

    @app.all_query_string_types

    def all_query_string_types(input: input::AllQueryStringTypesInput, ctx: Context) -> output::AllQueryStringTypesOutput:

        raise NotImplementedError

    # This example uses fixed query string params and variable query string params.

    # The fixed query string parameters and variable parameters must both be

    # serialized (implementations may need to merge them together).

    @app.constant_and_variable_query_string

    def constant_and_variable_query_string(input: input::ConstantAndVariableQueryStringInput, ctx: Context) -> output::ConstantAndVariableQueryStringOutput:

        raise NotImplementedError

    # This example uses a constant query string parameters and a label.

    # This simply tests that labels and query string parameters are

    # compatible. The fixed query string parameter named "hello" should

    # in no way conflict with the label, `{hello}`.

    @app.constant_query_string

    def constant_query_string(input: input::ConstantQueryStringInput, ctx: Context) -> output::ConstantQueryStringOutput:

        raise NotImplementedError

    # The example tests how servers must support requests

    # containing a `Content-Type` header with parameters.

    @app.content_type_parameters

    def content_type_parameters(input: input::ContentTypeParametersInput, ctx: Context) -> output::ContentTypeParametersOutput:

        raise NotImplementedError

    @app.datetime_offsets

    def datetime_offsets(input: input::DatetimeOffsetsInput, ctx: Context) -> output::DatetimeOffsetsOutput:

        raise NotImplementedError

    # This example serializes a document as part of the payload.

    @app.document_type

    def document_type(input: input::DocumentTypeInput, ctx: Context) -> output::DocumentTypeOutput:

        raise NotImplementedError

    # This example serializes documents as the value of maps.

    @app.document_type_as_map_value

    def document_type_as_map_value(input: input::DocumentTypeAsMapValueInput, ctx: Context) -> output::DocumentTypeAsMapValueOutput:

        raise NotImplementedError

    # This example serializes a document as the entire HTTP payload.

    @app.document_type_as_payload

    def document_type_as_payload(input: input::DocumentTypeAsPayloadInput, ctx: Context) -> output::DocumentTypeAsPayloadOutput:

        raise NotImplementedError

    # The example tests how requests and responses are serialized when there's

    # no request or response payload because the operation has an empty input

    # and empty output structure that reuses the same shape. While this should

    # be rare, code generators must support this.

    @app.empty_input_and_empty_output

    def empty_input_and_empty_output(input: input::EmptyInputAndEmptyOutputInput, ctx: Context) -> output::EmptyInputAndEmptyOutputOutput:

        raise NotImplementedError

    @app.endpoint_operation

    def endpoint_operation(input: input::EndpointOperationInput, ctx: Context) -> output::EndpointOperationOutput:

        raise NotImplementedError

    @app.endpoint_with_host_label_operation

    def endpoint_with_host_label_operation(input: input::EndpointWithHostLabelOperationInput, ctx: Context) -> output::EndpointWithHostLabelOperationOutput:

        raise NotImplementedError

    @app.fractional_seconds

    def fractional_seconds(input: input::FractionalSecondsInput, ctx: Context) -> output::FractionalSecondsOutput:

        raise NotImplementedError

    # This operation has four possible return values:

    #

    # 1. A successful response in the form of GreetingWithErrorsOutput

    # 2. An InvalidGreeting error.

    # 3. A BadRequest error.

    # 4. A FooError.

    #

    # Implementations must be able to successfully take a response and

    # properly (de)serialize successful and error responses based on the

    # the presence of the

    @app.greeting_with_errors

    def greeting_with_errors(input: input::GreetingWithErrorsInput, ctx: Context) -> output::GreetingWithErrorsOutput:

        raise NotImplementedError

    @app.host_with_path_operation

    def host_with_path_operation(input: input::HostWithPathOperationInput, ctx: Context) -> output::HostWithPathOperationOutput:

        raise NotImplementedError

    # This example tests httpChecksumRequired trait

    @app.http_checksum_required

    def http_checksum_required(input: input::HttpChecksumRequiredInput, ctx: Context) -> output::HttpChecksumRequiredOutput:

        raise NotImplementedError

    # Clients that perform this test extract all headers from the response.

    @app.http_empty_prefix_headers

    def http_empty_prefix_headers(input: input::HttpEmptyPrefixHeadersInput, ctx: Context) -> output::HttpEmptyPrefixHeadersOutput:

        raise NotImplementedError

    @app.http_enum_payload

    def http_enum_payload(input: input::HttpEnumPayloadInput, ctx: Context) -> output::HttpEnumPayloadOutput:

        raise NotImplementedError

    # This example serializes a blob shape in the payload.

    #

    # In this example, no JSON document is synthesized because the payload is

    # not a structure or a union type.

    @app.http_payload_traits

    def http_payload_traits(input: input::HttpPayloadTraitsInput, ctx: Context) -> output::HttpPayloadTraitsOutput:

        raise NotImplementedError

    # This example uses a `@mediaType` trait on the payload to force a custom

    # content-type to be serialized.

    @app.http_payload_traits_with_media_type

    def http_payload_traits_with_media_type(input: input::HttpPayloadTraitsWithMediaTypeInput, ctx: Context) -> output::HttpPayloadTraitsWithMediaTypeOutput:

        raise NotImplementedError

    # This example serializes a structure in the payload.

    #

    # Note that serializing a structure changes the wrapper element name

    # to match the targeted structure.

    @app.http_payload_with_structure

    def http_payload_with_structure(input: input::HttpPayloadWithStructureInput, ctx: Context) -> output::HttpPayloadWithStructureOutput:

        raise NotImplementedError

    # This example serializes a union in the payload.

    @app.http_payload_with_union

    def http_payload_with_union(input: input::HttpPayloadWithUnionInput, ctx: Context) -> output::HttpPayloadWithUnionOutput:

        raise NotImplementedError

    # This examples adds headers to the input of a request and response by prefix.

    @app.http_prefix_headers

    def http_prefix_headers(input: input::HttpPrefixHeadersInput, ctx: Context) -> output::HttpPrefixHeadersOutput:

        raise NotImplementedError

    # Clients that perform this test extract all headers from the response.

    @app.http_prefix_headers_in_response

    def http_prefix_headers_in_response(input: input::HttpPrefixHeadersInResponseInput, ctx: Context) -> output::HttpPrefixHeadersInResponseOutput:

        raise NotImplementedError

    @app.http_request_with_float_labels

    def http_request_with_float_labels(input: input::HttpRequestWithFloatLabelsInput, ctx: Context) -> output::HttpRequestWithFloatLabelsOutput:

        raise NotImplementedError

    @app.http_request_with_greedy_label_in_path

    def http_request_with_greedy_label_in_path(input: input::HttpRequestWithGreedyLabelInPathInput, ctx: Context) -> output::HttpRequestWithGreedyLabelInPathOutput:

        raise NotImplementedError

    # The example tests how requests are serialized when there's no input

    # payload but there are HTTP labels.

    @app.http_request_with_labels

    def http_request_with_labels(input: input::HttpRequestWithLabelsInput, ctx: Context) -> output::HttpRequestWithLabelsOutput:

        raise NotImplementedError

    # The example tests how requests serialize different timestamp formats in the

    # URI path.

    @app.http_request_with_labels_and_timestamp_format

    def http_request_with_labels_and_timestamp_format(input: input::HttpRequestWithLabelsAndTimestampFormatInput, ctx: Context) -> output::HttpRequestWithLabelsAndTimestampFormatOutput:

        raise NotImplementedError

    @app.http_request_with_regex_literal

    def http_request_with_regex_literal(input: input::HttpRequestWithRegexLiteralInput, ctx: Context) -> output::HttpRequestWithRegexLiteralOutput:

        raise NotImplementedError

    @app.http_response_code

    def http_response_code(input: input::HttpResponseCodeInput, ctx: Context) -> output::HttpResponseCodeOutput:

        raise NotImplementedError

    @app.http_string_payload

    def http_string_payload(input: input::HttpStringPayloadInput, ctx: Context) -> output::HttpStringPayloadOutput:

        raise NotImplementedError

    # This example ensures that query string bound request parameters are

    # serialized in the body of responses if the structure is used in both

    # the request and response.

    @app.ignore_query_params_in_response

    def ignore_query_params_in_response(input: input::IgnoreQueryParamsInResponseInput, ctx: Context) -> output::IgnoreQueryParamsInResponseOutput:

        raise NotImplementedError

    # The example tests how requests and responses are serialized when there is

    # no input or output payload but there are HTTP header bindings.

    @app.input_and_output_with_headers

    def input_and_output_with_headers(input: input::InputAndOutputWithHeadersInput, ctx: Context) -> output::InputAndOutputWithHeadersOutput:

        raise NotImplementedError

    # Blobs are base64 encoded

    @app.json_blobs

    def json_blobs(input: input::JsonBlobsInput, ctx: Context) -> output::JsonBlobsOutput:

        raise NotImplementedError

    # This example serializes enums as top level properties, in lists, sets, and maps.

    @app.json_enums

    def json_enums(input: input::JsonEnumsInput, ctx: Context) -> output::JsonEnumsOutput:

        raise NotImplementedError

    # This example serializes intEnums as top level properties, in lists, sets, and maps.

    @app.json_int_enums

    def json_int_enums(input: input::JsonIntEnumsInput, ctx: Context) -> output::JsonIntEnumsOutput:

        raise NotImplementedError

    # This test case serializes JSON lists for the following cases for both

    # input and output:

    #

    # 1. Normal JSON lists.

    # 2. Normal JSON sets.

    # 3. JSON lists of lists.

    # 4. Lists of structures.

    @app.json_lists

    def json_lists(input: input::JsonListsInput, ctx: Context) -> output::JsonListsOutput:

        raise NotImplementedError

    # The example tests basic map serialization.

    @app.json_maps

    def json_maps(input: input::JsonMapsInput, ctx: Context) -> output::JsonMapsOutput:

        raise NotImplementedError

    # This tests how timestamps are serialized, including using the

    # default format of date-time and various @timestampFormat trait

    # values.

    @app.json_timestamps

    def json_timestamps(input: input::JsonTimestampsInput, ctx: Context) -> output::JsonTimestampsOutput:

        raise NotImplementedError

    # This operation uses unions for inputs and outputs.

    @app.json_unions

    def json_unions(input: input::JsonUnionsInput, ctx: Context) -> output::JsonUnionsOutput:

        raise NotImplementedError

    @app.malformed_accept_with_body

    def malformed_accept_with_body(input: input::MalformedAcceptWithBodyInput, ctx: Context) -> output::MalformedAcceptWithBodyOutput:

        raise NotImplementedError

    @app.malformed_accept_with_generic_string

    def malformed_accept_with_generic_string(input: input::MalformedAcceptWithGenericStringInput, ctx: Context) -> output::MalformedAcceptWithGenericStringOutput:

        raise NotImplementedError

    @app.malformed_accept_with_payload

    def malformed_accept_with_payload(input: input::MalformedAcceptWithPayloadInput, ctx: Context) -> output::MalformedAcceptWithPayloadOutput:

        raise NotImplementedError

    @app.malformed_blob

    def malformed_blob(input: input::MalformedBlobInput, ctx: Context) -> output::MalformedBlobOutput:

        raise NotImplementedError

    @app.malformed_boolean

    def malformed_boolean(input: input::MalformedBooleanInput, ctx: Context) -> output::MalformedBooleanOutput:

        raise NotImplementedError

    @app.malformed_byte

    def malformed_byte(input: input::MalformedByteInput, ctx: Context) -> output::MalformedByteOutput:

        raise NotImplementedError

    @app.malformed_content_type_with_body

    def malformed_content_type_with_body(input: input::MalformedContentTypeWithBodyInput, ctx: Context) -> output::MalformedContentTypeWithBodyOutput:

        raise NotImplementedError

    @app.malformed_content_type_with_generic_string

    def malformed_content_type_with_generic_string(input: input::MalformedContentTypeWithGenericStringInput, ctx: Context) -> output::MalformedContentTypeWithGenericStringOutput:

        raise NotImplementedError

    @app.malformed_content_type_without_body

    def malformed_content_type_without_body(input: input::MalformedContentTypeWithoutBodyInput, ctx: Context) -> output::MalformedContentTypeWithoutBodyOutput:

        raise NotImplementedError

    @app.malformed_content_type_without_body_empty_input

    def malformed_content_type_without_body_empty_input(input: input::MalformedContentTypeWithoutBodyEmptyInputInput, ctx: Context) -> output::MalformedContentTypeWithoutBodyEmptyInputOutput:

        raise NotImplementedError

    @app.malformed_content_type_with_payload

    def malformed_content_type_with_payload(input: input::MalformedContentTypeWithPayloadInput, ctx: Context) -> output::MalformedContentTypeWithPayloadOutput:

        raise NotImplementedError

    @app.malformed_double

    def malformed_double(input: input::MalformedDoubleInput, ctx: Context) -> output::MalformedDoubleOutput:

        raise NotImplementedError

    @app.malformed_float

    def malformed_float(input: input::MalformedFloatInput, ctx: Context) -> output::MalformedFloatOutput:

        raise NotImplementedError

    @app.malformed_integer

    def malformed_integer(input: input::MalformedIntegerInput, ctx: Context) -> output::MalformedIntegerOutput:

        raise NotImplementedError

    @app.malformed_list

    def malformed_list(input: input::MalformedListInput, ctx: Context) -> output::MalformedListOutput:

        raise NotImplementedError

    @app.malformed_long

    def malformed_long(input: input::MalformedLongInput, ctx: Context) -> output::MalformedLongOutput:

        raise NotImplementedError

    @app.malformed_map

    def malformed_map(input: input::MalformedMapInput, ctx: Context) -> output::MalformedMapOutput:

        raise NotImplementedError

    @app.malformed_request_body

    def malformed_request_body(input: input::MalformedRequestBodyInput, ctx: Context) -> output::MalformedRequestBodyOutput:

        raise NotImplementedError

    @app.malformed_short

    def malformed_short(input: input::MalformedShortInput, ctx: Context) -> output::MalformedShortOutput:

        raise NotImplementedError

    @app.malformed_string

    def malformed_string(input: input::MalformedStringInput, ctx: Context) -> output::MalformedStringOutput:

        raise NotImplementedError

    @app.malformed_timestamp_body_date_time

    def malformed_timestamp_body_date_time(input: input::MalformedTimestampBodyDateTimeInput, ctx: Context) -> output::MalformedTimestampBodyDateTimeOutput:

        raise NotImplementedError

    @app.malformed_timestamp_body_default

    def malformed_timestamp_body_default(input: input::MalformedTimestampBodyDefaultInput, ctx: Context) -> output::MalformedTimestampBodyDefaultOutput:

        raise NotImplementedError

    @app.malformed_timestamp_body_http_date

    def malformed_timestamp_body_http_date(input: input::MalformedTimestampBodyHttpDateInput, ctx: Context) -> output::MalformedTimestampBodyHttpDateOutput:

        raise NotImplementedError

    @app.malformed_timestamp_header_date_time

    def malformed_timestamp_header_date_time(input: input::MalformedTimestampHeaderDateTimeInput, ctx: Context) -> output::MalformedTimestampHeaderDateTimeOutput:

        raise NotImplementedError

    @app.malformed_timestamp_header_default

    def malformed_timestamp_header_default(input: input::MalformedTimestampHeaderDefaultInput, ctx: Context) -> output::MalformedTimestampHeaderDefaultOutput:

        raise NotImplementedError

    @app.malformed_timestamp_header_epoch

    def malformed_timestamp_header_epoch(input: input::MalformedTimestampHeaderEpochInput, ctx: Context) -> output::MalformedTimestampHeaderEpochOutput:

        raise NotImplementedError

    @app.malformed_timestamp_path_default

    def malformed_timestamp_path_default(input: input::MalformedTimestampPathDefaultInput, ctx: Context) -> output::MalformedTimestampPathDefaultOutput:

        raise NotImplementedError

    @app.malformed_timestamp_path_epoch

    def malformed_timestamp_path_epoch(input: input::MalformedTimestampPathEpochInput, ctx: Context) -> output::MalformedTimestampPathEpochOutput:

        raise NotImplementedError

    @app.malformed_timestamp_path_http_date

    def malformed_timestamp_path_http_date(input: input::MalformedTimestampPathHttpDateInput, ctx: Context) -> output::MalformedTimestampPathHttpDateOutput:

        raise NotImplementedError

    @app.malformed_timestamp_query_default

    def malformed_timestamp_query_default(input: input::MalformedTimestampQueryDefaultInput, ctx: Context) -> output::MalformedTimestampQueryDefaultOutput:

        raise NotImplementedError

    @app.malformed_timestamp_query_epoch

    def malformed_timestamp_query_epoch(input: input::MalformedTimestampQueryEpochInput, ctx: Context) -> output::MalformedTimestampQueryEpochOutput:

        raise NotImplementedError

    @app.malformed_timestamp_query_http_date

    def malformed_timestamp_query_http_date(input: input::MalformedTimestampQueryHttpDateInput, ctx: Context) -> output::MalformedTimestampQueryHttpDateOutput:

        raise NotImplementedError

    @app.malformed_union

    def malformed_union(input: input::MalformedUnionInput, ctx: Context) -> output::MalformedUnionOutput:

        raise NotImplementedError

    # This example ensures that mediaType strings are base64 encoded in headers.

    @app.media_type_header

    def media_type_header(input: input::MediaTypeHeaderInput, ctx: Context) -> output::MediaTypeHeaderOutput:

        raise NotImplementedError

    # The example tests how requests and responses are serialized when there's

    # no request or response payload because the operation has no input or output.

    # While this should be rare, code generators must support this.

    @app.no_input_and_no_output

    def no_input_and_no_output(input: input::NoInputAndNoOutputInput, ctx: Context) -> output::NoInputAndNoOutputOutput:

        raise NotImplementedError

    # The example tests how requests and responses are serialized when there's

    # no request or response payload because the operation has no input and the

    # output is empty. While this should be rare, code generators must support

    # this.

    @app.no_input_and_output

    def no_input_and_output(input: input::NoInputAndOutputInput, ctx: Context) -> output::NoInputAndOutputOutput:

        raise NotImplementedError

    # Null headers are not sent over the wire, empty headers are serialized to ""

    @app.null_and_empty_headers_client

    def null_and_empty_headers_client(input: input::NullAndEmptyHeadersClientInput, ctx: Context) -> output::NullAndEmptyHeadersClientOutput:

        raise NotImplementedError

    # Null headers are not sent over the wire, empty headers are serialized to ""

    @app.null_and_empty_headers_server

    def null_and_empty_headers_server(input: input::NullAndEmptyHeadersServerInput, ctx: Context) -> output::NullAndEmptyHeadersServerOutput:

        raise NotImplementedError

    # Omits null, but serializes empty string value.

    @app.omits_null_serializes_empty_string

    def omits_null_serializes_empty_string(input: input::OmitsNullSerializesEmptyStringInput, ctx: Context) -> output::OmitsNullSerializesEmptyStringOutput:

        raise NotImplementedError

    # Omits serializing empty lists. Because empty strings are serilized as

    # `Foo=`, empty lists cannot also be serialized as `Foo=` and instead

    # must be omitted.

    @app.omits_serializing_empty_lists

    def omits_serializing_empty_lists(input: input::OmitsSerializingEmptyListsInput, ctx: Context) -> output::OmitsSerializingEmptyListsOutput:

        raise NotImplementedError

    @app.operation_with_defaults

    def operation_with_defaults(input: input::OperationWithDefaultsInput, ctx: Context) -> output::OperationWithDefaultsOutput:

        raise NotImplementedError

    @app.operation_with_nested_structure

    def operation_with_nested_structure(input: input::OperationWithNestedStructureInput, ctx: Context) -> output::OperationWithNestedStructureOutput:

        raise NotImplementedError

    # This operation defines a union with a Unit member.

    @app.post_player_action

    def post_player_action(input: input::PostPlayerActionInput, ctx: Context) -> output::PostPlayerActionOutput:

        raise NotImplementedError

    # This operation defines a union that uses jsonName on some members.

    @app.post_union_with_json_name

    def post_union_with_json_name(input: input::PostUnionWithJsonNameInput, ctx: Context) -> output::PostUnionWithJsonNameOutput:

        raise NotImplementedError

    @app.put_with_content_encoding

    def put_with_content_encoding(input: input::PutWithContentEncodingInput, ctx: Context) -> output::PutWithContentEncodingOutput:

        raise NotImplementedError

    # Automatically adds idempotency tokens.

    @app.query_idempotency_token_auto_fill

    def query_idempotency_token_auto_fill(input: input::QueryIdempotencyTokenAutoFillInput, ctx: Context) -> output::QueryIdempotencyTokenAutoFillOutput:

        raise NotImplementedError

    @app.query_params_as_string_list_map

    def query_params_as_string_list_map(input: input::QueryParamsAsStringListMapInput, ctx: Context) -> output::QueryParamsAsStringListMapOutput:

        raise NotImplementedError

    @app.query_precedence

    def query_precedence(input: input::QueryPrecedenceInput, ctx: Context) -> output::QueryPrecedenceOutput:

        raise NotImplementedError

    # Recursive shapes

    @app.recursive_shapes

    def recursive_shapes(input: input::RecursiveShapesInput, ctx: Context) -> output::RecursiveShapesOutput:

        raise NotImplementedError

    @app.response_code_http_fallback

    def response_code_http_fallback(input: input::ResponseCodeHttpFallbackInput, ctx: Context) -> output::ResponseCodeHttpFallbackOutput:

        raise NotImplementedError

    @app.response_code_required

    def response_code_required(input: input::ResponseCodeRequiredInput, ctx: Context) -> output::ResponseCodeRequiredOutput:

        raise NotImplementedError

    @app.simple_scalar_properties

    def simple_scalar_properties(input: input::SimpleScalarPropertiesInput, ctx: Context) -> output::SimpleScalarPropertiesOutput:

        raise NotImplementedError

    @app.sparse_json_lists

    def sparse_json_lists(input: input::SparseJsonListsInput, ctx: Context) -> output::SparseJsonListsOutput:

        raise NotImplementedError

    # This example tests sparse map serialization.

    @app.sparse_json_maps

    def sparse_json_maps(input: input::SparseJsonMapsInput, ctx: Context) -> output::SparseJsonMapsOutput:

        raise NotImplementedError

    # This examples serializes a streaming blob shape in the request body.

    #

    # In this example, no JSON document is synthesized because the payload is

    # not a structure or a union type.

    @app.streaming_traits

    def streaming_traits(input: input::StreamingTraitsInput, ctx: Context) -> output::StreamingTraitsOutput:

        raise NotImplementedError

    # This examples serializes a streaming blob shape with a required content

    # length in the request body.

    #

    # In this example, no JSON document is synthesized because the payload is

    # not a structure or a union type.

    @app.streaming_traits_require_length

    def streaming_traits_require_length(input: input::StreamingTraitsRequireLengthInput, ctx: Context) -> output::StreamingTraitsRequireLengthOutput:

        raise NotImplementedError

    # This examples serializes a streaming media-typed blob shape in the request body.

    #

    # This examples uses a `@mediaType` trait on the payload to force a custom

    # content-type to be serialized.

    @app.streaming_traits_with_media_type

    def streaming_traits_with_media_type(input: input::StreamingTraitsWithMediaTypeInput, ctx: Context) -> output::StreamingTraitsWithMediaTypeOutput:

        raise NotImplementedError

    # This example operation serializes a structure in the HTTP body.

    #

    # It should ensure Content-Type: application/json is

    # used in all requests and that an "empty" body is

    # an empty JSON document ({}).

    #

    @app.test_body_structure

    def test_body_structure(input: input::TestBodyStructureInput, ctx: Context) -> output::TestBodyStructureOutput:

        raise NotImplementedError

    # This example GET operation has no input and serializes a request without a HTTP body.

    #

    # These tests are to ensure we do not attach a body or related headers

    # (Content-Length, Content-Type) to operations that semantically

    # cannot produce an HTTP body.

    #

    @app.test_get_no_input_no_payload

    def test_get_no_input_no_payload(input: input::TestGetNoInputNoPayloadInput, ctx: Context) -> output::TestGetNoInputNoPayloadOutput:

        raise NotImplementedError

    # This example GET operation serializes a request without a modeled HTTP body.

    #

    # These tests are to ensure we do not attach a body or related headers

    # (Content-Length, Content-Type) to operations that semantically

    # cannot produce an HTTP body.

    #

    @app.test_get_no_payload

    def test_get_no_payload(input: input::TestGetNoPayloadInput, ctx: Context) -> output::TestGetNoPayloadOutput:

        raise NotImplementedError

    # This example operation serializes a payload targeting a blob.

    #

    # The Blob shape is not structured content and we cannot

    # make assumptions about what data will be sent. This test ensures

    # only a generic "Content-Type: application/octet-stream" header

    # is used, and that we are not treating an empty body as an

    # empty JSON document.

    #

    @app.test_payload_blob

    def test_payload_blob(input: input::TestPayloadBlobInput, ctx: Context) -> output::TestPayloadBlobOutput:

        raise NotImplementedError

    # This example operation serializes a payload targeting a structure.

    #

    # This enforces the same requirements as TestBodyStructure

    # but with the body specified by the @httpPayload trait.

    #

    @app.test_payload_structure

    def test_payload_structure(input: input::TestPayloadStructureInput, ctx: Context) -> output::TestPayloadStructureOutput:

        raise NotImplementedError

    # This example POST operation has no input and serializes a request without a HTTP body.

    #

    # These tests are to ensure we do not attach a body or related headers

    # (Content-Type) to a POST operation with no modeled input.

    #

    @app.test_post_no_input_no_payload

    def test_post_no_input_no_payload(input: input::TestPostNoInputNoPayloadInput, ctx: Context) -> output::TestPostNoInputNoPayloadOutput:

        raise NotImplementedError

    # This example POST operation serializes a request without a modeled HTTP body.

    #

    # These tests are to ensure we do not attach a body or related headers

    # (Content-Type) to a POST operation with no modeled payload.

    #

    @app.test_post_no_payload

    def test_post_no_payload(input: input::TestPostNoPayloadInput, ctx: Context) -> output::TestPostNoPayloadOutput:

        raise NotImplementedError

    # This example tests how timestamp request and response headers are serialized.

    @app.timestamp_format_headers

    def timestamp_format_headers(input: input::TimestampFormatHeadersInput, ctx: Context) -> output::TimestampFormatHeadersOutput:

        raise NotImplementedError

    # This test is similar to NoInputAndNoOutput, but uses explicit Unit types.

    @app.unit_input_and_output

    def unit_input_and_output(input: input::UnitInputAndOutputInput, ctx: Context) -> output::UnitInputAndOutputOutput:

        raise NotImplementedError

    app.run()

    ```

    Any of operations above can be written as well prepending the `async` keyword and

    the Python application will automatically handle it and schedule it on the event loop for you.

    """

    def all_query_string_types(self, func: typing.Union[typing.Callable[[rest_json.input.AllQueryStringTypesInput, Ctx], typing.Union[rest_json.output.AllQueryStringTypesOutput, typing.Awaitable[rest_json.output.AllQueryStringTypesOutput]]], typing.Callable[[rest_json.input.AllQueryStringTypesInput], typing.Union[rest_json.output.AllQueryStringTypesOutput, typing.Awaitable[rest_json.output.AllQueryStringTypesOutput]]]]) -> None:

        """

        Method to register `all_query_string_types` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def constant_and_variable_query_string(self, func: typing.Union[typing.Callable[[rest_json.input.ConstantAndVariableQueryStringInput, Ctx], typing.Union[rest_json.output.ConstantAndVariableQueryStringOutput, typing.Awaitable[rest_json.output.ConstantAndVariableQueryStringOutput]]], typing.Callable[[rest_json.input.ConstantAndVariableQueryStringInput], typing.Union[rest_json.output.ConstantAndVariableQueryStringOutput, typing.Awaitable[rest_json.output.ConstantAndVariableQueryStringOutput]]]]) -> None:

        """

        Method to register `constant_and_variable_query_string` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def constant_query_string(self, func: typing.Union[typing.Callable[[rest_json.input.ConstantQueryStringInput, Ctx], typing.Union[rest_json.output.ConstantQueryStringOutput, typing.Awaitable[rest_json.output.ConstantQueryStringOutput]]], typing.Callable[[rest_json.input.ConstantQueryStringInput], typing.Union[rest_json.output.ConstantQueryStringOutput, typing.Awaitable[rest_json.output.ConstantQueryStringOutput]]]]) -> None:

        """

        Method to register `constant_query_string` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def content_type_parameters(self, func: typing.Union[typing.Callable[[rest_json.input.ContentTypeParametersInput, Ctx], typing.Union[rest_json.output.ContentTypeParametersOutput, typing.Awaitable[rest_json.output.ContentTypeParametersOutput]]], typing.Callable[[rest_json.input.ContentTypeParametersInput], typing.Union[rest_json.output.ContentTypeParametersOutput, typing.Awaitable[rest_json.output.ContentTypeParametersOutput]]]]) -> None:

        """

        Method to register `content_type_parameters` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def context(self, context: Ctx) -> None:

        """

        Register a context object that will be shared between handlers.

        """

        ...

    def datetime_offsets(self, func: typing.Union[typing.Callable[[rest_json.input.DatetimeOffsetsInput, Ctx], typing.Union[rest_json.output.DatetimeOffsetsOutput, typing.Awaitable[rest_json.output.DatetimeOffsetsOutput]]], typing.Callable[[rest_json.input.DatetimeOffsetsInput], typing.Union[rest_json.output.DatetimeOffsetsOutput, typing.Awaitable[rest_json.output.DatetimeOffsetsOutput]]]]) -> None:

        """

        Method to register `datetime_offsets` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def document_type(self, func: typing.Union[typing.Callable[[rest_json.input.DocumentTypeInput, Ctx], typing.Union[rest_json.output.DocumentTypeOutput, typing.Awaitable[rest_json.output.DocumentTypeOutput]]], typing.Callable[[rest_json.input.DocumentTypeInput], typing.Union[rest_json.output.DocumentTypeOutput, typing.Awaitable[rest_json.output.DocumentTypeOutput]]]]) -> None:

        """

        Method to register `document_type` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def document_type_as_map_value(self, func: typing.Union[typing.Callable[[rest_json.input.DocumentTypeAsMapValueInput, Ctx], typing.Union[rest_json.output.DocumentTypeAsMapValueOutput, typing.Awaitable[rest_json.output.DocumentTypeAsMapValueOutput]]], typing.Callable[[rest_json.input.DocumentTypeAsMapValueInput], typing.Union[rest_json.output.DocumentTypeAsMapValueOutput, typing.Awaitable[rest_json.output.DocumentTypeAsMapValueOutput]]]]) -> None:

        """

        Method to register `document_type_as_map_value` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def document_type_as_payload(self, func: typing.Union[typing.Callable[[rest_json.input.DocumentTypeAsPayloadInput, Ctx], typing.Union[rest_json.output.DocumentTypeAsPayloadOutput, typing.Awaitable[rest_json.output.DocumentTypeAsPayloadOutput]]], typing.Callable[[rest_json.input.DocumentTypeAsPayloadInput], typing.Union[rest_json.output.DocumentTypeAsPayloadOutput, typing.Awaitable[rest_json.output.DocumentTypeAsPayloadOutput]]]]) -> None:

        """

        Method to register `document_type_as_payload` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def empty_input_and_empty_output(self, func: typing.Union[typing.Callable[[rest_json.input.EmptyInputAndEmptyOutputInput, Ctx], typing.Union[rest_json.output.EmptyInputAndEmptyOutputOutput, typing.Awaitable[rest_json.output.EmptyInputAndEmptyOutputOutput]]], typing.Callable[[rest_json.input.EmptyInputAndEmptyOutputInput], typing.Union[rest_json.output.EmptyInputAndEmptyOutputOutput, typing.Awaitable[rest_json.output.EmptyInputAndEmptyOutputOutput]]]]) -> None:

        """

        Method to register `empty_input_and_empty_output` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def endpoint_operation(self, func: typing.Union[typing.Callable[[rest_json.input.EndpointOperationInput, Ctx], typing.Union[rest_json.output.EndpointOperationOutput, typing.Awaitable[rest_json.output.EndpointOperationOutput]]], typing.Callable[[rest_json.input.EndpointOperationInput], typing.Union[rest_json.output.EndpointOperationOutput, typing.Awaitable[rest_json.output.EndpointOperationOutput]]]]) -> None:

        """

        Method to register `endpoint_operation` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def endpoint_with_host_label_operation(self, func: typing.Union[typing.Callable[[rest_json.input.EndpointWithHostLabelOperationInput, Ctx], typing.Union[rest_json.output.EndpointWithHostLabelOperationOutput, typing.Awaitable[rest_json.output.EndpointWithHostLabelOperationOutput]]], typing.Callable[[rest_json.input.EndpointWithHostLabelOperationInput], typing.Union[rest_json.output.EndpointWithHostLabelOperationOutput, typing.Awaitable[rest_json.output.EndpointWithHostLabelOperationOutput]]]]) -> None:

        """

        Method to register `endpoint_with_host_label_operation` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def fractional_seconds(self, func: typing.Union[typing.Callable[[rest_json.input.FractionalSecondsInput, Ctx], typing.Union[rest_json.output.FractionalSecondsOutput, typing.Awaitable[rest_json.output.FractionalSecondsOutput]]], typing.Callable[[rest_json.input.FractionalSecondsInput], typing.Union[rest_json.output.FractionalSecondsOutput, typing.Awaitable[rest_json.output.FractionalSecondsOutput]]]]) -> None:

        """

        Method to register `fractional_seconds` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def greeting_with_errors(self, func: typing.Union[typing.Callable[[rest_json.input.GreetingWithErrorsInput, Ctx], typing.Union[rest_json.output.GreetingWithErrorsOutput, typing.Awaitable[rest_json.output.GreetingWithErrorsOutput]]], typing.Callable[[rest_json.input.GreetingWithErrorsInput], typing.Union[rest_json.output.GreetingWithErrorsOutput, typing.Awaitable[rest_json.output.GreetingWithErrorsOutput]]]]) -> None:

        """

        Method to register `greeting_with_errors` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def host_with_path_operation(self, func: typing.Union[typing.Callable[[rest_json.input.HostWithPathOperationInput, Ctx], typing.Union[rest_json.output.HostWithPathOperationOutput, typing.Awaitable[rest_json.output.HostWithPathOperationOutput]]], typing.Callable[[rest_json.input.HostWithPathOperationInput], typing.Union[rest_json.output.HostWithPathOperationOutput, typing.Awaitable[rest_json.output.HostWithPathOperationOutput]]]]) -> None:

        """

        Method to register `host_with_path_operation` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_checksum_required(self, func: typing.Union[typing.Callable[[rest_json.input.HttpChecksumRequiredInput, Ctx], typing.Union[rest_json.output.HttpChecksumRequiredOutput, typing.Awaitable[rest_json.output.HttpChecksumRequiredOutput]]], typing.Callable[[rest_json.input.HttpChecksumRequiredInput], typing.Union[rest_json.output.HttpChecksumRequiredOutput, typing.Awaitable[rest_json.output.HttpChecksumRequiredOutput]]]]) -> None:

        """

        Method to register `http_checksum_required` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_empty_prefix_headers(self, func: typing.Union[typing.Callable[[rest_json.input.HttpEmptyPrefixHeadersInput, Ctx], typing.Union[rest_json.output.HttpEmptyPrefixHeadersOutput, typing.Awaitable[rest_json.output.HttpEmptyPrefixHeadersOutput]]], typing.Callable[[rest_json.input.HttpEmptyPrefixHeadersInput], typing.Union[rest_json.output.HttpEmptyPrefixHeadersOutput, typing.Awaitable[rest_json.output.HttpEmptyPrefixHeadersOutput]]]]) -> None:

        """

        Method to register `http_empty_prefix_headers` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_enum_payload(self, func: typing.Union[typing.Callable[[rest_json.input.HttpEnumPayloadInput, Ctx], typing.Union[rest_json.output.HttpEnumPayloadOutput, typing.Awaitable[rest_json.output.HttpEnumPayloadOutput]]], typing.Callable[[rest_json.input.HttpEnumPayloadInput], typing.Union[rest_json.output.HttpEnumPayloadOutput, typing.Awaitable[rest_json.output.HttpEnumPayloadOutput]]]]) -> None:

        """

        Method to register `http_enum_payload` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_payload_traits(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPayloadTraitsInput, Ctx], typing.Union[rest_json.output.HttpPayloadTraitsOutput, typing.Awaitable[rest_json.output.HttpPayloadTraitsOutput]]], typing.Callable[[rest_json.input.HttpPayloadTraitsInput], typing.Union[rest_json.output.HttpPayloadTraitsOutput, typing.Awaitable[rest_json.output.HttpPayloadTraitsOutput]]]]) -> None:

        """

        Method to register `http_payload_traits` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_payload_traits_with_media_type(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPayloadTraitsWithMediaTypeInput, Ctx], typing.Union[rest_json.output.HttpPayloadTraitsWithMediaTypeOutput, typing.Awaitable[rest_json.output.HttpPayloadTraitsWithMediaTypeOutput]]], typing.Callable[[rest_json.input.HttpPayloadTraitsWithMediaTypeInput], typing.Union[rest_json.output.HttpPayloadTraitsWithMediaTypeOutput, typing.Awaitable[rest_json.output.HttpPayloadTraitsWithMediaTypeOutput]]]]) -> None:

        """

        Method to register `http_payload_traits_with_media_type` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_payload_with_structure(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPayloadWithStructureInput, Ctx], typing.Union[rest_json.output.HttpPayloadWithStructureOutput, typing.Awaitable[rest_json.output.HttpPayloadWithStructureOutput]]], typing.Callable[[rest_json.input.HttpPayloadWithStructureInput], typing.Union[rest_json.output.HttpPayloadWithStructureOutput, typing.Awaitable[rest_json.output.HttpPayloadWithStructureOutput]]]]) -> None:

        """

        Method to register `http_payload_with_structure` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_payload_with_union(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPayloadWithUnionInput, Ctx], typing.Union[rest_json.output.HttpPayloadWithUnionOutput, typing.Awaitable[rest_json.output.HttpPayloadWithUnionOutput]]], typing.Callable[[rest_json.input.HttpPayloadWithUnionInput], typing.Union[rest_json.output.HttpPayloadWithUnionOutput, typing.Awaitable[rest_json.output.HttpPayloadWithUnionOutput]]]]) -> None:

        """

        Method to register `http_payload_with_union` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_prefix_headers(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPrefixHeadersInput, Ctx], typing.Union[rest_json.output.HttpPrefixHeadersOutput, typing.Awaitable[rest_json.output.HttpPrefixHeadersOutput]]], typing.Callable[[rest_json.input.HttpPrefixHeadersInput], typing.Union[rest_json.output.HttpPrefixHeadersOutput, typing.Awaitable[rest_json.output.HttpPrefixHeadersOutput]]]]) -> None:

        """

        Method to register `http_prefix_headers` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_prefix_headers_in_response(self, func: typing.Union[typing.Callable[[rest_json.input.HttpPrefixHeadersInResponseInput, Ctx], typing.Union[rest_json.output.HttpPrefixHeadersInResponseOutput, typing.Awaitable[rest_json.output.HttpPrefixHeadersInResponseOutput]]], typing.Callable[[rest_json.input.HttpPrefixHeadersInResponseInput], typing.Union[rest_json.output.HttpPrefixHeadersInResponseOutput, typing.Awaitable[rest_json.output.HttpPrefixHeadersInResponseOutput]]]]) -> None:

        """

        Method to register `http_prefix_headers_in_response` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.

        """

        ...

    def http_request_with_float_labels(self, func: typing.Union[typing.Callable[[rest_json.input.HttpRequestWithFloatLabelsInput, Ctx], typing.Union[rest_json.output.HttpRequestWithFloatLabelsOutput, typing.Awaitable[rest_json.output.HttpRequestWithFloatLabelsOutput]]], typing.Callable[[rest_json.input.HttpRequestWithFloatLabelsInput], typing.Union[rest_json.output.HttpRequestWithFloatLabelsOutput, typing.Awaitable[rest_json.output.HttpRequestWithFloatLabelsOutput]]]]) -> None:

        """

        Method to register `http_request_with_float_labels` Python implementation inside the handlers map.

        It can be used as a function decorator in Python.