HTTPServer Basic Example

server.py

"""HTTPServer Basic Example
$ python3 server.py
# HEAD
$ curl -I http://localhost:3000
# GET
$ curl http://localhost:3000/app?id=123
# POST
$ curl http://localhost:3000 -F name=curl_form -F user=n-bar -F [email protected]
"""

import threading
from urllib.parse import urlparse
from socketserver import ThreadingMixIn
from http.server import HTTPServer, BaseHTTPRequestHandler
from cgi import FieldStorage

SERVED_HOST = "localhost"
SERVED_PORT = 3000
SERVED_ADDR = SERVED_HOST, SERVED_PORT


class TestEnvHandler(BaseHTTPRequestHandler):
    def message(self):
        parsed_path = urlparse(self.path)
        message_parts = [
            #  Parts to concatenate
            "CLIENT VALUES:",
            "clent_address=%s (%s)" %
            (self.client_address, self.address_string()),
            'command=%s' % self.command,
            'path=%s' % self.path,
            'real path=%s' % parsed_path.path,
            'query=%s' % parsed_path.query,
            'request_version=%s' % self.request_version,
            '',
            'SERVER VALUES:',
            'server_version=%s' % self.server_version,
            'sys_version=%s' % self.sys_version,
            'protocol_version=%s' % self.protocol_version,
            'thread#=%s' % threading.currentThread().getName(),
            'active_threads=%s' % threading.activeCount(),
            '',
            'HEADERS RECEIVED:'
        ]
        for name, value in sorted(self.headers.items()):
            message_parts.append('%s=%s' % (name, value.rstrip()))
        message_parts.append('')
        return '\n'.join(message_parts).encode('utf-8')

    def do_HEAD(self):
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.end_headers()

    def do_GET(self):
        self.do_HEAD()
        self.wfile.write(self.message())

    def do_POST(self):
        self.do_HEAD()
        form = FieldStorage(
            fp=self.rfile,
            environ={
                'REQUEST_METHOD': 'POST',
                'CONTENT_TYPE': self.headers['Content-Type']
            },
            headers=self.headers)

        for field in form.keys():
            form_item = form[field]
            if form_item.filename:
                file_data = form_item.file.read()
                file_size = len(file_data)  # bytes
                file_head = file_data.decode('utf-8')[:25]
                del file_data
                self.wfile.write(b'\tUploaded %s as "%s" (%d bytes)\n' % (
                    field.encode('utf-8'), form_item.filename.encode('utf-8'),
                    file_size))
                self.wfile.write(b'\tExample Data: %s\n' %
                                 file_head.encode('utf-8'))
            else:
                self.wfile.write(
                    b'\t%s=%s\n' %
                    (field.encode('utf-8'), form_item.value.encode('utf-8')))


class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
    pass


if __name__ == "__main__":
    httpd = ThreadedHTTPServer(SERVED_ADDR, TestEnvHandler)
    try:
        print("Server Starts - ", SERVED_ADDR)
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        httpd.server_close()
        print("Server Stops - ", SERVED_ADDR)