Created
May 13, 2018 18:08
-
-
Save danilobellini/5077f44df476d10f521a9f4a20e8e0a3 to your computer and use it in GitHub Desktop.
A really simple/incomplete bottle/flask-like "single blocking worker thread" web server implementation (Python 2)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from server import WebServer | |
| app = WebServer() | |
| @app.route("/") | |
| def root(): | |
| return "Hello World!" | |
| @app.route("/file") | |
| def file_route(): | |
| with open("server.py", "rb") as f: | |
| return f.read() | |
| @app.route("/img") | |
| def img_route(): | |
| with open("Free-stock-image.jpg", "rb") as f: | |
| return f.read(), {"Content-Type": "image/jpeg"} | |
| @app.route("/html") | |
| def html_route(): | |
| return "<html><body><h1>Image:</h1><img src="/img"></body></html>" | |
| if __name__ == "__main__": | |
| app.loop() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import socket, collections | |
| from itertools import chain | |
| import colorama | |
| colorama.init() | |
| def create_raw_response(protocol, status, headers, payload): | |
| protocol_line = protocol + " " + str(status) | |
| header_lines = (k + ": " + v for k, v in headers.items()) | |
| # An empty line separates the payload (body) from the header | |
| return "\r\n".join(chain([protocol_line], header_lines, ["", payload])) | |
| def create_response(status, headers, payload): | |
| return create_raw_response( | |
| protocol="HTTP/1.1", | |
| status=status, | |
| headers=dict({ | |
| "Content-Type": "text/html", | |
| "Content-Length": str(len(payload)) | |
| }, **headers), | |
| payload=payload, | |
| ) | |
| def cprint(color, content): | |
| print(getattr(colorama.Fore, color.upper()) | |
| + content + colorama.Style.RESET_ALL) | |
| class HTTPError: pass | |
| class NotFound(HTTPError): status = 404 | |
| class BadRequest(HTTPError): status = 400 | |
| class InternalServerError(HTTPError): status = 500 | |
| Request = collections.namedtuple("ParsedRequest", | |
| ["method", "route", "protocol", "headers", "payload"]) | |
| def parse_request(raw_data): | |
| try: | |
| protocol_line_break = raw_data.find("\r\n") | |
| header_line_break = raw_data.find("\r\n\r\n") | |
| method, route, protocol = raw_data[:protocol_line_break].split() | |
| headers = dict( | |
| tuple(el.strip() for el in line.split(":", 1)) | |
| for line in raw_data[protocol_line_break + 2:header_line_break] | |
| .splitlines() | |
| ) | |
| payload = raw_data[header_line_break + 4:] | |
| return Request(method, route, protocol, headers, payload) | |
| except: | |
| raise BadRequest | |
| class WebServer: | |
| def __init__(self): | |
| self.routes = collections.OrderedDict() | |
| def loop(self, host="0.0.0.0", port=8088): | |
| s = socket.socket( | |
| socket.AF_INET, # Socket family | |
| socket.SOCK_STREAM, # TCP (acknowledges the packages sent) | |
| ) | |
| s.setsockopt( # This fix a Windows-specific non-compliance bug | |
| socket.SOL_SOCKET, | |
| socket.SO_REUSEADDR, | |
| 1, | |
| ) | |
| s.bind((host, port)) | |
| s.listen(10) | |
| while True: | |
| conn, addr = s.accept() | |
| data = conn.recv(1024) | |
| cprint("cyan", data) | |
| try: | |
| raw_response = self.response_of(parse_request(data)) | |
| except HTTPError as exc: | |
| raw_response = "HTTP/1.1 " + str(exc.status) | |
| except: | |
| raw_response = "HTTP/1.1 500" | |
| finally: | |
| conn.sendall(raw_response) | |
| conn.close() | |
| cprint("green", raw_response) | |
| def route(self, route, methods=["GET"]): | |
| def decorator(func): | |
| for method in methods: | |
| self.routes[method, route] = func | |
| return func | |
| return decorator | |
| def response_of(self, request): | |
| try: | |
| func = self.routes[request.method, request.route] | |
| except KeyError: | |
| raise NotFound | |
| result = func() | |
| if isinstance(result, tuple): | |
| result, extra_headers = result | |
| else: | |
| extra_headers = {} | |
| return create_response( | |
| status=200, | |
| headers=extra_headers, | |
| payload=result, | |
| ) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment