danilobellini/example.py

## example.py
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()

## server.py
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,
        )
	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()
	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,
	)