Server Test Python

rev. 1d902e1fdfff69f8aafe42b2306fa317ef96d562

Files changed:

tmp-codegen-diff/codegen-server-test-python/naming_test_ops/rust-server-codegen-python/python/naming_test_ops/socket/init.pyi

@@ -1,0 +27,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_ops/rust-server-codegen-python/python/naming_test_ops/tls/init.pyi

@@ -1,0 +10,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_ops/rust-server-codegen-python/python/naming_test_ops/types/init.pyi

@@ -1,0 +209,0 @@

 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.
     """

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/init.pyi

@@ -1,0 +95,0 @@

 import naming_test_structs.input
 import naming_test_structs.middleware
 import naming_test_structs.output
 import naming_test_structs.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 naming_test_structs import input
     from naming_test_structs import output
     from naming_test_structs import error
     from naming_test_structs import middleware
     from naming_test_structs 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)
--
     @app.structs
     def structs(input: input::StructsInput, ctx: Context) -> output::StructsOutput:
         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 context(self, context: Ctx) -> None:
         """
         Register a context object that will be shared between handlers.
         """
         ...
--
--
     def middleware(self, func: typing.Callable[[naming_test_structs.middleware.Request, typing.Callable[[naming_test_structs.middleware.Request], typing.Awaitable[naming_test_structs.middleware.Response]]], typing.Awaitable[naming_test_structs.middleware.Response]]) -> None:
         """
         Register a Python function to be executed inside a Tower middleware layer.
         """
         ...
--
--
     def run(self, address: typing.Optional[str] = ..., port: typing.Optional[int] = ..., backlog: typing.Optional[int] = ..., workers: typing.Optional[int] = ..., tls: typing.Optional[naming_test_structs.tls.TlsConfig] = ...) -> None:
         """
         Main entrypoint: start the server on multiple workers.
         """
         ...
--
--
     def run_lambda(self) -> None:
         """
         Lambda entrypoint: start the server on Lambda.
         """
         ...
--
--
     def start_worker(self) -> None:
         """
         Build the service and start a single worker.
         """
         ...
--
--
     def structs(self, func: typing.Union[typing.Callable[[naming_test_structs.input.StructsInput, Ctx], typing.Union[naming_test_structs.output.StructsOutput, typing.Awaitable[naming_test_structs.output.StructsOutput]]], typing.Callable[[naming_test_structs.input.StructsInput], typing.Union[naming_test_structs.output.StructsOutput, typing.Awaitable[naming_test_structs.output.StructsOutput]]]]) -> None:
         """
         Method to register `structs` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def __init__(self) -> None:
         ...
--
--
 CODEGEN_VERSION: str = ...

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/aws_lambda/init.pyi

@@ -1,0 +143,0 @@

 import typing
--
 class ClientApplication:
     """
     AWS Mobile SDK client fields.
     """
--
     app_package_name: str
     """
     The package name for the mobile application invoking the function
     """
--
     app_title: str
     """
     The app title for the mobile app as registered with AWS' mobile services.
     """
--
     app_version_code: str
     """
     The app version code.
     """
--
     app_version_name: str
     """
     The version name of the application as registered with AWS' mobile services.
     """
--
     installation_id: str
     """
     The mobile app installation id
     """
--
 class ClientContext:
     """
     Client context sent by the AWS Mobile SDK.
     """
--
     client: ClientApplication
     """
     Information about the mobile application invoking the function.
     """
--
     custom: typing.Dict[str, str]
     """
     Custom properties attached to the mobile event context.
     """
--
     environment: typing.Dict[str, str]
     """
     Environment settings from the mobile client.
     """
--
 class CognitoIdentity:
     """
     Cognito identity information sent with the event
     """
--
     identity_id: str
     """
     The unique identity id for the Cognito credentials invoking the function.
     """
--
     identity_pool_id: str
     """
     The identity pool id the caller is "registered" with.
     """
--
 class Config:
     """
     Configuration derived from environment variables.
     """
--
     function_name: str
     """
     The name of the function.
     """
--
     log_group: str
     """
     The name of the Amazon CloudWatch Logs group for the function.
     """
--
     log_stream: str
     """
     The name of the Amazon CloudWatch Logs stream for the function.
     """
--
     memory: int
     """
     The amount of memory available to the function in MB.
     """
--
     version: str
     """
     The version of the function being executed.
     """
--
 class LambdaContext:
     """
     The Lambda function execution context. The values in this struct
     are populated using the [Lambda environment variables](https://docs.aws.amazon.com/lambda/latest/dg/current-supported-versions.html)
     and the headers returned by the poll request to the Runtime APIs.
     """
--
     client_context: typing.Optional[ClientContext]
     """
     The client context object sent by the AWS mobile SDK. This field is
     empty unless the function is invoked using an AWS mobile SDK.
     """
--
     deadline: int
     """
     The execution deadline for the current invocation in milliseconds.
     """
--
     env_config: Config
     """
     Lambda function configuration from the local environment variables.
     Includes information such as the function name, memory allocation,
     version, and log streams.
     """
--
     identity: typing.Optional[CognitoIdentity]
     """
     The Cognito identity that invoked the function. This field is empty
     unless the invocation request to the Lambda APIs was made using AWS
     credentials issues by Amazon Cognito Identity Pools.
     """
--
     invoked_function_arn: str
     """
     The ARN of the Lambda function being invoked.
     """
--
     request_id: str
     """
     The AWS request ID generated by the Lambda service.
     """
--
     xray_trace_id: typing.Optional[str]
     """
     The X-Ray trace ID for the current invocation.
     """

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/error/init.pyi

@@ -1,0 +8,0 @@

--
--
 class InternalServerError(Exception):
     message: str
--
     def __init__(self, message: str) -> None:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/input/init.pyi

@@ -1,0 +17,0 @@

 import naming_test_structs.model
 import typing
--
 class StructsInput:
     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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/logging/init.pyi

@@ -1,0 +26,0 @@

 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.
     """
     ...

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/middleware/init.pyi

@@ -1,0 +78,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/model/init.pyi

@@ -1,0 +76,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/output/init.pyi

@@ -1,0 +17,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/socket/init.pyi

@@ -1,0 +27,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/tls/init.pyi

@@ -1,0 +10,0 @@

 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:
         ...
--

tmp-codegen-diff/codegen-server-test-python/naming_test_structs/rust-server-codegen-python/python/naming_test_structs/types/init.pyi

@@ -1,0 +209,0 @@

 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.
     """

tmp-codegen-diff/codegen-server-test-python/pokemon-service-server-sdk/rust-server-codegen-python/python/pokemon_service_server_sdk/init.pyi

@@ -1,0 +175,0 @@

 import pokemon_service_server_sdk.input
 import pokemon_service_server_sdk.middleware
 import pokemon_service_server_sdk.output
 import pokemon_service_server_sdk.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 pokemon_service_server_sdk import input
     from pokemon_service_server_sdk import output
     from pokemon_service_server_sdk import error
     from pokemon_service_server_sdk import middleware
     from pokemon_service_server_sdk 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)
--
     # Capture Pokémons via event streams.
     @app.capture_pokemon
     def capture_pokemon(input: input::CapturePokemonInput, ctx: Context) -> output::CapturePokemonOutput:
         raise NotImplementedError
--
     # Health check operation, to check the service is up
     # Not yet a deep check
     @app.check_health
     def check_health(input: input::CheckHealthInput, ctx: Context) -> output::CheckHealthOutput:
         raise NotImplementedError
--
     # DoNothing operation, used to stress test the framework.
     @app.do_nothing
     def do_nothing(input: input::DoNothingInput, ctx: Context) -> output::DoNothingOutput:
         raise NotImplementedError
--
     # Retrieve information about a Pokémon species.
     @app.get_pokemon_species
     def get_pokemon_species(input: input::GetPokemonSpeciesInput, ctx: Context) -> output::GetPokemonSpeciesOutput:
         raise NotImplementedError
--
     # Retrieve HTTP server statistiscs, such as calls count.
     @app.get_server_statistics
     def get_server_statistics(input: input::GetServerStatisticsInput, ctx: Context) -> output::GetServerStatisticsOutput:
         raise NotImplementedError
--
     # Retrieve information about your Pokédex.
     @app.get_storage
     def get_storage(input: input::GetStorageInput, ctx: Context) -> output::GetStorageOutput:
         raise NotImplementedError
--
     # Fetch a radio song from the database and stream it back as a playable audio.
     @app.stream_pokemon_radio
     def stream_pokemon_radio(input: input::StreamPokemonRadioInput, ctx: Context) -> output::StreamPokemonRadioOutput:
         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 capture_pokemon(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.CapturePokemonInput, Ctx], typing.Union[pokemon_service_server_sdk.output.CapturePokemonOutput, typing.Awaitable[pokemon_service_server_sdk.output.CapturePokemonOutput]]], typing.Callable[[pokemon_service_server_sdk.input.CapturePokemonInput], typing.Union[pokemon_service_server_sdk.output.CapturePokemonOutput, typing.Awaitable[pokemon_service_server_sdk.output.CapturePokemonOutput]]]]) -> None:
         """
         Method to register `capture_pokemon` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def check_health(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.CheckHealthInput, Ctx], typing.Union[pokemon_service_server_sdk.output.CheckHealthOutput, typing.Awaitable[pokemon_service_server_sdk.output.CheckHealthOutput]]], typing.Callable[[pokemon_service_server_sdk.input.CheckHealthInput], typing.Union[pokemon_service_server_sdk.output.CheckHealthOutput, typing.Awaitable[pokemon_service_server_sdk.output.CheckHealthOutput]]]]) -> None:
         """
         Method to register `check_health` 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 do_nothing(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.DoNothingInput, Ctx], typing.Union[pokemon_service_server_sdk.output.DoNothingOutput, typing.Awaitable[pokemon_service_server_sdk.output.DoNothingOutput]]], typing.Callable[[pokemon_service_server_sdk.input.DoNothingInput], typing.Union[pokemon_service_server_sdk.output.DoNothingOutput, typing.Awaitable[pokemon_service_server_sdk.output.DoNothingOutput]]]]) -> None:
         """
         Method to register `do_nothing` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def get_pokemon_species(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.GetPokemonSpeciesInput, Ctx], typing.Union[pokemon_service_server_sdk.output.GetPokemonSpeciesOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetPokemonSpeciesOutput]]], typing.Callable[[pokemon_service_server_sdk.input.GetPokemonSpeciesInput], typing.Union[pokemon_service_server_sdk.output.GetPokemonSpeciesOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetPokemonSpeciesOutput]]]]) -> None:
         """
         Method to register `get_pokemon_species` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def get_server_statistics(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.GetServerStatisticsInput, Ctx], typing.Union[pokemon_service_server_sdk.output.GetServerStatisticsOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetServerStatisticsOutput]]], typing.Callable[[pokemon_service_server_sdk.input.GetServerStatisticsInput], typing.Union[pokemon_service_server_sdk.output.GetServerStatisticsOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetServerStatisticsOutput]]]]) -> None:
         """
         Method to register `get_server_statistics` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def get_storage(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.GetStorageInput, Ctx], typing.Union[pokemon_service_server_sdk.output.GetStorageOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetStorageOutput]]], typing.Callable[[pokemon_service_server_sdk.input.GetStorageInput], typing.Union[pokemon_service_server_sdk.output.GetStorageOutput, typing.Awaitable[pokemon_service_server_sdk.output.GetStorageOutput]]]]) -> None:
         """
         Method to register `get_storage` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def middleware(self, func: typing.Callable[[pokemon_service_server_sdk.middleware.Request, typing.Callable[[pokemon_service_server_sdk.middleware.Request], typing.Awaitable[pokemon_service_server_sdk.middleware.Response]]], typing.Awaitable[pokemon_service_server_sdk.middleware.Response]]) -> None:
         """
         Register a Python function to be executed inside a Tower middleware layer.
         """
         ...
--
--
     def run(self, address: typing.Optional[str] = ..., port: typing.Optional[int] = ..., backlog: typing.Optional[int] = ..., workers: typing.Optional[int] = ..., tls: typing.Optional[pokemon_service_server_sdk.tls.TlsConfig] = ...) -> None:
         """
         Main entrypoint: start the server on multiple workers.
         """
         ...
--
--
     def run_lambda(self) -> None:
         """
         Lambda entrypoint: start the server on Lambda.
         """
         ...
--
--
     def start_worker(self) -> None:
         """
         Build the service and start a single worker.
         """
         ...
--
--
     def stream_pokemon_radio(self, func: typing.Union[typing.Callable[[pokemon_service_server_sdk.input.StreamPokemonRadioInput, Ctx], typing.Union[pokemon_service_server_sdk.output.StreamPokemonRadioOutput, typing.Awaitable[pokemon_service_server_sdk.output.StreamPokemonRadioOutput]]], typing.Callable[[pokemon_service_server_sdk.input.StreamPokemonRadioInput], typing.Union[pokemon_service_server_sdk.output.StreamPokemonRadioOutput, typing.Awaitable[pokemon_service_server_sdk.output.StreamPokemonRadioOutput]]]]) -> None:
         """
         Method to register `stream_pokemon_radio` Python implementation inside the handlers map.
         It can be used as a function decorator in Python.
         """
         ...
--
--
     def __init__(self) -> None:
         ...
--
--
 CODEGEN_VERSION: str = ...

Pages: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17