Skip to content

Instantly share code, notes, and snippets.

@bergwerf
Created May 31, 2023 16:39
Show Gist options
  • Save bergwerf/c06bddc015c911de743fe8d94930f6ad to your computer and use it in GitHub Desktop.
Save bergwerf/c06bddc015c911de743fe8d94930f6ad to your computer and use it in GitHub Desktop.
Minimalistic server
import io
import json
import http.server
import socketserver
HOST = "localhost"
PORT = 8080
class AnalyzerHandler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
# Note that read() without content length may block indefinitely, because
# not all clients send an EOF character.
byte_data = self.rfile.read(int(self.headers.get("Content-Length")))
data = json.loads(byte_data.decode("utf-8"))
# Write headers.
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
# Write output data. The option write_through=True prevents TextIOWrapper
# from buffering internally, so that all data has been written to wfile when
# it is detached.
output = io.TextIOWrapper(self.wfile, encoding="utf-8", write_through=True)
json.dump({"status": 200, "data": data}, output)
# Detach so wfile won't be closed when output is cleaned up.
# Note that wfile is flushed by BaseHTTPRequestHandler.
output.detach()
# Allow immediate port re-use after termination.
socketserver.TCPServer.allow_reuse_address = True
# Open a TCP server with a handler derived from BaseHTTPRequestHandler.
with socketserver.TCPServer((HOST, PORT), AnalyzerHandler) as server:
try:
server.serve_forever()
except KeyboardInterrupt:
server.server_close()
@bergwerf
Copy link
Author

I tried to implement streamed reading and writing of JSON data, but I only managed to get streamed writing using the built-in json.dump and an io.TextIOWrapper.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment