-
Notifications
You must be signed in to change notification settings - Fork 46
Expand file tree
/
Copy pathserver.py
More file actions
71 lines (53 loc) · 2 KB
/
Copy pathserver.py
File metadata and controls
71 lines (53 loc) · 2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
#!/usr/bin/env python3
"""A non-stoppable Python 3 HTTP server example."""
import json
import logging
import os
import signal
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
DEFAULT_PORT = 8080
class RequestHandler(BaseHTTPRequestHandler):
def _send(self, status: int, body: bytes, content_type: str = "text/plain") -> None:
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
if self.path == "/":
self._send(200, b"Hello, World!\n")
elif self.path == "/health":
self._send(200, json.dumps({"status": "ok"}, separators=(",", ":")).encode(), "application/json")
else:
self._send(404, b"Not Found\n")
def log_message(self, format: str, *args) -> None: # noqa: A002
logging.info("%s - %s", self.address_string(), format % args)
def _handle_signal(signum: int, frame) -> None:
signal_name = signal.Signals(signum).name
logging.info("Received %s; ignoring and continuing to serve.", signal_name)
def get_port() -> int:
raw = os.environ.get("PORT", str(DEFAULT_PORT))
try:
return int(raw)
except ValueError:
logging.warning("Invalid PORT value %r; falling back to %d.", raw, DEFAULT_PORT)
return DEFAULT_PORT
def main() -> int:
port = get_port()
server = HTTPServer(("0.0.0.0", port), RequestHandler)
signal.signal(signal.SIGTERM, _handle_signal)
signal.signal(signal.SIGINT, _handle_signal)
logging.info("Starting non-stoppable server on 0.0.0.0:%d", port)
try:
server.serve_forever()
except Exception as exc: # noqa: BLE001
logging.exception("Server failed: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())