Simple HTTP REST server in python3

simple_server.py

#!/usr/bin/env python

"""A simple HTTP server with REST and json for python 3.

addrecord takes utf8-encoded URL parameters
getrecord returns utf8-encoded json.
"""

import argparse
import json
import re
import threading
from email.message import EmailMessage
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib import parse


def _parse_header(content_type):
    m = EmailMessage()
    m["content-type"] = content_type
    return m.get_content_type(), m["content-type"].params


class LocalData(object):
    records = {}


class HTTPRequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if re.search("/api/v1/addrecord/*", self.path):
            ctype, pdict = _parse_header(self.headers.get("content-type"))
            if ctype == "application/json":
                length = int(self.headers.get("content-length"))
                rfile_str = self.rfile.read(length).decode("utf8")
                data = parse.parse_qs(rfile_str, keep_blank_values=True)
                record_id = self.path.split("/")[-1]
                LocalData.records[record_id] = data
                print("addrecord %s: %s" % (record_id, data))
                self.send_response(HTTPStatus.OK)
            else:
                self.send_response(
                    HTTPStatus.BAD_REQUEST, "Bad Request: must give data"
                )
        else:
            self.send_response(HTTPStatus.FORBIDDEN)

        self.end_headers()

    def do_GET(self):
        if re.search("/api/v1/shutdown", self.path):
            # Must shutdown in another thread or we'll hang
            def kill_me_please():
                self.server.shutdown()

            threading.Thread(target=kill_me_please).start()

            # Send out a 200 before we go
            self.send_response(HTTPStatus.OK)
        elif re.search("/api/v1/getrecord/*", self.path):
            record_id = self.path.split("/")[-1]
            if record_id in LocalData.records:
                self.send_response(HTTPStatus.OK)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                # Return json, even though it came in as POST URL params
                data = json.dumps(LocalData.records[record_id])
                print("getrecord %s: %s" % (record_id, data))
                self.wfile.write(data.encode("utf8"))
            else:
                self.send_response(
                    HTTPStatus.NOT_FOUND, "Not Found: record does not exist"
                )
        else:
            self.send_response(HTTPStatus.BAD_REQUEST)

        self.end_headers()


def main():
    parser = argparse.ArgumentParser(description="HTTP Server")
    parser.add_argument("port", type=int, help="Listening port for HTTP Server")
    parser.add_argument("ip", help="HTTP Server IP")
    args = parser.parse_args()

    server = HTTPServer((args.ip, args.port), HTTPRequestHandler)
    print("HTTP Server Running...........")
    server.serve_forever()


if __name__ == "__main__":
    main()

Based on https://gist.github.com/mafayaz/faf938a896357c3a4c9d6da27edcff08, but adapted for Python 3.

Start server with:

 ./simple_server.py 7000 127.0.0.1

Example client calls and responses using httpie:

$ http -f POST http://127.0.0.1:7000/api/v1/addrecord/one \
  Content-Type:application/json var1=value1 var2=value2
HTTP/1.0 200 OK
Date: Sun, 13 Oct 2019 18:02:15 GMT
Server: BaseHTTP/0.6 Python/3.7.4

$ http http://127.0.0.1:7000/api/v1/getrecord/one
HTTP/1.0 200 OK
Content-Type: application/json
Date: Sun, 13 Oct 2019 18:14:09 GMT
Server: BaseHTTP/0.6 Python/3.7.4

{
    "var1": [
        "value1"
    ],
    "var2": [
        "value2"
    ]
}

$ http http://127.0.0.1:7000/api/v1/shutdown
HTTP/1.0 200 OK
Date: Sun, 13 Oct 2019 18:02:22 GMT
Server: BaseHTTP/0.6 Python/3.7.4

Thank you very much for your effort 👍

@loomsen Sure. I ended up moving to flask. I find it more straightforward. See https://gist.github.com/dfrankow/14fa994d7edcbe2e88f54823b90b41a3.

thanks.veryuseful!

Great! Thanks for letting me know.

Again, just FYI, I decided I liked flask better. See link above.

cgi is deprecated and slated for removal in 3.13

Here's a replacement for cgi.parse_header

from email.message import EmailMessage

def parse_header(content_type):
    m = EmailMessage()
    m['content-type'] = content_type
    return m.get_content_type(), m['content-type'].params

Thanks @iwconfig, I added your changes. I'd still likely use flask. :)

Yeah, I would too. :)

Hello dfrankow. Maybe it would be better to use codes from HTTPStaus python module? For example here. It may go like this:

from http import HTTPStatus
...
    self.send_response(HTTPStatus.OK)

Also, in this case, your code will not require a comments such as this
What do you think?

Hello dfrankow. Maybe it would be better to use codes from HTTPStatus python module?

Good idea. Done.