1 - | import json_rpc10.input
|
2 - | import json_rpc10.middleware
|
3 - | import json_rpc10.output
|
4 - | import json_rpc10.tls
|
5 - | import typing
|
6 - |
|
7 - | Ctx = typing.TypeVar('Ctx')
|
8 - |
|
9 - | class App(typing.Generic[Ctx]):
|
10 - | """
|
11 - | Main Python application, used to register operations and context and start multiple
|
12 - | workers on the same shared socket.
|
13 - |
|
14 - | Operations can be registered using the application object as a decorator (`@app.operation_name`).
|
15 - |
|
16 - | Here's a full example to get you started:
|
17 - |
|
18 - | ```python
|
19 - | from json_rpc10 import input
|
20 - | from json_rpc10 import output
|
21 - | from json_rpc10 import error
|
22 - | from json_rpc10 import middleware
|
23 - | from json_rpc10 import App
|
24 - |
|
25 - | @dataclass
|
26 - | class Context:
|
27 - | counter: int = 0
|
28 - |
|
29 - | app = App()
|
30 - | app.context(Context())
|
31 - |
|
32 - | @app.request_middleware
|
33 - | def request_middleware(request: middleware::Request):
|
34 - | if request.get_header("x-amzn-id") != "secret":
|
35 - | raise middleware.MiddlewareException("Unsupported `x-amz-id` header", 401)
|
36 - |
|
37 - | # The example tests how servers must support requests
|
38 - | # containing a `Content-Type` header with parameters.
|
39 - | @app.content_type_parameters
|
40 - | def content_type_parameters(input: input::ContentTypeParametersInput, ctx: Context) -> output::ContentTypeParametersOutput:
|
41 - | raise NotImplementedError
|
42 - |
|
43 - | # The example tests how requests and responses are serialized when there's
|
44 - | # no request or response payload because the operation has an empty input
|
45 - | # and empty output structure that reuses the same shape. While this should
|
46 - | # be rare, code generators must support this.
|
47 - | @app.empty_input_and_empty_output
|
48 - | def empty_input_and_empty_output(input: input::EmptyInputAndEmptyOutputInput, ctx: Context) -> output::EmptyInputAndEmptyOutputOutput:
|
49 - | raise NotImplementedError
|
50 - |
|
51 - | @app.endpoint_operation
|
52 - | def endpoint_operation(input: input::EndpointOperationInput, ctx: Context) -> output::EndpointOperationOutput:
|
53 - | raise NotImplementedError
|
54 - |
|
55 - | @app.endpoint_with_host_label_operation
|
56 - | def endpoint_with_host_label_operation(input: input::EndpointWithHostLabelOperationInput, ctx: Context) -> output::EndpointWithHostLabelOperationOutput:
|
57 - | raise NotImplementedError
|
58 - |
|
59 - | # This operation has three possible return values:
|
60 - | #
|
61 - | # 1. A successful response in the form of GreetingWithErrorsOutput
|
62 - | # 2. An InvalidGreeting error.
|
63 - | # 3. A ComplexError error.
|
64 - | #
|
65 - | # Implementations must be able to successfully take a response and
|
66 - | # properly deserialize successful and error responses.
|
67 - | @app.greeting_with_errors
|
68 - | def greeting_with_errors(input: input::GreetingWithErrorsInput, ctx: Context) -> output::GreetingWithErrorsOutput:
|
69 - | raise NotImplementedError
|
70 - |
|
71 - | @app.host_with_path_operation
|
72 - | def host_with_path_operation(input: input::HostWithPathOperationInput, ctx: Context) -> output::HostWithPathOperationOutput:
|
73 - | raise NotImplementedError
|
74 - |
|
75 - | # This operation uses unions for inputs and outputs.
|
76 - | @app.json_unions
|
77 - | def json_unions(input: input::JsonUnionsInput, ctx: Context) -> output::JsonUnionsOutput:
|
78 - | raise NotImplementedError
|
79 - |
|
80 - | # The example tests how requests and responses are serialized when there's
|
81 - | # no request or response payload because the operation has no input or output.
|
82 - | # While this should be rare, code generators must support this.
|
83 - | @app.no_input_and_no_output
|
84 - | def no_input_and_no_output(input: input::NoInputAndNoOutputInput, ctx: Context) -> output::NoInputAndNoOutputOutput:
|
85 - | raise NotImplementedError
|
86 - |
|
87 - | # The example tests how requests and responses are serialized when there's
|
88 - | # no request or response payload because the operation has no input and the
|
89 - | # output is empty. While this should be rare, code generators must support
|
90 - | # this.
|
91 - | @app.no_input_and_output
|
92 - | def no_input_and_output(input: input::NoInputAndOutputInput, ctx: Context) -> output::NoInputAndOutputOutput:
|
93 - | raise NotImplementedError
|
94 - |
|
95 - | @app.operation_with_defaults
|
96 - | def operation_with_defaults(input: input::OperationWithDefaultsInput, ctx: Context) -> output::OperationWithDefaultsOutput:
|
97 - | raise NotImplementedError
|
98 - |
|
99 - | @app.operation_with_nested_structure
|
100 - | def operation_with_nested_structure(input: input::OperationWithNestedStructureInput, ctx: Context) -> output::OperationWithNestedStructureOutput:
|
101 - | raise NotImplementedError
|
102 - |
|
103 - | @app.operation_with_required_members
|
104 - | def operation_with_required_members(input: input::OperationWithRequiredMembersInput, ctx: Context) -> output::OperationWithRequiredMembersOutput:
|
105 - | raise NotImplementedError
|
106 - |
|
107 - | @app.put_with_content_encoding
|
108 - | def put_with_content_encoding(input: input::PutWithContentEncodingInput, ctx: Context) -> output::PutWithContentEncodingOutput:
|
109 - | raise NotImplementedError
|
110 - |
|
111 - | @app.simple_scalar_properties
|
112 - | def simple_scalar_properties(input: input::SimpleScalarPropertiesInput, ctx: Context) -> output::SimpleScalarPropertiesOutput:
|
113 - | raise NotImplementedError
|
114 - |
|
115 - | app.run()
|
116 - | ```
|
117 - |
|
118 - | Any of operations above can be written as well prepending the `async` keyword and
|
119 - | the Python application will automatically handle it and schedule it on the event loop for you.
|
120 - | """
|
121 - |
|
122 - | def content_type_parameters(self, func: typing.Union[typing.Callable[[json_rpc10.input.ContentTypeParametersInput, Ctx], typing.Union[json_rpc10.output.ContentTypeParametersOutput, typing.Awaitable[json_rpc10.output.ContentTypeParametersOutput]]], typing.Callable[[json_rpc10.input.ContentTypeParametersInput], typing.Union[json_rpc10.output.ContentTypeParametersOutput, typing.Awaitable[json_rpc10.output.ContentTypeParametersOutput]]]]) -> None:
|
123 - | """
|
124 - | Method to register `content_type_parameters` Python implementation inside the handlers map.
|
125 - | It can be used as a function decorator in Python.
|
126 - | """
|
127 - | ...
|
128 - |
|
129 - |
|
130 - | def context(self, context: Ctx) -> None:
|
131 - | """
|
132 - | Register a context object that will be shared between handlers.
|
133 - | """
|
134 - | ...
|
135 - |
|
136 - |
|
137 - | def empty_input_and_empty_output(self, func: typing.Union[typing.Callable[[json_rpc10.input.EmptyInputAndEmptyOutputInput, Ctx], typing.Union[json_rpc10.output.EmptyInputAndEmptyOutputOutput, typing.Awaitable[json_rpc10.output.EmptyInputAndEmptyOutputOutput]]], typing.Callable[[json_rpc10.input.EmptyInputAndEmptyOutputInput], typing.Union[json_rpc10.output.EmptyInputAndEmptyOutputOutput, typing.Awaitable[json_rpc10.output.EmptyInputAndEmptyOutputOutput]]]]) -> None:
|
138 - | """
|
139 - | Method to register `empty_input_and_empty_output` Python implementation inside the handlers map.
|
140 - | It can be used as a function decorator in Python.
|
141 - | """
|
142 - | ...
|
143 - |
|
144 - |
|
145 - | def endpoint_operation(self, func: typing.Union[typing.Callable[[json_rpc10.input.EndpointOperationInput, Ctx], typing.Union[json_rpc10.output.EndpointOperationOutput, typing.Awaitable[json_rpc10.output.EndpointOperationOutput]]], typing.Callable[[json_rpc10.input.EndpointOperationInput], typing.Union[json_rpc10.output.EndpointOperationOutput, typing.Awaitable[json_rpc10.output.EndpointOperationOutput]]]]) -> None:
|
146 - | """
|
147 - | Method to register `endpoint_operation` Python implementation inside the handlers map.
|
148 - | It can be used as a function decorator in Python.
|
149 - | """
|
150 - | ...
|
151 - |
|
152 - |
|
153 - | def endpoint_with_host_label_operation(self, func: typing.Union[typing.Callable[[json_rpc10.input.EndpointWithHostLabelOperationInput, Ctx], typing.Union[json_rpc10.output.EndpointWithHostLabelOperationOutput, typing.Awaitable[json_rpc10.output.EndpointWithHostLabelOperationOutput]]], typing.Callable[[json_rpc10.input.EndpointWithHostLabelOperationInput], typing.Union[json_rpc10.output.EndpointWithHostLabelOperationOutput, typing.Awaitable[json_rpc10.output.EndpointWithHostLabelOperationOutput]]]]) -> None:
|
154 - | """
|
155 - | Method to register `endpoint_with_host_label_operation` Python implementation inside the handlers map.
|
156 - | It can be used as a function decorator in Python.
|
157 - | """
|
158 - | ...
|
159 - |
|
160 - |
|
161 - | def greeting_with_errors(self, func: typing.Union[typing.Callable[[json_rpc10.input.GreetingWithErrorsInput, Ctx], typing.Union[json_rpc10.output.GreetingWithErrorsOutput, typing.Awaitable[json_rpc10.output.GreetingWithErrorsOutput]]], typing.Callable[[json_rpc10.input.GreetingWithErrorsInput], typing.Union[json_rpc10.output.GreetingWithErrorsOutput, typing.Awaitable[json_rpc10.output.GreetingWithErrorsOutput]]]]) -> None:
|
162 - | """
|
163 - | Method to register `greeting_with_errors` Python implementation inside the handlers map.
|
164 - | It can be used as a function decorator in Python.
|
165 - | """
|
166 - | ...
|
167 - |
|
168 - |
|
169 - | def host_with_path_operation(self, func: typing.Union[typing.Callable[[json_rpc10.input.HostWithPathOperationInput, Ctx], typing.Union[json_rpc10.output.HostWithPathOperationOutput, typing.Awaitable[json_rpc10.output.HostWithPathOperationOutput]]], typing.Callable[[json_rpc10.input.HostWithPathOperationInput], typing.Union[json_rpc10.output.HostWithPathOperationOutput, typing.Awaitable[json_rpc10.output.HostWithPathOperationOutput]]]]) -> None:
|
170 - | """
|
171 - | Method to register `host_with_path_operation` Python implementation inside the handlers map.
|
172 - | It can be used as a function decorator in Python.
|
173 - | """
|
174 - | ...
|
175 - |
|
176 - |
|
177 - | def json_unions(self, func: typing.Union[typing.Callable[[json_rpc10.input.JsonUnionsInput, Ctx], typing.Union[json_rpc10.output.JsonUnionsOutput, typing.Awaitable[json_rpc10.output.JsonUnionsOutput]]], typing.Callable[[json_rpc10.input.JsonUnionsInput], typing.Union[json_rpc10.output.JsonUnionsOutput, typing.Awaitable[json_rpc10.output.JsonUnionsOutput]]]]) -> None:
|
178 - | """
|
179 - | Method to register `json_unions` Python implementation inside the handlers map.
|
180 - | It can be used as a function decorator in Python.
|
181 - | """
|
182 - | ...
|
183 - |
|
184 - |
|
185 - | def middleware(self, func: typing.Callable[[json_rpc10.middleware.Request, typing.Callable[[json_rpc10.middleware.Request], typing.Awaitable[json_rpc10.middleware.Response]]], typing.Awaitable[json_rpc10.middleware.Response]]) -> None:
|
186 - | """
|
187 - | Register a Python function to be executed inside a Tower middleware layer.
|
188 - | """
|
189 - | ...
|
190 - |
|
191 - |
|
192 - | def no_input_and_no_output(self, func: typing.Union[typing.Callable[[json_rpc10.input.NoInputAndNoOutputInput, Ctx], typing.Union[json_rpc10.output.NoInputAndNoOutputOutput, typing.Awaitable[json_rpc10.output.NoInputAndNoOutputOutput]]], typing.Callable[[json_rpc10.input.NoInputAndNoOutputInput], typing.Union[json_rpc10.output.NoInputAndNoOutputOutput, typing.Awaitable[json_rpc10.output.NoInputAndNoOutputOutput]]]]) -> None:
|
193 - | """
|
194 - | Method to register `no_input_and_no_output` Python implementation inside the handlers map.
|
195 - | It can be used as a function decorator in Python.
|
196 - | """
|
197 - | ...
|
198 - |
|
199 - |
|
200 - | def no_input_and_output(self, func: typing.Union[typing.Callable[[json_rpc10.input.NoInputAndOutputInput, Ctx], typing.Union[json_rpc10.output.NoInputAndOutputOutput, typing.Awaitable[json_rpc10.output.NoInputAndOutputOutput]]], typing.Callable[[json_rpc10.input.NoInputAndOutputInput], typing.Union[json_rpc10.output.NoInputAndOutputOutput, typing.Awaitable[json_rpc10.output.NoInputAndOutputOutput]]]]) -> None:
|
201 - | """
|
202 - | Method to register `no_input_and_output` Python implementation inside the handlers map.
|
203 - | It can be used as a function decorator in Python.
|
204 - | """
|
205 - | ...
|
206 - |
|
207 - |
|
208 - | def operation_with_defaults(self, func: typing.Union[typing.Callable[[json_rpc10.input.OperationWithDefaultsInput, Ctx], typing.Union[json_rpc10.output.OperationWithDefaultsOutput, typing.Awaitable[json_rpc10.output.OperationWithDefaultsOutput]]], typing.Callable[[json_rpc10.input.OperationWithDefaultsInput], typing.Union[json_rpc10.output.OperationWithDefaultsOutput, typing.Awaitable[json_rpc10.output.OperationWithDefaultsOutput]]]]) -> None:
|
209 - | """
|
210 - | Method to register `operation_with_defaults` Python implementation inside the handlers map.
|
211 - | It can be used as a function decorator in Python.
|
212 - | """
|
213 - | ...
|
214 - |
|
215 - |
|
216 - | def operation_with_nested_structure(self, func: typing.Union[typing.Callable[[json_rpc10.input.OperationWithNestedStructureInput, Ctx], typing.Union[json_rpc10.output.OperationWithNestedStructureOutput, typing.Awaitable[json_rpc10.output.OperationWithNestedStructureOutput]]], typing.Callable[[json_rpc10.input.OperationWithNestedStructureInput], typing.Union[json_rpc10.output.OperationWithNestedStructureOutput, typing.Awaitable[json_rpc10.output.OperationWithNestedStructureOutput]]]]) -> None:
|
217 - | """
|
218 - | Method to register `operation_with_nested_structure` Python implementation inside the handlers map.
|
219 - | It can be used as a function decorator in Python.
|
220 - | """
|
221 - | ...
|
222 - |
|
223 - |
|
224 - | def operation_with_required_members(self, func: typing.Union[typing.Callable[[json_rpc10.input.OperationWithRequiredMembersInput, Ctx], typing.Union[json_rpc10.output.OperationWithRequiredMembersOutput, typing.Awaitable[json_rpc10.output.OperationWithRequiredMembersOutput]]], typing.Callable[[json_rpc10.input.OperationWithRequiredMembersInput], typing.Union[json_rpc10.output.OperationWithRequiredMembersOutput, typing.Awaitable[json_rpc10.output.OperationWithRequiredMembersOutput]]]]) -> None:
|
225 - | """
|
226 - | Method to register `operation_with_required_members` Python implementation inside the handlers map.
|
227 - | It can be used as a function decorator in Python.
|
228 - | """
|
229 - | ...
|
230 - |
|
231 - |
|
232 - | def put_with_content_encoding(self, func: typing.Union[typing.Callable[[json_rpc10.input.PutWithContentEncodingInput, Ctx], typing.Union[json_rpc10.output.PutWithContentEncodingOutput, typing.Awaitable[json_rpc10.output.PutWithContentEncodingOutput]]], typing.Callable[[json_rpc10.input.PutWithContentEncodingInput], typing.Union[json_rpc10.output.PutWithContentEncodingOutput, typing.Awaitable[json_rpc10.output.PutWithContentEncodingOutput]]]]) -> None:
|
233 - | """
|
234 - | Method to register `put_with_content_encoding` Python implementation inside the handlers map.
|
235 - | It can be used as a function decorator in Python.
|
236 - | """
|
237 - | ...
|
238 - |
|
239 - |
|
240 - | def run(self, address: typing.Optional[str] = ..., port: typing.Optional[int] = ..., backlog: typing.Optional[int] = ..., workers: typing.Optional[int] = ..., tls: typing.Optional[json_rpc10.tls.TlsConfig] = ...) -> None:
|
241 - | """
|
242 - | Main entrypoint: start the server on multiple workers.
|
243 - | """
|
244 - | ...
|
245 - |
|
246 - |
|
247 - | def run_lambda(self) -> None:
|
248 - | """
|
249 - | Lambda entrypoint: start the server on Lambda.
|
250 - | """
|
251 - | ...
|
252 - |
|
253 - |
|
254 - | def simple_scalar_properties(self, func: typing.Union[typing.Callable[[json_rpc10.input.SimpleScalarPropertiesInput, Ctx], typing.Union[json_rpc10.output.SimpleScalarPropertiesOutput, typing.Awaitable[json_rpc10.output.SimpleScalarPropertiesOutput]]], typing.Callable[[json_rpc10.input.SimpleScalarPropertiesInput], typing.Union[json_rpc10.output.SimpleScalarPropertiesOutput, typing.Awaitable[json_rpc10.output.SimpleScalarPropertiesOutput]]]]) -> None:
|
255 - | """
|
256 - | Method to register `simple_scalar_properties` Python implementation inside the handlers map.
|
257 - | It can be used as a function decorator in Python.
|
258 - | """
|
259 - | ...
|
260 - |
|
261 - |
|
262 - | def start_worker(self) -> None:
|
263 - | """
|
264 - | Build the service and start a single worker.
|
265 - | """
|
266 - | ...
|
267 - |
|
268 - |
|
269 - | def __init__(self) -> None:
|
270 - | ...
|
271 - |
|
272 - |
|
273 - | CODEGEN_VERSION: str = ...
|