GitLocker: The Coding Marketplace

Description:

kaychen 0.1.3

Table of Contents

Introduction

Table of contents

Part 1

WSGI

What is WSGI
Application side

Routing
Unit test and test client
Templates
Static Files
Middleware

The middleware class, base functionality
the convoluted part
static files

allowing methods
Custom Responses
Pypi
example web app
Deploying to Heroku

workflow
other heroku commands

Part 2 - ORM

Design

Connection
table definition
creating tables
inserting data
fetch all data
query
save object with foreign key reference
fetch object with foreign key reference
update an object
delete an object

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

Introduction
Table of contents
Part 1

WSGI
Routing
Unit test and test client
Templates
Static Files
Middleware
allowing methods
Custom Responses
Pypi
example web app
Deploying to Heroku

Part 2 - ORM

Design
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 testclient 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 templatesdir 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 processrequest, then the handler of the app, then the processresponse, 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 startresponse 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

Define Procfile
heroku create

git remote is create alongside the app on heroku account
deplying via git push

git push heroku main
Check if application is deployed: heroku ps:scale web=1
View logs: heroku logs --tail

other heroku commands

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

Part 2 - ORM
ORMs allow you to

interact wiht db in own language of choice
abstract away the database (easy switching)
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)