kaychen 0.1.3

This is a web framework build while completing the course https://testdriven.io/courses/python-web-framework/

Table of Contents

1. Introduction

2. Table of contents

3. Part 1

   1. WSGI
      1. What is WSGI
      2. Application side
   2. Routing
   3. Unit test and test client
   4. Templates
   5. Static Files
   6. Middleware
      1. The middleware class, base functionality
      2. the convoluted part
      3. static files
   7. allowing methods
   8. Custom Responses
   9. Pypi
   10. example web app
   11. Deploying to Heroku
       1. workflow
       2. other heroku commands

4. Part 2 - ORM

   1. Design
      1. Connection
      2. table definition
      3. creating tables
      4. inserting data
      5. fetch all data
      6. query
      7. save object with foreign key reference
      8. fetch object with foreign key reference
      9. update an object
      10. delete an object
   2. Implementing the Database, Tables, Columns and ForeignKeys

   

Introduction

This Repo follows the course “Building your own Python Framework” over at testdriven.io. Over this course we learn about WSGI , how frameworks like Django and Flask implement their route functionality and other features like

• templates
• exception handling
• middleware
• allowing methods

and additionally about building your own ORM and Deployment.

Table of contents

Table of Contents

1. Introduction
2. Table of contents
3. Part 1
   1. WSGI
   2. Routing
   3. Unit test and test client
   4. Templates
   5. Static Files
   6. Middleware
   7. allowing methods
   8. Custom Responses
   9. Pypi
   10. example web app
   11. Deploying to Heroku
4. Part 2 - ORM
   1. Design
   2. Implementing the Database, Tables, Columns and ForeignKeys

Part 1

WSGI

What is WSGI

WSGI (Web Server Gateway Interface) is a proposed standard as of PEP333 of how a Web Server should talk to a python web appilcation.

This gives way for a unified way of talking to python web applications for web servers, which in turn permits to deploy python web applications in a standardized way.

Application side

On the application side, we have the application object, which shall be callable and take 2 positional arguments

def simple_app(environ, start_response):

It shall return an iterable yielding zero or more strings

This application can then be served e.g. with gunicorn or for development purposes with wsgiref.simple_server

from wsgiref.simple_server import make_server

server = make_server('localhost', 8000, app=simple_app)
server.serve_forever()

Routing

To acheive Decorator like registering of routes like in Flask or injection-like registering like in Django, one needs to implement a method on its application object for registering the routes. The application can make use of the Parse library to easily retrieve the routes via route-patterns

def add_route(self, path, handler):
    assert path not in self.routes, "Such route already exists"
    self.routes[path] = handler

def find_handler(self, request_path: str):
    for path, handler in self.routes.items():
        res = parse(path, request_path)

For easier and more intuitive handling of environ and start_response one can use webob Request and Response objects.

Unit test and test client

Using unit test one can verify the base functionality. For extending the functionality like default-responses, templates, exception handlers and static files, we write the tests first, see them fail and add the functionality itself, followed by refactoring.

To test the app in an fast, isolated and repeatable way, one would need a test_client to call the api without spinning it up with a web server each time. This can be acheive using the request-wsgi-adapter.

def test_session(self, base_url="http://testserver"):
    session = RequestsSession()
    session.mount(prefix=base_url, adapter=RequestsWSGIAdapter(self))
    return session

Templates

Templates are as easy as providing the templates_dir on app initialization and using it inside the route

def __init__(self, templates_dir="templates"):
    self.routes = {}
    self.templates_env = Environment(
        loader=FileSystemLoader(os.path.abspath(templates_dir))
    )

def template(self, template_name: str, context: dict):
    return self.templates_env.get_template(template_name).render(context)

@app.route("/html")
def html_handler(req, resp):
    resp.body = app.template(
        "home.html", context={"title": "Some Title", "name": "Some Name"}
    ).encode()

Static Files

To use static files we make use of the package Whitenoise. Whitenoise wraps a wsgi-application and provides it with static files. Since a wsgi application is just a callable with a specific function signature, we can wrap whatever we had inside the __call__ method of our API class, and call that with whitenoise.

def __init__(self, templates_dir="templates", static_dir="static"):
    self.whitenoise = WhiteNoise(self.wsgi_app, root=static_dir)
    ...
def __call__(self, environ, start_response):
    return self.whitenoise(environ, start_response)

Middleware

The middleware class, base functionality

To use middleware, we write a Class Middleware. It defines two methods to process request and response: process_request and process_response. These functions do nothing on the base class, but can be overwritten when creating a child.

When handling requests, it first calls process_request, then the handler of the app, then the process_response, before returning the response.

class Middleware:
    ...
    def handle_request(self, request):
        self.process_request(request)
        response = self.app.handle_request(request)
        self.process_response(request)
        return response

Since each middleware serves as the Server-side implementation of the WSGI protocol for the application that gets called after it, it needs to be callable in the WSGI sense.

class Middleware:
    ...
    def __call__(self, environ, start_response):
        request = Request(environ)
        response = Response(self.handle_request)
        return response(environ, start_response)

The wsgi logic of using environ and start_response is hidden in the behavior of the webob objects Request and Response.

the convoluted part

Furthermore, to add another middleware to the middleware stack, one wraps a given middleware aroung the app.

class Middleware:
    ...
    def add(mid: Middleware):
        self.app = mid(self.app)

We can then apply the same logic on our framework api, by initialising a base middleware with our app, and calling the middleware when handling requests

class API:
    def __init__(self, templates_dir="templates", static_dir="static"):
        ...
        self.mid = Middleware(self)

    ...

    def add(mid: Middleware):
        self.app = mid(self.app)

    ...

    def __call__(self, environ, start_response):
        self.middleware(environ, start_response)

static files

This would unable our handling of static files. Therefore we oblige to be the static files being served on route, which root is /static

def __call__(self, environ, start_response):
    path_info = environ["PATH_INFO"]
    if path_info.startswith("/static"):
        environ["PATH_INFO"] = path_info[len("/static") :]
        return self.whitenoise(environ, start_response)

    return self.middleware(environ, start_response)

allowing methods

Adding allowed methods to all our ways of adding routes, requires us to change our data structure a little bit. From

self.routes[path] = handler

to

self.routes[path] = {"handler": handler, "allowed_methods": allowed_methods}

Which we then can exploit when we’re handling the request

...
handler_data, kwargs = self.find_handler(request.path)
try:
    if handler_data is not None:
        if request.method.lower() not in handler_data["allowed_methods"]:
            raise AttributeError("Method not allowed", request.method)

        handler = handler_data["handler"]
        if inspect.isclass(handler):
            handler = getattr(handler(), request.method.lower(), None)
            if handler is None:
                raise AttributeError("Method not allowed", request.method)
            handler(request, response, **kwargs)
        handler(request, response, **kwargs)
...

Custom Responses

Next we make it possible to respond with json, html or plain text. Therefore one may implement a Custom Response that makes use of the Webob Response object. The user has access to that response object via the handler (as before).

@app.route("/home")
def html(req, resp):
    resp.json = {"name": "kaychen"}

When the framework sends back the response, as in

def handle_request(self, request):
    response = CustomResponse
    ...
    return response()

the response call method is executed. This is where the logic is applied then

from webob import Response

def CustomResponse:
    self.json = None
    self.status_code = 200
    ...                         # setting of other variables

   def __call__(self):
       self.set_body_and_content_type()
       response = Response(
           body=self.body, content_type=self.content_type, status=f"{self.status_code}"
       )
       return response(environ, start_response)

    def set_body_and_content_type(self):
        if self.json is not None:
            self.body = json.dumps(self.json).encode("UTF-8")
            self.content_type = "application/json"
        ...                     # more handling of html and text

Pypi

Next we publish the package to Pypi using setup.py (for humans). A few things to keep in mind

• find_packages used in setup.py, therefore need to have __init__.py so it finds the package
• when using the package in combination with gunicorn, one still needs to install gunicorn inside the virtualenv
• need to create directories (/static, /templates)

example web app

To see the framework in action we build an example application: kaychen-web-app

Deploying to Heroku

workflow

1. Define Procfile
2. heroku create
   • git remote is create alongside the app on heroku account
   • deplying via git push
3. git push heroku main
4. Check if application is deployed: heroku ps:scale web=1
5. View logs: heroku logs --tail

other heroku commands

1. Scaling = number of running dynos (lightweight container) heroku ps:scale web={number_of_dynos}

Part 2 - ORM

ORMs allow you to

1. interact wiht db in own language of choice
2. abstract away the database (easy switching)
3. Usually written by SQL experts for performance reasons

Design

Connection

from kaychen import Database

db = Database("./test.db")

table definition

from kaychen import Table, Column, ForeignKey

class Author(Table):
    name = Column(str)
    age = Column(int)

class Book(Table):
    title = Column(str)
    published = Column(bool)
    author = ForeignKey(Author)

creating tables

db.create(Author)
db.create(Book)

inserting data

kay = Author("Kay", age=12)
db.insert(kay)

fetch all data

authors = db.all(Author)

query

author = db.query(Author, 47)

save object with foreign key reference

book = Book(title="Building an ORM", published=True, author=greg)
db.save(book)

fetch object with foreign key reference

print(Book.get(55).author.name)

update an object

book.title = "How to build an ORM"
db.update(book)

delete an object

db.delete(Book, id=book.id)

Implementing the Database, Tables, Columns and ForeignKeys

The database holds primarily a database connection and has the ability to create new tables. Furthermore it has the ability to print the tables. Create and print tables are wrappers for executing sql commands.

class Database:
    def __init__(self, path: str):
        self.conn = sqlite3.Connection(path)

    def create(self, table: type[Table]):
        self.conn.execute(table._get_create_sql())

    @property
    def tables(self) -> list[type[Table]]:
        SELECT_TABLES_SQL = "SELECT name FROM sqlite_master WHERE type = 'table';"
        return [x[0] for x in self.conn.execute(SELECT_TABLES_SQL).fetchall()]

It is only the database that executes sql commands via its db connection. Other objects may provide the Database with how it should query for them, e.g. Table.

class Table:
    ...
    @classmethod
    def _get_create_sql(cls):
        CREATE_TABLE_SQL = "CREATE TABLE IF NOT EXISTS {name} ({fields});"
        ...

New models inherit from the table class and set Columns as their class variables.

class Author(Table):
    name = Column(str)
    age = Column(int)

Columns hold information about the type of the attributes that a certain table, e.g. Author, holds. It provides methods to translate those types to SQL-types.

class Column:
    def __init__(self, column_type: type):
        self.type = column_type

    @property
    def sql_type(self):
        SQLITE_TYPE_MAP = {
            int: "INTEGER",
            float: "REAL",
            str: "TEXT",
            bytes: "BLOB",
            bool: "INTEGER",  # 0 or 1
        }
        return SQLITE_TYPE_MAP[self.type]

ForeignKeys are similar to Columns but instead of holding holding fundamental types like int or str, it holds other specific table types, e.g. Author

class ForeignKey:
    def __init__(self, table: type[Table]):
        self._table = table

    @property
    def table(self):
        return self._table

# example usage
class Book(Table):
    title = Column(str)
    published = Column(bool)
    author = ForeignKey(Author)