-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(framework): Converted WSGI server to a Snow Framework
After creating a static HTML render and migrating it to a simple WSGI dynamic server, now we improve our knowledge by converting the WSGI server to a simple Web Framework called Snow. We followed these steps: - Create a class called Snow, that is our framework - Instantiate an app object from our class - Create business rules in a specific module to decouple our logic from our implementation Closes #13
- Loading branch information
1 parent
64573a3
commit b8f69e5
Showing
4 changed files
with
132 additions
and
0 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,38 @@ | ||
from business_rules import add_new_post, get_posts_from_database | ||
from snow import Snow | ||
|
||
app = Snow() | ||
|
||
|
||
@app.route('^/$', template='list.template.html') | ||
def post_list(): | ||
posts = get_posts_from_database() | ||
return {'post_list': posts} | ||
|
||
|
||
@app.route('^/api$') | ||
def post_list_api(): | ||
posts = get_posts_from_database() | ||
return {'post_list': posts}, '200 OK', 'application/json' | ||
|
||
|
||
@app.route(r'^/(?P<post_id>\d{1,}$)', template='post.template.html') | ||
def post_detail(post_id): | ||
post = get_posts_from_database(post_id=post_id)[0] | ||
return {'post': post} | ||
|
||
|
||
@app.route('^/new$', template='form.template.html') | ||
def new_post_form(): | ||
return {} | ||
|
||
|
||
@app.route('^/new$', method='POST') | ||
def new_post_add(form): | ||
post = {item.name: item.value for item in form.list} | ||
add_new_post(post) | ||
return "New post Created with Success", '201 Created', 'text/plain' | ||
|
||
|
||
if __name__ == '__main__': | ||
app.run() |
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,23 @@ | ||
from database import conn | ||
|
||
cursor = conn.cursor() | ||
|
||
|
||
def get_posts_from_database(post_id=None): | ||
fields = ('id', 'title', 'content', 'author') | ||
if post_id: | ||
results = cursor.execute('SELECT * FROM post WHERE id = ?;', post_id) | ||
else: | ||
results = cursor.execute('SELECT * FROM post;') | ||
posts = [dict(zip(fields, post)) for post in results] | ||
return posts | ||
|
||
|
||
def add_new_post(post): | ||
cursor.execute( | ||
'''\ | ||
INSERT INTO post (title, content, author) VALUES (:title, :content, :author) | ||
''', | ||
post | ||
) | ||
conn.commit() |
This file contains 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,61 @@ | ||
import cgi | ||
import json | ||
import re | ||
from wsgiref.simple_server import make_server | ||
|
||
from jinja2 import Environment, FileSystemLoader | ||
|
||
|
||
class Snow: | ||
def __init__(self, template_folder='templates'): | ||
self.url_map = [] | ||
self.env = Environment(loader=FileSystemLoader(template_folder)) | ||
|
||
def route(self, rule, method='GET', template=None): | ||
def decorator(view): | ||
self.url_map.append((rule, method, view, template)) | ||
return view | ||
|
||
return decorator | ||
|
||
def render_template(self, template_name, **context): | ||
template = self.env.get_template(template_name) | ||
return template.render(**context).encode('utf-8') | ||
|
||
def __call__(self, environ, start_response): | ||
body = b'Content not found' | ||
status = '404 Not Found' | ||
content_type = 'text/html' | ||
path = environ['PATH_INFO'] | ||
request_method = environ['REQUEST_METHOD'] | ||
|
||
for rule, method, view, template in self.url_map: | ||
if match := re.match(rule, path): | ||
if method != request_method: | ||
continue | ||
view_args = match.groupdict() | ||
|
||
if method == 'POST': | ||
view_args['form'] = cgi.FieldStorage(fp=environ['wsgi.input'], environ=environ, keep_blank_values=1) | ||
|
||
view_result = view(**view_args) | ||
|
||
if isinstance(view_result, tuple): | ||
view_result, status, content_type = view_result | ||
else: | ||
status = '200 OK' | ||
|
||
if template: | ||
body = self.render_template(template, **view_result) | ||
elif isinstance(view_result, dict) and content_type == 'application/json': | ||
body = json.dumps(view_result).encode('utf-8') | ||
else: | ||
body = str(view_result).encode('utf-8') | ||
|
||
headers = [('Content-type', content_type)] | ||
start_response(status, headers) | ||
return [body] | ||
|
||
def run(self, host='0.0.0.0', port=8000): | ||
server = make_server(host, port, self) | ||
server.serve_forever() |