from nbdev import *
from fastcore.utils import *
import time,socket

Python's standard library servers

Python's socketserver classes call handle in a BaseRequestHandler subclass that you pass in to its constructor. You use a BaseRequestHandler with any of the server classes/mixins provided by socketserver.

class ReuseThreadingServer[source]

ReuseThreadingServer(server_address, RequestHandlerClass, bind_and_activate=True) :: ThreadingTCPServer

Mix-in class to handle each request in a new thread.

It's easiest to use allow_reuse_address to avoid having to wait for sockets to close, especially when testing. This class adds that functionality to ThreadingTCPServer.

Here's an example of using Python's standard library features along with ReuseThreadingServer:

host,port = 'localhost',8000

class _TestHandler(StreamRequestHandler):
    def handle(self):
        print('received', self.rfile.readline())
        self.wfile.write(bytes(f'pong {self.client_address[0]}\r\n', 'utf8'))

@startthread
def _f():
    with ReuseThreadingServer((host,port), _TestHandler) as srv: srv.handle_request()

time.sleep(0.5) # wait for server to start

c = start_client(port,host)
c.send(b'ping\r\n')
c.recv(1024)

received b'ping\r\n'

b'pong 127.0.0.1\r\n'

class HandlerException[source]

HandlerException(code, err='') :: Exception

Class for exceptions from setup of MinimalHTTPHandler

class MinimalHTTPHandler[source]

MinimalHTTPHandler(request, client_address, server) :: StreamRequestHandler

A greatly simplified version of BaseHTTPHandler. Overriding handle is required.

MinimalHTTPHandler parses the HTTP command and headers, and sets command, path, request_version, and headers. It is based on the code in Python's BaseHTTPHandler, but is greatly simplified, and made consistent with the other socketserver servers.

To send a response, call send_response(code), optionally send_header a few times, then end_headers, and finally write to wfile. For instance:

class _TestHandler(MinimalHTTPHandler):
    def handle(self):
        print(f'Command/path/version: {self.command} {self.path} {self.request_version}')
        print(self.headers)
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header('Content-Length', '2')
        self.end_headers()
        self.wfile.write(b'ok')

@startthread
def _f():
    with ReuseThreadingServer(('localhost',8000), _TestHandler) as httpd: httpd.handle_request()

time.sleep(0.5) # wait for server to start
test_eq(urlread("http://localhost:8000"), b'ok')

Command/path/version: GET / 1.1
Accept-Encoding: identity
Host: localhost:8000
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.183 Safari/537.36
Connection: close