Flask

Table of Contents

1 WSGI

Web Server Gateway Interface 

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    return '<h1>Hello, world!</h1>'
  • environ: [dict] http request
  • start_response: a function to send response
  • return body(html)

2 Context

Variable name Context Description
current_app Application context The application instance for the active application
g Application context An object can use for temporary storage. reset with each request
request Request context The request object, which encapsulates a http request
session Request context The user session, a dictionary storage shared by requests within session

3 Request hooks

hook decorators

  • before_first_request: register a function to run before the first request
  • before_request: register a function to run before each request
  • after_request: register a function to run after each request if no unhandled exceptions occurred
  • teardown_request: register a function to run after each request even if unhandled exceptions occurred

4 Response

4.1 tuple

  • Form: (html, status_code, headerdict)

4.2 response object

from flask import make_response

@app.route('/')
def index():
    response = make_response('<h1>This document carries a cookie!</h1>')
    response.set_cookie('answer', '42')
    return response

4.3 redirect

return redirect('http://www.example.com')

4.4 abort

used for error handling

  • abort does not return control back to the function that calls it but gives control back to the web server by raising an exception.
abort(404)

5 Jinja2 template engine

5.1 Variable

  • Form: {{ name }}
  • Jinja2 recognizes variables of any typ
<p>A value from a dictionary: {{ mydict['key'] }}.</p>
<p>A value from a list: {{ mylist[3] }}.</p>
<p>A value from a list, with a variable index: {{ mylist[myintvar] }}.</p>
<p>A value from an object's method: {{ myobj.somemethod() }}.</p>

5.2 Filters

Filter name Description
safe Renders the value without applying escaping
capitalize Converts the first char to uppercase
lower  
upper  
title  
trim Removes leading and trailing whitespace
striptags Removes any HTML tags 
<h1>Hello, {{ name|capitalize }}!</h1>

5.2.1 safe

  • Never use the safe filter on values that aren’t trusted, such as text entered by users on web forms.

5.3 Control structures

5.3.1 if

{% if user %}
    Hello, {{ user }}!
{% else %}
    Hello, Stranger!
{% endif %}

5.3.2 for loop

<ul>
    {% for comment in comments %}
	<li>{{ comment }}</li>
    {% endfor %}
</ul>

5.3.3 macro

macro are similar to functions in Python code 

{% macro render_comment(comment) %}
    <li>{{ comment }}</li>
{% endmacro %}

<ul>
    {% for comment in comments %}
	{{ render_comment(comment) }}
    {% endfor %}
</ul>
import macro from standalone macro file
{% import 'macros.html' as macros %}
<ul>
    {% for comment in comments %}
	{{ macros.render_comment(comment) }}
    {% endfor %}
</ul>

5.3.4 include common file

{% include 'common.html' %}

5.4 Inheritance

Block tags define elements that a derived template can change.

5.4.1 base template

<html>
<head>
    {% block head %}
    <title>{% block title %}{% endblock %} - My Application</title>
    {% endblock %}
</head>
<body>
    {% block body %}
    {% endblock %}
</body>
</html>

5.4.2 extendsion template

If the application needs to add its own content to a block that already has some content, then Jinja2’s super() function must be used. 

{% extends "base.html" %}
{% block title %}Index{% endblock %}
{% block head %}
    {{ super() }} <!--super() to retain the original contents -->
    <style>
    </style>
{% endblock %}
{% block body %}
<h1>Hello, World!</h1>
{% endblock %}

5.5 Custom Error Pages

@app.errorhandler(404)
def page_not_found(e):
    return render_template('404.html'), 404

@app.errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

5.6 Links

url_for() function to generate dynamic URLs from the imformation stored in the app's URL map. 

url_for('user', name='john', _external=True) # return http://localhost:5000/user/john
  • arg1: view function name
  • args: view function args
  • _external: return an absolute URL

5.7 Static files

Flask looks for static files in a subdirectory called static located in the application’s root folder. 

url_for('static', filename='css/styles.css', _external=True)

6 Web forms

Form data from clients is in request.form (POST)

6.1 flask_wtf

flask_wtf wraps the WTForms packages, handles two things:

  • generate HTML code for forms
  • validate the submitted form data

6.1.1 CSRF protection

flask_wtf uses token to verify the authenticity of requests with form data 

app = Flask(__name__)
app.config['SECRET_KEY'] = 'hard to guess string'
  • the SECRET_KEY configuration is aslo used by Flask and other third-party extendsions
  • For added security, the secret key should be stored in an environment variable

instead of being embedded in the code.

6.1.2 Form class

from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import Required

class NameForm(Form):
    name = StringField('What is your name?', validators=[Required()])
    submit = SubmitField('Submit')
  • the StringField class represents an <input> element with a type="text" attribute
  • the SubmitField class represents an <input> element with a type="submit" attribute
Fields
  • Text field: StringField, TextAreaField, PasswordField, HiddenField, DateField, DateTimeField,

IntegerField, DecimalField, FloatField

  • BooleanField: Checkbox with True and False values
  • RadioField: List of radio buttons
  • SelectField: Drop-down list of choices
  • SelectMultipleField: Drop-down list of choices with multiple selection
  • FileField: File upload field
  • SubmitField: Form submission button
  • FormField: Embed a form as a field in a container form
  • FieldList: List of fields of a given type
Validators

Email, IPAddress, Length, NumberRange, Optional, Required, Regexp, URL, AnyOf, NoneOf

  • EqualTo: useful when requesting a password to be entered twice for confirmation
Template
{% import "bootstrap/wtf.html" as wtf %}
{{ wtf.quick_form(form) }}
View function
@app.route('/', methods=['GET', 'POST'])
def index():
    name = None
    form = NameForm()
    if form.validate_on_submit():
	name = form.name.data
	form.name.data = ''
    return render_template('index.html', form=form, name=name)
  • methods argument register the view function as a handler for GET and POST requests, default GET only.
  • validate_on_submit is True when the form was submitted and the data has been accepted by all the field validators

6.2 redirect issue

Browsers repeat the last request they have sent when they are asked to refresh the page. When the last request sent is a POST request with form data, a refresh would cause a duplicate form submission.

  • Good practice: never leave a POST request as a last request sent by the browser. Respond to POST requests with a redirect instead of a normal response.
  • The trick is known as the Post/Redirect/Get pattern.

6.3 Sessions

from flask import Flask, render_template, session, redirect, url_for

@app.route('/', methods=['GET', 'POST'])
def index():
    form = NameForm()
    if form.validate_on_submit():
	session['name'] = form.name.data
	return redirect(url_for('index'))
    return render_template('index.html', form=form, name=session.get('name'))

6.4 flash

  • To give the user a confirmation message after a request is completed
from flask import flash

old_name = session.get('name')
if old_name is not None and old_name != form.name.data:
    flash('Looks like you have changed your name!')

6.4.1 render messages

The best place to render flashed messages is the base template.

  • use get_flashed_messages() to retrieve the messages and render them 

    {% for message in get_flashed_messages() %}
<div class="alert alert-warning">
    <button type="button" class="close" data-dismiss="alert">&times;</button>
    {{ message }}
</div>
{% endfor %}

7 Databases

7.1 ORMs/ODMs

  • SQLAlchemy
  • MongoEngine

7.2 flask_sqlalchemy

from flask.ext.sqlalchemy import SQLAlchemy

basedir = os.path.abspath(os.path.dirname(__file__))

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] =\
    'sqlite:///' + os.path.join(basedir, 'data.sqlite')
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True

db = SQLAlchemy(app)

7.2.1 Model

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role', laze='dynamic')

    def __repr__(self):
	return '<Role %r>' % self.name

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
	return '<User %r>' % self.username

7.2.2 Relationship

SQLAlchemy relationship options: backref, primaryjoin, lazy, uselist, order_by, secondary, secondaryjoin

7.2.3 Ops

Creating
db = SQLAlchemy(app)

db.drop_all()
db.create_all()
Inserting
admin_role = Role(name='Admin')
user_john = User(username='john', role=admin_role)
db.session.add(admin_role)
db.session.add(user_john)

or 

db.session.add_all([admin_role, user_john])

  • commit 

    db.session.commit()

  • check 

    print(admin_role.id)
  • db.session.rollback()
Modifying
admin_role.name = 'Administrator'
db.session.add(admin_role)
db.session.commit()
Deleting
db.session.delete(mod_role)
db.session.commit()
Querying
Role.query.all()
User.query.filter_by(role=admin_role).all()
str(User.query.filter_by(role=user_role)) # check SQL query

7.2.4 Integration with the Python Shell

from flask_script import Shell
from flask.ext.script import Manager
manager = Manager(app)
def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)
manager.add_command("shell", Shell(make_context=make_shell_context))

7.2.5 Database Migrations

use flask-migrate 

from flask_migrate import Migrate, MigrateCommand
migrate = Migrate(app, db)
manager.add_command('db', MigrateCommand)

python main.py db init
python main.py db migrate -m "initial migration"
python main.py db upgrade

8 Application Structure

├── __init__.py
├── main/
    ├── __init__.py
    ├── errors.py
    ├── forms.py
    └── views.py
├── static/
├── templates/
├── email.py
├── models.py
├── migrations/
├── venv/
├── tests/
├── requirements.txt
├── config.py
└── manage.py        launches the application and other application tasks.

8.1 Configuration Example

import os
basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <flasky@example.com>'
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')

    @staticmethod
    def init_app(app):
	pass

class DevelopmentConfig(Config):
    DEBUG = True
    MAIL_SERVER = 'smtp.googlemail.com'
    MAIL_PORT = 587
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
	'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
	'sqlite:///' + os.path.join(basedir, 'data-test.sqlite')

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
	'sqlite:///' + os.path.join(basedir, 'data.sqlite')

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,

    'default': DevelopmentConfig
}

8.2 Dynamic App Creation

  • app/__init__.py
from flask import Flask, render_template
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)

    from main import main as main_blueprint # see below
    app.register_blueprint(main_blueprint)

    return app
Blueprint

A blueprint is similar to an application in that it can also define routes

  • app/main/__init__.py
from flask import Blueprint
main = Blueprint('main', __name__)
from . import views, errors
  • app/main/error.py
from flask import render_template
from . import main

@main.app_errorhandler(404)
def page_not_found(e):
    return render_template('404.html'), 404

@main.app_errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

8.3 Launch Script

  • app/manage.py 

    #!/usr/bin/env python
import os
from app import create_app, db
from app.models import User, Role
from flask_script import Manager, Shell
from flask_migrate import Migrate, MigrateCommand

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
manager = Manager(app)
migrate = Migrate(app, db)

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)
manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command('db', MigrateCommand)

if __name__ == '__main__':
    manager.run()

  • add test command 

    @manager.command
def test():
    """Run the unit tests."""
    import unittest
    tests = unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

    the function’s docstring is displayed in the help messages

8.4 Project Generator

8.4.1 cookiecutter

  • cookiecutter-flask
  • cookiecutter-flask-restful

9 RESTful API

9.1 characteristics

9.1.1 Stateless

A client request must contain all the information that is necessary to carry it out. The server must not store any state about the client that persists from one request to the next.

9.1.2 Cache

Responses from the server can be labeled as cacheable or noncacheable so that clients (or intermediaries between clients and servers) can use a cache for optimization purposes.

9.1.3 Uniform Interface

often HTTP, HTTPs

9.1.4 Layered System

Proxy servers, caches, or gateways can be inserted between clients and servers as necessary to improve performance, reliability, and scalability.

9.1.5 Code-on-Demand

Clients can optionally download code from the server to execute in their context.

9.2 Request Methods

Request method Target Description HTTP status code
GET Individual resource URL Obtain the resource 200
GET Resource collection URL Obtain the collection of resources 200
POST Resource collection URL Create a new resource and add it to the collection 201
PUT Individual resource URL Create/Modify an existing resource 200
DELETE Individual resource URL Delete a resource 200
DELETE Resource collection URL Delete all resources in the collection 200

9.3 URL

/api/v1.0/[collection]/[individual]

9.4 Error Handling

  • 200: OK
  • 201: Created
  • 400: Bad request
  • 403: Forbidden
  • 404: Not found
  • 405: Method not allowed
  • 500: internal server error

9.4.1 rehandle 404, 500

404/500 are generated by Flask on its own and will usually return an HTML response 

@main.app_errorhandler(404)
def page_not_found(e):
    if request.accept_mimetypes.accept_json and \
       not request.accept_mimetypes.accept_html:
	response = jsonify({'error': 'not found'})
	response.status_code = 404
	return response
    return render_template('404.html'), 404

  • implement rest of errors 

    # app/api/errors.py: API error handler for status code 403
def forbidden(message):
    response = jsonify({'error': 'forbidden', 'message': message})
    response.status_code = 403
    return response

9.5 Authentication

9.5.1 flask-httpauth Usage

from flask_httpauth import HTTPBasicAuth
auth = HTTPBasicAuth()

@auth.verify_password
def verify_password(email, password):
    if email == '':
	g.current_user = AnonymousUser()
	return True
    user = User.query.filter_by(email = email).first()
    if not user:
	return False
    g.current_user = user
    return user.verify_password(password)
  • protect a route with the auth.login_required decorator

9.5.2 Token-based Authentication

class User(db.Model):
    # ...
    def generate_auth_token(self, expiration):
	s = Serializer(current_app.config['SECRET_KEY'],
		       expires_in=expiration)
	return s.dumps({'id': self.id})

    @staticmethod
    def verify_auth_token(token):
	s = Serializer(current_app.config['SECRET_KEY'])
	try:
	    data = s.loads(token)
	except:
	    return None
	return User.query.get(data['id'])
Verification
  • app/api_1_0/authentication.py
@auth.verify_password
def verify_password(email_or_token, password):
    if email_or_token == '':
	g.current_user = AnonymousUser()
	return True
    if password == '':
	g.current_user = User.verify_auth_token(email_or_token)
	g.token_used = True
	return g.current_user is not None
    user = User.query.filter_by(email=email_or_token).first()
    if not user:
	return False
    g.current_user = user
    g.token_used = False
    return user.verify_password(password)

@api.route('/token')
def get_token():
    if g.current_user.is_anonymous() or g.token_used:
	return unauthorized('Invalid credentials')
    return jsonify({'token': g.current_user.generate_auth_token(
	expiration=3600), 'expiration': 3600})

9.5.3 Serializing

Model should implement to_json method and @staticmethod from_json

9.5.4 Pagination of Large Resource

Example: 

@api.route('/posts/')
def get_posts():
    page = request.args.get('page', 1, type=int)
    pagination = Post.query.paginate(
	page, per_page=current_app.config['FLASKY_POSTS_PER_PAGE'],
	error_out=False)
    posts = pagination.items
    prev = None
    if pagination.has_prev:
	prev = url_for('api.get_posts', page=page-1, _external=True)
    next = None
    if pagination.has_next:
	next = url_for('api.get_posts', page=page+1, _external=True)
    return jsonify({
	'posts': [post.to_json() for post in posts],
	'prev': prev,
	'next': next,
	'count': pagination.total
    })

9.5.5 Use httpie to test API

http --json --auth : GET http://127.0.0.1:5000/api/v1.0/posts/

10 Extensions

10.1 flask_script

Command-Line Options

10.2 flask_bootstrap

base template blocks

  • doc, html_attribs, html, head, title, metas, styles, body_attribs, body, navbar, content, scripts

10.3 flask_moment

  • Localization of dates and times

client-side moment.js do the localization. flask_moment module integrates moment.js into Jinja2 templates.

10.4 flask_mail

from flask_mail import Mail
mail = Mail(app)

msg = Message('test subject', sender='you@example.com',
	      recipients=['you@example.com'])
msg.body = 'text body'
msg.html = '<b>HTML</b> body'

with app.app_context(): # send() uses current_app, so it needs to be executed with an activated application context.
    mail.send(msg)

10.4.1 Simplify Mail Sending

from flask_mail import Message

app.config['FLASKY_MAIL_SUBJECT_PREFIX'] = '[Flasky]' #application-specific configuration
app.config['FLASKY_MAIL_SENDER'] = 'Flasky Admin <flasky@example.com>' # application-specific configuration

def send_email(to, subject, template, **kwargs):
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + subject,
		  sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs) # content can be rendered!
    msg.html = render_template(template + '.html', **kwargs)
    mail.send(msg)

10.4.2 Configuration

MAIL_HOSTNAME, MAIL_PORT, MAIL_USE_TLS, MAIL_USE_SSL, MAIL_USERNAME, MAIL_PASSWORD

10.4.3 Asynchronous Sending

thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()

10.5 flask_restful