Helper app to monitor what thread the app was called on. This can later be used to validate thread/event loop operations.
(scope: Scope, receive: Receive, send: Send)
| 110 | |
| 111 | |
| 112 | async def app_monitor_thread(scope: Scope, receive: Receive, send: Send) -> None: |
| 113 | class="st">"""Helper app to monitor what thread the app was called on. |
| 114 | |
| 115 | This can later be used to validate thread/event loop operations. |
| 116 | class="st">""" |
| 117 | request = Request(scope, receive) |
| 118 | |
| 119 | class="cm"># Make sure we parse the form |
| 120 | await request.form() |
| 121 | await request.close() |
| 122 | |
| 123 | class="cm"># Send back the current thread id |
| 124 | response = JSONResponse({class="st">"thread_ident": threading.current_thread().ident}) |
| 125 | await response(scope, receive, send) |
| 126 | |
| 127 | |
| 128 | def make_app_max_parts(max_files: int = 1000, max_fields: int = 1000, max_part_size: int = 1024 * 1024) -> ASGIApp: |
nothing calls this directly
no test coverage detected