I have wanted to demystify what goes behind the Python Flask framework. How does defining something as simple as handle HTTP Requests? How does create a server and maintain it? app.route app.run To demystify , I had two options: read Flask code end to end and understand or Reverse engineer flask by building one on my own. I chose the latter, and this blog is a step-by-step log of how it went. flask Side Note If you are new to Flask, then might be a good place to start. How to build your 1st flask app Reverse Engineering Reverse engineering started in my head. I am going to be working with just two files, and . ownflask.py demo.py Here is how a simple flask application would look # demo.py
from flask import Flask
​
app = Flask(__name__)
​
​
@app.route("/", methods=["GET", "POST"])
def hello():
    return "hello"
​
if __name__ == "__main__":
    app.run() Looking at this sample snippet, I want to mimic the same interface. Take a pass from top to bottom and see what all we need We need a class which initializes an object Flask app The class has a method And it starts a server Flask run The class also has a method that registers the endpoints Flask route Let's lay them down #ownflask.py
​​
class Flask:
    def __init__(self, name):
        self.name = name
    
    def run(self):
        pass
        
    def route(self, path, methods):
        def wrapper(f):
            pass
        return wrapper That's gives us the basic skeleton. Let's add the functionality one by own. Python module provides a let's use that. http HTTPServer Starting a Server In Flask, is responsible for starting a development webserver. The server then listens to all HTTP requests and responds to them. app.run #ownflask.py
​
from http.server import HTTPServer, BaseHTTPRequestHandler
​
class Flask:
    ...
    
    def run(self, server_class=HTTPServer, handler_class=BaseHTTPRequestHandler, port=8000):
        server_address = ('', port)
        print (f"Running server in port {port}")
        httpd = server_class(server_address, handler_class)
        httpd.serve_forever() In , change to to work with the module, we just created and run . On hitting the You get a 501 error from the browser since we haven't implemented anything to handle the incoming request. demo.py from flask from ownflask demo.py http://127.0.0.1:8000 Mapping Requests The method in Flask registers an endpoint. When an HTTP request comes, it maps it to the associated function call. These routes are maintained in a global object so that the request handler can refer to it. For our , let's use a global dictionary. app.route ownflask Here I have two methods one to record to its associated functions and to associate endpoints and its HTTPMethods. routes route_methods routes = {}
route_methods = {}
​
class Flask:
    ...
    def route(self, path, methods):
        def wrapper(f):
            routes[path] = f
            route_methods[path] = methods
        return wrapper Handling GET Request When running our server, we have used a . From the Python documentation, it is clear that we have to extend it to support handling requests. BaseHTTPRequestHandler By itself, it cannot respond to any actual HTTP requests; it must be subclassed to handle each request method (e.g., GET or POST). class RequestHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(str.encode("Handling GET"))
    
    def do_POST(self):
        pass The above snippet sends as a response despite what the route function returns. Let's change that. Handling GET returns that is the URL, mapping that with routes dict, we can call the respective function. dir(self) self.path class RequestHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        resp = routes[self.path]()
        ...
        ...
        self.wfile.write(str.encode(resp)) Handling URL Params Flask is known for passing URL params as a part of a URL string or a query string. /book/<int:id>
/book?id=10 The 1st one would require some form of regex in the routes and the way we store them. Let's handle them later. Let's handle with the name hello world http://127.0.0.1:8000?name=Joe The current code fails with a since the query string is also a part of the route. KeyError KeyError: '/?name=jpe' To parse this and separate the URL path and the query params, we will use urllib import urllib.parse as urlparse
​
class RequestHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        path = urlparse.urlparse(self.path).path
        qs = urlparse.parse_qs(urlparse.urlparse(self.path).query)
        resp = routes[path]()
        
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(str.encode(resp)) In Flask, the routing function can access the request params via the global object. In our case, for the route to access query params, we need the means to pass it to them. Request hello Request Class class Request:
    def __init__(self, request, method):
        self.request = request
        self.method = method
        self.path = urlparse.urlparse(request.path).path
        self.qs = urlparse.parse_qs(urlparse.urlparse(request.path).query)
        self.headers = request.headers Let's pass this Request object to the route. class RequestHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        request = Request(self, "GET")
        resp = routes[request.path](request)
        
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(str.encode(resp)) With the current state, let's capture hello() takes 0 positional arguments but 1 was given request @app.route("/")
def hello(request):
    return f"hello {request.qs["name"][0]}" Handling JSON Response If we modify the endpoint to return a instead of , we will receive an error. hello dict str the descriptor 'encode' for 'str' objects doesn't apply to a 'dict' object. It happens because we convert dict to a bytes object. To do this, we should convert the response dict to and then encode it. str def do_GET(self):
    request = Request(self, "GET")
    resp = routes[request.path](request)
    if isinstance(resp, dict):
        resp = json.dumps(resp) Handling POST Request For handling POST requests, you need to access the request body along with other parameters. Let's update the request class to support the same. class Request:
    def __init__(self, request, method):
        ...
        ...
        self.content_length = int(self.headers.get('content-length', 0))
        self.body = request.rfile.read(self.content_length)
        try:
            self.json = json.loads(self.body)
        except json.decoder.JSONDecodeError: 
            self.json = {} Let's consume the same via a POST API @app.route("/todo", methods=["POST"])
def todo(request):
    return {"status": "success", "data": request.json} Handling Unsupported Request Right now, if you hit from the browser, you will get the response. This is wrong since we have clearly defined that on supports post request. This is where comes in really handy. /todo /todo route_methods def do_GET(self):
    ...
    if "GET" not in route_methods[request.path]:
        self.send_response(401)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(str.encode(f"{request.path} {request.method} not supported"))
        return Looks like we are repeating ourselves a lot; let's move them to a common function class RequestHandler(SimpleHTTPRequestHandler):
    ...
    ...
    def write_response(self, response, status_code):
        self.send_response(status_code)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        if isinstance(response, dict):
            response = json.dumps(response)
        self.wfile.write(str.encode(response)) The final and method looks like this. do_GET do_POST def do_GET(self):
        request = Request(self, "GET")
        if "GET" not in route_methods[request.path]:
            self.write_response("Method not supported", 401)
            return
        resp = routes[request.path](request)
        self.write_response(resp, 200)
    
    def do_POST(self):
        request = Request(self, "POST")
        if "POST" not in route_methods[request.path]:
            self.write_response("Method not supported", 401)
            return
        resp = routes[request.path](request)
        self.write_response(resp, 200) We can further refactor them into def not_found(self, request):
        return self.write_response(f"{request.path} 404 NOT FOUND", 404)
​
    def method_not_supported(self, request):
        return self.write_response(f"{request.path} {request.method} not supported", 401)
        
    def process_request(self, request):
        if request.path not in routes:
            return self.not_found(request)
        if request.method in route_methods[request.path]:
            return self.method_not_supported(request)
        
        resp = routes[request.path](request)
        self.write_response(resp)
    
            
    def do_GET(self):
        request = Request(self, method='GET')
        return self.process_request(request)
​
    def do_POST(self):
        request = Request(self, method='POST')
        return self.process_request(request) Introducing Multithreading At this point, if you write a small multithreading script and hit our server, it will hang because is not designed to handle multiple requests. Replacing it with HTTPServer ThreadingHTTPServer. from http.server import ThreadingHTTPServer
​
class Flask:
    ...
    def run(self, name):
        ...
        ...
        self.server = WSGIServer((self.host, self.port), ThreadingHTTPServer) WSGI vs. HTTP At this point, I was happy with what I had accomplished and had already posted a , and nudged me in the direction to explore . tweet ArunMozhi WSGIServer from wsgiref.simple_server import WSGIServer
​
class Flask:
    ...
    def run(self, name):
        ...
        ...
        self.server = WSGIServer((self.host, self.port), HttpReqHandler) What started as an experiment to Demyistify flask and understood it better got me into a rabbit hole of new questions. How is different from the interface look the same? WSGIServer HTTPServer How can we plug the to work with Gunicorn ownflask How to add async to ownflask? Going one step further, How does Gunicorn work? What are my unknown unknowns? If you know the answer to any of these, you can send them to me via . Twitter Also published . here

Twitter

Caches in Python

Read My Stories

Too Long; Didn't Read

Make resilience your competitive advantage

Writing Your Own Flask(like) Framework

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

Caches in Python

10 Ways Stand Out as a Java Developer and Land that Dream Job

10 Fantastic JavaScript Projects for Beginners

11 Great Tips From A Guy Who Leveled Up From Intern To Dev

125 Stories To Learn How To Do X

Caches in Python

10 Ways Stand Out as a Java Developer and Land that Dream Job

10 Fantastic JavaScript Projects for Beginners

11 Great Tips From A Guy Who Leveled Up From Intern To Dev

125 Stories To Learn How To Do X

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps