0
# Utilities
1
2
Utility functions for application loading, file monitoring, address parsing, header processing, and other common server operations. These functions provide essential infrastructure for Hypercorn's operation and can be used for custom server implementations.
3
4
## Capabilities
5
6
### Application Loading and Wrapping
7
8
Functions for loading applications from module paths and wrapping them with appropriate adapters.
9
10
```python { .api }
11
def load_application(path: str, wsgi_max_body_size: int):
12
"""
13
Load application from module path.
14
15
Dynamically imports and loads an ASGI or WSGI application from
16
a module path specification. Handles both module and attribute
17
loading with proper error handling.
18
19
Args:
20
path: Module path in format "module:attribute" or "module"
21
Examples: "myapp:app", "myapp.wsgi:application"
22
wsgi_max_body_size: Maximum body size for WSGI applications
23
24
Returns:
25
Loaded application object (ASGI or WSGI)
26
27
Raises:
28
ImportError: If module cannot be imported
29
AttributeError: If attribute doesn't exist in module
30
NoAppError: If loaded object is not a valid application
31
"""
32
33
def wrap_app(app, wsgi_max_body_size: int, mode: str | None = None):
34
"""
35
Wrap application in appropriate wrapper.
36
37
Automatically detects application type (ASGI/WSGI) and wraps
38
it with the appropriate wrapper class for use with Hypercorn.
39
40
Args:
41
app: Application object to wrap
42
wsgi_max_body_size: Maximum body size for WSGI applications
43
mode: Optional mode specification ("asgi" or "wsgi")
44
If None, auto-detects based on application signature
45
46
Returns:
47
Wrapped application (ASGIWrapper or WSGIWrapper instance)
48
49
Raises:
50
ValueError: If mode is specified but doesn't match application
51
TypeError: If application is neither ASGI nor WSGI compatible
52
"""
53
54
def is_asgi(app) -> bool:
55
"""
56
Check if application is ASGI-compatible.
57
58
Examines application signature to determine if it follows
59
the ASGI interface specification.
60
61
Args:
62
app: Application object to check
63
64
Returns:
65
True if application is ASGI-compatible, False otherwise
66
67
Checks for:
68
- Callable with 3 parameters (scope, receive, send)
69
- Async function or method
70
- Proper ASGI signature patterns
71
"""
72
```
73
74
### File Monitoring
75
76
Functions for file watching and change detection, used for development auto-reload functionality.
77
78
```python { .api }
79
def files_to_watch() -> dict[Path, float]:
80
"""
81
Get list of files to watch for changes.
82
83
Discovers Python modules and other files that should be
84
monitored for changes to trigger server reloads during
85
development.
86
87
Returns:
88
Dictionary mapping file paths to modification times
89
90
Includes:
91
- All imported Python modules
92
- Configuration files
93
- Template files (if applicable)
94
- Static files (if applicable)
95
"""
96
97
def check_for_updates(files: dict[Path, float]) -> bool:
98
"""
99
Check if watched files have been modified.
100
101
Examines file modification times to determine if any
102
of the watched files have changed since last check.
103
104
Args:
105
files: Dictionary mapping file paths to last known modification times
106
107
Returns:
108
True if any files have been modified, False otherwise
109
110
Used by auto-reload functionality to trigger server
111
restarts when source code changes during development.
112
"""
113
```
114
115
### Process Management
116
117
Functions for process identification and management.
118
119
```python { .api }
120
def write_pid_file(pid_path: str):
121
"""
122
Write process PID to file.
123
124
Creates a PID file containing the current process ID,
125
used for process management and monitoring.
126
127
Args:
128
pid_path: Path where PID file should be written
129
130
Raises:
131
IOError: If PID file cannot be written
132
PermissionError: If insufficient permissions for file location
133
134
The PID file is typically used by process managers,
135
monitoring systems, and shutdown scripts.
136
"""
137
```
138
139
### Network Address Handling
140
141
Functions for parsing and formatting network socket addresses.
142
143
```python { .api }
144
def parse_socket_addr(family: int, address: tuple):
145
"""
146
Parse socket address based on family.
147
148
Parses socket addresses for different address families,
149
handling IPv4, IPv6, and Unix socket addresses.
150
151
Args:
152
family: Socket family (socket.AF_INET, socket.AF_INET6, socket.AF_UNIX)
153
address: Address tuple from socket operations
154
155
Returns:
156
Parsed address information (varies by family)
157
158
Address formats:
159
- AF_INET: (host, port)
160
- AF_INET6: (host, port, flowinfo, scopeid)
161
- AF_UNIX: path string
162
"""
163
164
def repr_socket_addr(family: int, address: tuple) -> str:
165
"""
166
Create string representation of socket address.
167
168
Formats socket addresses as human-readable strings
169
for logging and display purposes.
170
171
Args:
172
family: Socket family constant
173
address: Address tuple
174
175
Returns:
176
String representation of address
177
178
Examples:
179
- IPv4: "192.168.1.100:8000"
180
- IPv6: "[::1]:8000"
181
- Unix: "/tmp/socket"
182
"""
183
```
184
185
### HTTP Processing
186
187
Functions for HTTP header and request processing.
188
189
```python { .api }
190
def build_and_validate_headers(headers) -> list[tuple[bytes, bytes]]:
191
"""
192
Build and validate HTTP headers.
193
194
Processes and validates HTTP headers, ensuring proper
195
formatting and compliance with HTTP specifications.
196
197
Args:
198
headers: Headers in various formats (list, dict, etc.)
199
200
Returns:
201
List of (name, value) byte tuples
202
203
Raises:
204
ValueError: If headers are malformed
205
TypeError: If headers are wrong type
206
207
Performs validation:
208
- Header name format and characters
209
- Header value format and encoding
210
- Prohibited headers (e.g., connection-specific)
211
"""
212
213
def filter_pseudo_headers(headers: list[tuple[bytes, bytes]]) -> list[tuple[bytes, bytes]]:
214
"""
215
Filter out HTTP/2 pseudo-headers.
216
217
Removes HTTP/2 pseudo-headers (starting with ':') from
218
header list, typically when converting from HTTP/2 to HTTP/1.
219
220
Args:
221
headers: List of (name, value) header tuples
222
223
Returns:
224
Filtered headers list without pseudo-headers
225
226
HTTP/2 pseudo-headers include:
227
- :method, :path, :scheme, :authority (request)
228
- :status (response)
229
"""
230
231
def suppress_body(method: str, status_code: int) -> bool:
232
"""
233
Check if response body should be suppressed.
234
235
Determines whether an HTTP response should include a body
236
based on the request method and response status code.
237
238
Args:
239
method: HTTP request method (GET, HEAD, POST, etc.)
240
status_code: HTTP response status code
241
242
Returns:
243
True if body should be suppressed, False otherwise
244
245
Bodies are suppressed for:
246
- HEAD requests (always)
247
- 1xx informational responses
248
- 204 No Content responses
249
- 304 Not Modified responses
250
"""
251
```
252
253
### Server Validation
254
255
Functions for validating server configuration and requests.
256
257
```python { .api }
258
def valid_server_name(config: Config, request) -> bool:
259
"""
260
Validate server name against configuration.
261
262
Checks if the server name from request matches the
263
configured valid server names in the server configuration.
264
265
Args:
266
config: Server configuration object
267
request: HTTP request object containing Host header
268
269
Returns:
270
True if server name is valid, False otherwise
271
272
Validation includes:
273
- Exact hostname matches
274
- Wildcard pattern matching
275
- Port number handling
276
- Default server name fallback
277
"""
278
```
279
280
### Async Utilities
281
282
Async utility functions for server operation and shutdown handling.
283
284
```python { .api }
285
async def raise_shutdown(shutdown_event: Callable[..., Awaitable]) -> None:
286
"""
287
Raise shutdown signal by awaiting the shutdown event.
288
289
Waits for the shutdown event to complete and then raises
290
ShutdownError to trigger server shutdown sequence.
291
292
Args:
293
shutdown_event: Async callable that when awaited triggers shutdown
294
295
Raises:
296
ShutdownError: Always raised after shutdown event completes
297
"""
298
299
async def check_multiprocess_shutdown_event(
300
shutdown_event: EventType,
301
sleep: Callable[[float], Awaitable[Any]]
302
) -> None:
303
"""
304
Check for multiprocess shutdown events.
305
306
Periodically checks for shutdown events in multiprocess
307
environments, using the provided sleep function for timing.
308
309
Args:
310
shutdown_event: Multiprocess event object to check
311
sleep: Async sleep function for periodic checking
312
313
Used in multiprocess worker implementations to detect
314
when the master process signals shutdown.
315
"""
316
```
317
318
### Exception Classes
319
320
Utility-related exceptions for error handling.
321
322
```python { .api }
323
class NoAppError(Exception):
324
"""
325
Raised when application cannot be loaded.
326
327
This exception occurs when the application loading
328
process fails due to import errors, missing attributes,
329
or invalid application objects.
330
"""
331
332
class ShutdownError(Exception):
333
"""
334
Raised during server shutdown process.
335
336
This exception indicates errors that occur during
337
the server shutdown sequence, such as timeout issues
338
or resource cleanup failures.
339
"""
340
341
class LifespanTimeoutError(Exception):
342
"""
343
Raised when ASGI lifespan events timeout.
344
345
This exception occurs when ASGI application lifespan
346
startup or shutdown events take longer than the
347
configured timeout period.
348
"""
349
350
def __init__(self, stage: str) -> None:
351
"""
352
Initialize lifespan timeout error.
353
354
Args:
355
stage: Lifespan stage that timed out ("startup" or "shutdown")
356
"""
357
358
class LifespanFailureError(Exception):
359
"""
360
Raised when ASGI lifespan events fail.
361
362
This exception indicates that an ASGI application's
363
lifespan startup or shutdown event completed with
364
an error or unexpected state.
365
"""
366
367
def __init__(self, stage: str, message: str) -> None:
368
"""
369
Initialize lifespan failure error.
370
371
Args:
372
stage: Lifespan stage that failed ("startup" or "shutdown")
373
message: Error message describing the failure
374
"""
375
376
class UnexpectedMessageError(Exception):
377
"""
378
Raised for unexpected ASGI messages.
379
380
This exception occurs when an ASGI application sends
381
messages that don't conform to the expected protocol
382
sequence or message format.
383
"""
384
385
def __init__(self, state: Enum, message_type: str) -> None:
386
"""
387
Initialize unexpected message error.
388
389
Args:
390
state: Current protocol state when error occurred
391
message_type: Type of unexpected message received
392
"""
393
394
class FrameTooLargeError(Exception):
395
"""
396
Raised when protocol frame size exceeds limits.
397
398
This exception indicates that a protocol frame (HTTP/2,
399
WebSocket, etc.) exceeds the configured maximum size
400
limits, potentially indicating a malicious request.
401
"""
402
```
403
404
## Usage Examples
405
406
### Application Loading
407
408
```python
409
from hypercorn.utils import load_application, wrap_app
410
411
# Load application from module path
412
try:
413
app = load_application("myproject.wsgi:application", wsgi_max_body_size=16*1024*1024)
414
print(f"Loaded application: {app}")
415
except Exception as e:
416
print(f"Failed to load application: {e}")
417
418
# Wrap application automatically
419
wrapped_app = wrap_app(app, wsgi_max_body_size=16*1024*1024)
420
print(f"Wrapped application: {wrapped_app}")
421
422
# Check application type
423
from hypercorn.utils import is_asgi
424
if is_asgi(app):
425
print("Application is ASGI-compatible")
426
else:
427
print("Application is WSGI-compatible")
428
```
429
430
### File Monitoring for Development
431
432
```python
433
import time
434
from hypercorn.utils import files_to_watch, check_for_updates
435
436
# Get files to monitor
437
watch_files = files_to_watch()
438
print(f"Monitoring {len(watch_files)} files for changes")
439
440
# Check for changes periodically
441
while True:
442
if check_for_updates(watch_files):
443
print("Files changed - restarting server")
444
# Trigger server restart
445
break
446
time.sleep(1)
447
```
448
449
### Process Management
450
451
```python
452
import os
453
from hypercorn.utils import write_pid_file
454
455
# Write PID file for process management
456
try:
457
write_pid_file("/var/run/hypercorn.pid")
458
print(f"PID {os.getpid()} written to file")
459
except Exception as e:
460
print(f"Failed to write PID file: {e}")
461
```
462
463
### Address Parsing
464
465
```python
466
import socket
467
from hypercorn.utils import parse_socket_addr, repr_socket_addr
468
469
# Parse different address types
470
ipv4_addr = ("192.168.1.100", 8000)
471
ipv6_addr = ("::1", 8000, 0, 0)
472
473
parsed_ipv4 = parse_socket_addr(socket.AF_INET, ipv4_addr)
474
parsed_ipv6 = parse_socket_addr(socket.AF_INET6, ipv6_addr)
475
476
# Create string representations
477
ipv4_str = repr_socket_addr(socket.AF_INET, ipv4_addr)
478
ipv6_str = repr_socket_addr(socket.AF_INET6, ipv6_addr)
479
480
print(f"IPv4: {ipv4_str}") # "192.168.1.100:8000"
481
print(f"IPv6: {ipv6_str}") # "[::1]:8000"
482
```
483
484
### Header Processing
485
486
```python
487
from hypercorn.utils import build_and_validate_headers, filter_pseudo_headers
488
489
# Build headers from various formats
490
headers_dict = {"content-type": "text/html", "server": "hypercorn"}
491
headers_list = [("content-type", "text/html"), ("server", "hypercorn")]
492
493
validated_headers = build_and_validate_headers(headers_dict)
494
print(f"Validated headers: {validated_headers}")
495
496
# Filter HTTP/2 pseudo-headers
497
http2_headers = [
498
(b":method", b"GET"),
499
(b":path", b"/test"),
500
(b"user-agent", b"test-client"),
501
(b":authority", b"example.com")
502
]
503
504
filtered = filter_pseudo_headers(http2_headers)
505
print(f"Filtered headers: {filtered}") # Only user-agent remains
506
```
507
508
### Response Body Suppression
509
510
```python
511
from hypercorn.utils import suppress_body
512
513
# Check if body should be suppressed
514
cases = [
515
("HEAD", 200), # HEAD requests - suppress
516
("GET", 204), # No Content - suppress
517
("GET", 304), # Not Modified - suppress
518
("GET", 200), # Normal GET - don't suppress
519
("POST", 201), # Normal POST - don't suppress
520
]
521
522
for method, status in cases:
523
should_suppress = suppress_body(method, status)
524
print(f"{method} {status}: {'suppress' if should_suppress else 'include'} body")
525
```
526
527
### Server Name Validation
528
529
```python
530
from hypercorn.config import Config
531
from hypercorn.utils import valid_server_name
532
533
# Configure server names
534
config = Config()
535
config.server_names = ["example.com", "*.example.com", "api.service.local"]
536
537
# Mock request object (simplified)
538
class MockRequest:
539
def __init__(self, host):
540
self.headers = {"host": host}
541
542
# Validate different server names
543
test_hosts = ["example.com", "api.example.com", "invalid.com", "api.service.local"]
544
545
for host in test_hosts:
546
request = MockRequest(host)
547
is_valid = valid_server_name(config, request)
548
print(f"{host}: {'valid' if is_valid else 'invalid'}")
549
```