Server Test Python

    let model = ::pyo3::types::PyModule::new(py, "model")?;

    model.add_class::<crate::model::ValidationExceptionField>()?;

    error.add_class::<crate::error::StorageAccessNotAuthorized>()?;

    output.add_class::<crate::output::GetStorageOutput>()?;

    output.add_class::<crate::output::GetPokemonSpeciesOutput>()?;

    model.add_class::<crate::model::FlavorText>()?;

    model.add_class::<crate::model::Language>()?;

    input.add_class::<crate::input::GetPokemonSpeciesInput>()?;

    output.add_class::<crate::output::StreamPokemonRadioOutput>()?;

    input.add_class::<crate::input::StreamPokemonRadioInput>()?;

    output.add_class::<crate::output::CheckHealthOutput>()?;

    input.add_class::<crate::input::CheckHealthInput>()?;

    error.add_class::<crate::error::InvalidPokeballError>()?;

    error.add_class::<crate::error::MasterBallUnsuccessful>()?;

    error.add_class::<crate::error::ThrottlingError>()?;

    error.add_class::<crate::error::UnsupportedRegionError>()?;

    output.add_class::<crate::output::CapturePokemonOutput>()?;

    model.add_class::<crate::model::PyUnionMarkerCapturePokemonEvents>()?;

    model.add_class::<crate::model::CaptureEvent>()?;

    input.add_class::<crate::input::CapturePokemonInput>()?;

    output.add_class::<crate::output::DoNothingOutput>()?;

    input.add_class::<crate::input::DoNothingInput>()?;

    output.add_class::<crate::output::GetServerStatisticsOutput>()?;

    input.add_class::<crate::input::GetServerStatisticsInput>()?;

/// Python handler for operation `GetStorage`.

pub(crate) async fn get_storage(

    input: crate::input::GetStorageInput,

) -> std::result::Result<crate::output::GetStorageOutput, crate::error::GetStorageError> {

        ::tracing::trace!(name = "get_storage", "executing python handler coroutine");

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::GetStorageOutput>(py))

        ::tracing::trace!(name = "get_storage", "executing python handler function");

            output.extract::<crate::output::GetStorageOutput>()

/// Python handler for operation `GetPokemonSpecies`.

pub(crate) async fn get_pokemon_species(

    input: crate::input::GetPokemonSpeciesInput,

) -> std::result::Result<crate::output::GetPokemonSpeciesOutput, crate::error::GetPokemonSpeciesError>

{

        ::tracing::trace!(

            name = "get_pokemon_species",

            "executing python handler coroutine"

        );

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::GetPokemonSpeciesOutput>(py))

        ::tracing::trace!(

            name = "get_pokemon_species",

            "executing python handler function"

        );

            output.extract::<crate::output::GetPokemonSpeciesOutput>()

/// Python handler for operation `StreamPokemonRadio`.

pub(crate) async fn stream_pokemon_radio(

    input: crate::input::StreamPokemonRadioInput,

) -> std::result::Result<

    crate::output::StreamPokemonRadioOutput,

    crate::error::StreamPokemonRadioError,

> {

            name = "stream_pokemon_radio",

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::StreamPokemonRadioOutput>(py))

            name = "stream_pokemon_radio",

            output.extract::<crate::output::StreamPokemonRadioOutput>()

/// Python handler for operation `CapturePokemon`.

pub(crate) async fn capture_pokemon(

    input: crate::input::CapturePokemonInput,

) -> std::result::Result<crate::output::CapturePokemonOutput, crate::error::CapturePokemonError> {

            name = "capture_pokemon",

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::CapturePokemonOutput>(py))

            name = "capture_pokemon",

            output.extract::<crate::output::CapturePokemonOutput>()

/// Python handler for operation `DoNothing`.

pub(crate) async fn do_nothing(

    input: crate::input::DoNothingInput,

) -> std::result::Result<crate::output::DoNothingOutput, crate::error::DoNothingError> {

        ::tracing::trace!(name = "do_nothing", "executing python handler coroutine");

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::DoNothingOutput>(py))

        ::tracing::trace!(name = "do_nothing", "executing python handler function");

            output.extract::<crate::output::DoNothingOutput>()

/// Python handler for operation `GetServerStatistics`.

pub(crate) async fn get_server_statistics(

    input: crate::input::GetServerStatisticsInput,

) -> std::result::Result<

    crate::output::GetServerStatisticsOutput,

    crate::error::GetServerStatisticsError,

> {

        ::tracing::trace!(

            name = "get_server_statistics",

            "executing python handler coroutine"

        );

            ::pyo3::Python::with_gil(|py| r.extract::<crate::output::GetServerStatisticsOutput>(py))

        ::tracing::trace!(

            name = "get_server_statistics",

            "executing python handler function"

        );

            output.extract::<crate::output::GetServerStatisticsOutput>()

features = ["http-02x"]

features = ["attributes", "tokio-runtime"]

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.