實作部落格程式 part 3¶

Author: 毛毛, Alicia¶

Outline¶

  • Ch12: Followers
  • Ch13: User Comments
  • Ch14: Application Programming Interfaces

Ch12: Followers¶

Implement follower feature and filter blog posts to include only those from the users they follow¶

  • Keep track of diretional links between pairs of users and using these links in database queries

Database relationship¶

  • One-to-one relationship
  • One-to-many relationship (many-to-one relatioinship)
    1. links user roles to lists of users
    2. links users to the blog posts they authored
  • Many-to-many relationship

Many-to-many relationship¶

  • Example: a database of students and the classes they are taking.
    • You cannot just add a foreign key to the student table because a student takes many classes.

Association table¶

  • The solution is to add a thrid table called an association table.
    • Decompose many-to-many relationship into two one-to-many relationships.

Students and classes example¶

  • The assocation table here is the registrations table.
  • Querying a many-to-many relationship is a two-step process.
    1. Start form the one-to-many relationship between students and registrations and get the list of registrations.
    2. Then from the many-to-one relationship to obtain all the classes associated with the registrations.
  • In many-to-many relationship, we must set secondary argument to the association table.
In [ ]:
# hello_db.py

import os

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy

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

app = Flask(__name__)
manager = Manager(app)
app.config["SQLALCHEMY_DATABASE_URI"] =\
    "sqlite:///" + os.path.join(basedir, "data-test-m2m.sqlite")
app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_ECHO"] = False

db = SQLAlchemy(app)

# ...
In [ ]:
# hello_db.py

# ...

registrations = db.Table('registrations',
    db.Column('student_id', db.Integer, db.ForeignKey('students.id')),
    db.Column('class_id', db.Integer, db.ForeignKey('classes.id')))

class Student(db.Model):
    __tablename__ = 'students'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    classes = db.relationship('Class', secondary=registrations,
                              backref=db.backref('students', lazy='dynamic'),
                              lazy='dynamic')

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

class Class(db.Model):
    __tablename__ = 'classes'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)

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

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

Demonstration¶

  • The classes relationship uses list semantics.
  • In this relationship the backref argument also has a lazy='dynamic' attribute, so both sides return a query that can accept additional filters.
(venv) $ python hello_db.py shell
>>> from hello_db import db, Student, Class

>>> db.create_all()

>>> s = Student(name='Student_A')
>>> c = Class(name='Flask_Web')
>>> db.session.add_all([s, c])
>>> db.session.commit()

>>> s.classes.append(c)
>>> db.session.add(s)

>>> s.classes.all()
[<Class 'Flask_Web'>]
>>> c.students.all()
[<Student 'Student_A'>]

>>> s.classes.remove(c)
>>> s.classes.all()
[]

Self-Referential Relationships¶

  • A relationship in which both sides belong to the same table.
    • The entities on the left side (follwers) and on the right side (followed) are all users.
    • The association table in this case is called follows and each row represents a user following another user.

Association table as a model¶

  • We want to record the time when a user follows another user.
  • To be able to work with custom data in the relationship, we use the Follow model to represent the assication table.
In [ ]:
# app/models.py

class Follow(db.Model):
    __tablename__ = 'follows'
    follower_id = db.Column(db.Integer, db.ForeignKey('users.id'),
                            primary_key=True)
    followed_id = db.Column(db.Integer, db.ForeignKey('users.id'),
                            primary_key=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow)

Decompose the many-to-many relationship¶

  • Decomposed the many-to-many relationship into the two one-to-many relationships.
  • The followed and followers relationships are defined as individual one-to-many relationships.
    • followed: Users who are followed by this user
    • followers: Users who follow this user
  • This is necessary to eliminate any ambiguity between foreign keys by specifying in each relationship which foreign key to use through the foreign_keys optional argument.
In [ ]:
# app/models.py

class User(UserMixin, db.Model):
    # ...
    followed = db.relationship('Follow',
                               foreign_keys=[Follow.follower_id],
                               backref=db.backref('follower', lazy='joined'),
                               lazy='dynamic',
                               cascade='all, delete-orphan')
    followers = db.relationship('Follow',
                                foreign_keys=[Follow.followed_id],
                                backref=db.backref('followed', lazy='joined'),
                                lazy='dynamic',
                                cascade='all, delete-orphan')

lazy argument¶

  • On the Follow side
    • The lazy='joined' causes the related object to be loaded immediatly from the join query.
    • If lazy is set to default value of select, then the follower and followed users are loaded lazily when they are first accessed and each attribute will require an individual query.
  • On the User side
    • The lazy argument is dynamic, so the relationship attributes return query objects instead of the items directly, so that additional filters can be added to the query.

Example¶

  • Please refer to https://win911.github.io/flask_class/3_flask.slides.html#/1/81

cascade argument¶

  • The cascade argument configures how actions performed on a parent object propagate to related objects.
  • When an object is deleted, the default cascade behavior is to set the foreign key in any related objects that link to it to a null value.
  • For an association table, we want to destory the link, so we delete the entries that point to a record that was deleted. This is what the delete-orphan cascade option does.
  • all, delete-orphan leaves the default cascade options enabled and adds the delete behavior for orphans.

More details¶

  • Please refer to http://docs.sqlalchemy.org/en/latest/orm/cascades.html

Add help funtions that control the relationships¶

  • The application now needs to work with the two one-to-many relationships to implement the many-to-many functionality.
In [ ]:
# app/models.py

class User(UserMixin, db.Model):
    # ...
    def follow(self, user):
        if not self.is_following(user):
            f = Follow(follower=self, followed=user)
            self.followed.append(f)

    def unfollow(self, user):
        f = self.followed.filter_by(followed_id=user.id).first()
        if f:
            self.followed.remove(f)

    def is_following(self, user):
        if user.id is None:
            return False
        return self.followed.filter_by(followed_id=user.id).first() is not None

    def is_followed_by(self, user):
        if user.id is None:
            return False
        return self.followers.filter_by(follower_id=user.id).first() is not None

Create database migration script because model was changed¶

(venv) $ python manage.py db migrate -m "Add Follow model, follower and followed relationships to User model"

Note:¶

  • If alembic.util.exc.CommandError: Target database is not up to date occurs, please do python manage.py db upgrade first

Upgrade database¶

(venv) $ python manage.py db upgrade

Database relationships unit test¶

In [ ]:
# tests/test_user_model.py

from datetime import datetime
# ...
from app.models import User, Role, Permission, AnonymousUser, Follow

class UserModelTestCase(unittest.TestCase):
    # ...
    def test_follows(self):
        u1 = User(email='john@example.com', password='cat')
        u2 = User(email='susan@example.org', password='dog')
        db.session.add(u1)
        db.session.add(u2)
        db.session.commit()
        self.assertFalse(u1.is_following(u2))
        self.assertFalse(u1.is_followed_by(u2))

        timestamp_before = datetime.utcnow()
        u1.follow(u2)
        db.session.add(u1)
        db.session.commit()
        timestamp_after = datetime.utcnow()

        self.assertTrue(u1.is_following(u2))
        self.assertFalse(u1.is_followed_by(u2))
        self.assertTrue(u2.is_followed_by(u1))
        self.assertTrue(u1.followed.count() == 1)
        self.assertTrue(u2.followers.count() == 1)
    
        # ...
In [ ]:
# tests/test_user_model.py

class UserModelTestCase(unittest.TestCase):
    # ...
    def test_follows(self):
        # ...

        f = u1.followed.all()[-1]
        self.assertTrue(f.followed == u2)
        self.assertTrue(timestamp_before <= f.timestamp <= timestamp_after)

        f = u2.followers.all()[-1]
        self.assertTrue(f.follower == u1)

        u1.unfollow(u2)
        db.session.add(u1)
        db.session.commit()
        self.assertTrue(u1.followed.count() == 0)
        self.assertTrue(u2.followers.count() == 0)
        self.assertTrue(Follow.query.count() == 0)

        u2.follow(u1)
        db.session.add(u2)
        db.session.commit()
        db.session.delete(u2)
        db.session.commit()
        self.assertTrue(Follow.query.count() == 0)

Update the profile page¶

  • Add "Follow" or "Unfollow" button
  • Show "Followers" and "Following" informations
In [ ]:
<!-- app/templates/user.html -->

{% block page_content %}
    ...
    <div class="profile-header">
        ...
        {% if user.about_me %}<p>{{ user.about_me }}</p>{% endif %}
        <p>
            Member since {{ moment(user.member_since).format('L') }}.
            Last seen {{ moment(user.last_seen).fromNow() }}.
        </p>
        <p>{{ user.posts.count() }} blog posts.</p>
        <p>
            {% if current_user.can(Permission.FOLLOW) and user != current_user %}
                {% if not current_user.is_following(user) %}
                <a href="{{ url_for('.follow', username=user.username) }}" class="btn btn-primary">Follow</a>
                {% else %}
                <a href="{{ url_for('.unfollow', username=user.username) }}" class="btn btn-default">Unfollow</a>
                {% endif %}
            {% endif %}
            <a href="{{ url_for('.followers', username=user.username) }}">Followers: <span class="badge">{{ user.followers.count() }}</span></a>
            <a href="{{ url_for('.followed_by', username=user.username) }}">Following: <span class="badge">{{ user.followed.count() }}</span></a>
            {% if current_user.is_authenticated and user != current_user and user.is_following(current_user) %}
            | <span class="label label-default">Follows you</span>
            {% endif %}
        </p>
        ...

Routings¶

In [ ]:
# app/main/views.py

@main.route('/follow/<username>')
@login_required
@permission_required(Permission.FOLLOW)
def follow(username):
    user = User.query.filter_by(username=username).first()
    if user is None:
        flash('Invalid user.')
        return redirect(url_for('.index'))
    if current_user.is_following(user):
        flash('You are already following this user.')
        return redirect(url_for('.user', username=username))

    current_user.follow(user)
    flash('You are now following %s.' % username)
    return redirect(url_for('.user_profile', username=username))
In [ ]:
# app/main/views.py

@main.route('/unfollow/<username>')
@login_required
@permission_required(Permission.FOLLOW)
def unfollow(username):
    user = User.query.filter_by(username=username).first()
    if user is None:
        flash('Invalid user.')
        return redirect(url_for('.index'))
    if not current_user.is_following(user):
        flash('You are not following this user.')
        return redirect(url_for('.user', username=username))

    current_user.unfollow(user)
    flash('You are not following %s anymore.' % username)
    return redirect(url_for('.user_profile', username=username))
In [ ]:
# app/main/views.py

@main.route('/followers/<username>')
def followers(username):
    user = User.query.filter_by(username=username).first()
    if user is None:
        flash('Invalid user.')
        return redirect(url_for('.index'))

    page = request.args.get('page', 1, type=int)
    pagination = user.followers.paginate(
        page, per_page=current_app.config['FLASKY_FOLLOWERS_PER_PAGE'],
        error_out=False)
    follows = [{'user': item.follower, 'timestamp': item.timestamp}
               for item in pagination.items]
    return render_template('followers.html', user=user, title="Followers of",
                           endpoint='.followers', pagination=pagination,
                           follows=follows)
In [ ]:
# app/main/views.py

@main.route('/followed-by/<username>')
def followed_by(username):
    user = User.query.filter_by(username=username).first()
    if user is None:
        flash('Invalid user.')
        return redirect(url_for('.index'))

    page = request.args.get('page', 1, type=int)
    pagination = user.followed.paginate(
        page, per_page=current_app.config['FLASKY_FOLLOWERS_PER_PAGE'],
        error_out=False)
    follows = [{'user': item.followed, 'timestamp': item.timestamp}
               for item in pagination.items]
    return render_template('followers.html', user=user, title="Followed by",
                           endpoint='.followed_by', pagination=pagination,
                           follows=follows)

Add FLASKY_FOLLOWERS_PER_PAGE to config.py¶

In [ ]:
# config.py

class Config(object):
    # ...
    FLASKY_FOLLOWERS_PER_PAGE = 50

Add followers.html¶

  • Both followers and followed_by use the same tempalte followers.html.
In [ ]:
<!-- app/templates/followers.html -->

{% extends "base.html" %}
{% import "_macros.html" as macros %}

{% block title %}Flasky - {{ title }} {{ user.username }}{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>{{ title }} {{ user.username }}</h1>
</div>
<table class="table table-hover followers">
    <thead><tr><th>User</th><th>Since</th></tr></thead>
    {% for follow in follows %}
    <tr>
        <td>
            <a href="{{ url_for('.user_profile', username = follow.user.username) }}">
                <img class="img-rounded" src="{{ follow.user.gravatar(size=32) }}">
                {{ follow.user.username }}
            </a>
        </td>
        <td>{{ moment(follow.timestamp).format('L') }}</td>
    </tr>
    {% endfor %}
</table>
<div class="pagination">
    {{ macros.pagination_widget(pagination, endpoint, username = user.username) }}
</div>
{% endblock %}

username2 doesn't follow any user at first¶

After username2 followed rose84¶

After clicking the "Followers" link on rose84's profile page¶

After clicking the "Following" link on username2's profile page¶

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 12a
    • If any error occurs, do "$ git pull origin master" first.

Remember to upgrade your DB¶

  • $ python manage.py db upgrade

Query followed posts using a database join¶

  • View blog posts from only the users they follow.
  • A join operation takes two or more tables and finds all the combination of rows that satisfy a given condition.

Example of join operation¶

users table¶

posts table¶

follows table¶

To obtain the list of posts followed by user susan¶

  1. Filter the follows table to keep just the rows that have susan as the follower
  2. Create a temporary join table from combinations of rows from the posts and filtered follows tables in which the author_id of the post is the same as the followed_id of the follow

joined table¶

Join operation query¶

return db.session.query(Post).select_from(Follow).\
       filter_by(follower_id=self.id).\
       join(Post, Follow.followed_id == Post.author_id)
  • db.session.query(Post): a query that returns Post objects
  • select_from(Follow): the query begins with the Follow model
  • filter_by(follower_id=self.id): performs the filtering of the follows table by the follower user
  • join(Post, Follow.followed_id == Post.author_id): joins the results of filter_by() with the Post objects.

The query can be simplified by swapping the order of the filter and the join.¶

return Post.query.join(Follow, Follow.followed_id == Post.author_id).\
       filter(Follow.follower_id == self.id)

Note¶

  • It may seem that doing the join first and then the filtering would be more work, but in reality these two queries are equivalent.
    • The native SQL instructions for these two queries are identical.
  • SQLAlchemy first collects all the filters and then generates the query in the most efficient way.

Add followed post¶

  • The followed_posts() method is defined as a property so that it does not need the ().
  • That way, all relationships have a consistent syntax.
In [ ]:
# app/models.py

class User(UserMixin, db.Model):
    # ...
    @property
    def followed_posts(self):
        return Post.query.join(Follow, Follow.followed_id == Post.author_id)\
            .filter(Follow.follower_id == self.id)

Show followed posts on the home page¶

  • The choice of showing all or followed posts is stored in a cookie called show_followed.
  • To show all the posts, the top-level query Post.query is used, and the User.followed_posts property is used when the list should be restricted to followers.
In [ ]:
# app/main/views.py

@main.route('/', methods=['GET', 'POST'])
def index():
    form = PostForm()
    if current_user.can(Permission.WRITE_ARTICLES) and form.validate_on_submit():
        post = Post(body=form.body.data,
                    author=current_user._get_current_object())
        db.session.add(post)
        return redirect(url_for('.index'))

    page = request.args.get('page', 1, type=int)
    show_followed = False
    if current_user.is_authenticated:
        show_followed = bool(request.cookies.get('show_followed', ''))
    if show_followed:
        query = current_user.followed_posts
    else:
        query = Post.query
    
    # ...
In [ ]:
# app/main/views.py

@main.route('/', methods=['GET', 'POST'])
def index():
    # ...
    
    pagination = query.order_by(
        Post.timestamp.desc()).paginate(page,
        per_page=current_app.config['FLASKY_POSTS_PER_PAGE'], error_out=False)
    posts = pagination.items
    return render_template('index.html', form=form,
                           posts=posts, pagination=pagination)

Cookie for selection of all or followed posts¶

  • The show_followed cookie is set in two new routes.
  • Cookies can be set only on a response object, so these routes need to create a response object through make_response().
  • set_cookie() takes the cookie name and the value as the first two arguments.
  • The max_age optional argument sets the number of seconds until the cookie expires.
In [ ]:
# app/main/views.py

from flask import render_template, abort, flash, redirect, url_for
from flask import request, current_app, make_response
# ...

@main.route('/all')
@login_required
def show_all():
    resp = make_response(redirect(url_for('.index')))
    resp.set_cookie('show_followed', '', max_age=30*24*60*60)
    return resp


@main.route('/followed')
@login_required
def show_followed():
    resp = make_response(redirect(url_for('.index')))
    resp.set_cookie('show_followed', '1', max_age=30*24*60*60)
    return resp

Add navigation tab to the home page¶

In [ ]:
<!-- app/templates/index.html -->

...
{% block page_content %}
    ...
<div>
    {% if current_user.can(Permission.WRITE_ARTICLES) %}
    {{ wtf.quick_form(form) }}
    {% endif %}
</div>

<div class="post-tabs">
    <ul class="nav nav-tabs">
        <li{% if not show_followed %} class="active"{% endif %}><a href="{{ url_for('.show_all') }}">All</a></li>
        {% if current_user.is_authenticated %}
        <li{% if show_followed %} class="active"{% endif %}><a href="{{ url_for('.show_followed') }}">Followers</a></li>
        {% endif %}
    </ul>
    {% include '_posts.html' %}
</div>
{% if pagination %}
<div class="pagination">
    {{ macros.pagination_widget(pagination, '.index') }}
</div>
{% endif %}
{% endblock %}

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 12b
    • If any error occurs, do "$ git pull origin master" first.

Register all users as their own followers¶

  • Most users will expect to see their own posts when they are looking at those of their friends but users are not followers of themselves.
  • Register all users as their own followers at the time they are created.
In [ ]:
# app/models.py

class User(UserMixin, db.Model):
    #...
    def __init__(self, **kwargs):
        #...
        self.follow(self)

Update existing users¶

In [ ]:
# app/models.py

class User(UserMixin, db.Model):
    # ...
    @staticmethod
    def add_self_follows():
        for user in User.query.all():
            if not user.is_following(user):
                user.follow(user)
                db.session.add(user)
                db.session.commit()

(venv) $ python manage.py shell
>>> User.add_self_follows()

Adjust user.html¶

  • The follower and followed user counts shown in the user profile page are now increased by one due to the self-follower links.
  • The numbers need to be decreased by one to be accurate.
In [ ]:
<!-- app/templates/user.html -->
...
{% block page_content %}
    ...
    <a href="{{ url_for('.followers', username=user.username) }}">Followers: 
        <span class="badge">{{ user.followers.count() - 1 }}</span></a>
    <a href="{{ url_for('.followed_by', username=user.username) }}">Following: 
        <span class="badge">{{ user.followed.count() - 1 }}</span></a>
    ...

The lists of follower and followed users also must be adjusted¶

In [ ]:
<!-- app/templates/followers.html -->

{% extends "base.html" %}
{% import "_macros.html" as macros %}

{% block title %}Flasky - {{ title }} {{ user.username }}{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>{{ title }} {{ user.username }}</h1>
</div>
<table class="table table-hover followers">
    <thead><tr><th>User</th><th>Since</th></tr></thead>
    {% for follow in follows %}
    {% if follow.user != user %}    <!-- new -->
    <tr>
        <td>
            <a href="{{ url_for('.user_profile', username = follow.user.username) }}">
                <img class="img-rounded" src="{{ follow.user.gravatar(size=32) }}">
                {{ follow.user.username }}
            </a>
        </td>
        <td>{{ moment(follow.timestamp).format('L') }}</td>
    </tr>
    {% endif %}    <!-- new -->
    {% endfor %}
</table>
<div class="pagination">
    {{ macros.pagination_widget(pagination, endpoint, username = user.username) }}
</div>
{% endblock %}

Any unit tests that check follower counts are also affected by the self-follower links and must be adjusted to account for the self-followers¶

In [ ]:
# tests/test_user_model.py

from datetime import datetime
# ...
from app.models import User, Role, Permission, AnonymousUser, Follow

class UserModelTestCase(unittest.TestCase):
    # ...
    def test_follows(self):
        u1 = User(email='john@example.com', password='cat')
        u2 = User(email='susan@example.org', password='dog')
        db.session.add(u1)
        db.session.add(u2)
        db.session.commit()
        self.assertFalse(u1.is_following(u2))
        self.assertFalse(u1.is_followed_by(u2))

        timestamp_before = datetime.utcnow()
        u1.follow(u2)
        db.session.add(u1)
        db.session.commit()
        timestamp_after = datetime.utcnow()
    
        self.assertTrue(u1.is_following(u2))
        self.assertFalse(u1.is_followed_by(u2))
        self.assertTrue(u2.is_followed_by(u1))
        self.assertTrue(u1.followed.count() == 2)    # modified
        self.assertTrue(u2.followers.count() == 2)   # modified
    
        # ...
In [ ]:
# tests/test_user_model.py

class UserModelTestCase(unittest.TestCase):
    # ...
    def test_follows(self):
        # ...

        f = u1.followed.all()[-1]
        self.assertTrue(f.followed == u2)
        self.assertTrue(timestamp_before <= f.timestamp <= timestamp_after)

        f = u2.followers.all()[-1]
        self.assertTrue(f.follower == u1)

        u1.unfollow(u2)
        db.session.add(u1)
        db.session.commit()
        self.assertTrue(u1.followed.count() == 1)    # modified
        self.assertTrue(u2.followers.count() == 1)   # modified
        self.assertTrue(Follow.query.count() == 2)   # modified

        u2.follow(u1)
        db.session.add(u2)
        db.session.commit()
        db.session.delete(u2)
        db.session.commit()
        self.assertTrue(Follow.query.count() == 1)   # modified

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 12c
    • If any error occurs, do "$ git pull origin master" first.

Ch13: User Comments¶

Database representation of comments¶

  • Comments apply specific blog posts, so a one-to-many relationship from the posts table is defined.
  • The comments table is also in a one-to-many relationship with the users table.
  • The attributes of the Comment model are almost the same as those of Post.
  • disabled field: a Boolean that will be used by moderators to suppress comments that are inappropriate or offensive.

In [ ]:
# app/models.py

class Comment(db.Model):
    __tablename__ = 'comments'
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.Text)
    body_html = db.Column(db.Text)
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    disabled = db.Column(db.Boolean)
    author_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    post_id = db.Column(db.Integer, db.ForeignKey('posts.id'))

    @staticmethod
    def on_changed_body(target, value, oldvalue, initiator):
        allowed_tags = ['a', 'abbr', 'acronym', 'b', 'code', 'em', 'i',
                        'strong']
        target.body_html = bleach.linkify(bleach.clean(
            markdown(value, output_format='html'),
            tags=allowed_tags, strip=True))

db.event.listen(Comment.body, 'set', Comment.on_changed_body)
In [ ]:
# app/models.py
    
class Post(db.Model):
    # ...
    comments = db.relationship('Comment', backref='post', lazy='dynamic')
    
class User(UserMixin, db.Model):
    # ...
    comments = db.relationship('Comment', backref='author', lazy='dynamic')

Create database migration script because model was changed¶

(venv) $ python manage.py db migrate -m "Add Comment model and comments relationship"

Note:¶

  • If alembic.util.exc.CommandError: Target database is not up to date occurs, please do python manage.py db upgrade first

Upgrade database¶

(venv) $ python manage.py db upgrade

Comment submission and display¶

Comment submission form¶

In [ ]:
# app/main/forms.py

class CommentForm(FlaskForm):
    body = StringField('', validators=[Required()])
    submit = SubmitField('Submit')

Update post route to support comments¶

  • When a new comment is entered, the url_for() function sets the page to -1, a special page number that is used to request the last page of comments so that the comment just entered is seen on the page.
In [ ]:
# app/main/views.py

#...
from .forms import EditProfileForm, EditProfileAdminForm, PostForm, CommentForm
from ..models import Permission, User, Role, Post, Comment

@main.route('/post/<int:id>', methods=['GET', 'POST'])
def post(id):
    post = Post.query.get_or_404(id)
    form = CommentForm()
    if form.validate_on_submit():
        comment = Comment(body=form.body.data,
                          post=post,
                          author=current_user._get_current_object())
        db.session.add(comment)
        db.session.commit()
        flash('Your comment has been published.')
        return redirect(url_for('.post', id=post.id, page=-1))

    page = request.args.get('page', 1, type=int)
    if page == -1:
        page = (post.comments.count() - 1) // \
            current_app.config['FLASKY_COMMENTS_PER_PAGE'] + 1

    pagination = post.comments.order_by(Comment.timestamp.asc()).paginate(
        page, per_page=current_app.config['FLASKY_COMMENTS_PER_PAGE'],
        error_out=False)
    comments = pagination.items
    return render_template('post.html', posts=[post], form=form,
                           comments=comments, pagination=pagination)

Add FLASKY_COMMENTS_PER_PAGE to config.py¶

In [ ]:
# config.py

class Config(object):
    #...
    FLASKY_COMMENTS_PER_PAGE = 30

Add _comments.html¶

  • The comment rendering is defined in a new template _comments.html that is similar to _posts.html.
In [ ]:
<!-- app/templates/_comments.html -->

<ul class="posts">
    {% for comment in comments %}
    <li class="post">
        <div class="post-thumbnail">
            <a href="{{ url_for('.user_profile', username=comment.author.username) }}">
                <img class="img-rounded profile-thumbnail" src="{{ comment.author.gravatar(size=40) }}">
            </a>
        </div>
        <div class="post-content">
            <div class="post-date">{{ moment(comment.timestamp).fromNow() }}</div>
            <div class="post-author"><a href="{{ url_for('.user_profile', username=comment.author.username) }}">{{ comment.author.username }}</a></div>
            <div class="post-body">
                {% if comment.body_html %}
                    {{ comment.body_html | safe }}
                {% else %}
                    {{ comment.body }}
                {% endif %}
            </div>
        </div>
    </li>
    {% endfor %}
</ul>

Adjust post.html¶

  • _comments.html is included by post.html below the body of the post, followed by a call to the pagination macro.
In [ ]:
<!-- app/templates/post.html -->

{% extends "base.html" %}
{% import "_macros.html" as macros %}
{% import "bootstrap/wtf.html" as wtf %}

{% block title %}Flasky - Post{% endblock %}

{% block page_content %}
{% include '_posts.html' %}

<h4 id="comments">Comments</h4>
{% if current_user.can(Permission.COMMENT) %}
<div class="comment-form">
    {{ wtf.quick_form(form) }}
</div>
{% endif %}
{% include '_comments.html' %}
{% if pagination %}
<div class="pagination">
    {{ macros.pagination_widget(pagination, '.post', fragment='#comments',
                                id=posts[0].id) }}
</div>
{% endif %}

{% endblock %}

Add a comment link¶

  • Blog posts shown in the home and profile pages need a link to the page with the comments.
  • #comments: it is an URL fragment and is used to indicate an initial scroll position for the page.
    • This initial position is set to the comments heading in the post.html template, which is written as <h4 id="comments">Comments<h4>.
In [ ]:
<!-- app/templates/_post.html -->

<ul class="posts">
    ...
    <div class="post-content">
        ...
        <div class="post-footer">
                ...
                {% endif %}
                <a href="{{ url_for('.post', id=post.id) }}">
                    <span class="label label-default">Permalink</span>
                </a>
                <a href="{{ url_for('.post', id=post.id) }}#comments">
                    <span class="label label-primary">{{ post.comments.count() }} Comments</span>
                </a>
                ...

Adjust pagination macro¶

  • The pagination links for comments also need the #comments fragment added.
In [ ]:
<!-- app/tempaltes/_macros.html -->

{% macro pagination_widget(pagination, endpoint, fragment='') %}
...
    {% for p in pagination.iter_pages() %}
        {% if p %}
            {% if p == pagination.page %}
            <li class="active">
                <a href="{{ url_for(endpoint, page = p, **kwargs) }}{{ fragment }}">{{ p }}</a>
            </li>
            {% else %}
            <li>
                <a href="{{ url_for(endpoint, page = p, **kwargs) }}{{ fragment }}">{{ p }}</a>
            </li>
            {% endif %}

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 13a
    • If any error occurs, do "$ git pull origin master" first.

Remember to upgrade your DB¶

  • $ python manage.py db upgrade

Comment Moderation¶

  • One of the permissions was Permission.MODERATE_COMMENTS, which gives users who have it in their roles the power to moderate comments made by others.

Add moderate comments link to the navigatioin bar¶

In [ ]:
<!-- app/templates/base.html -->

...
    <div class="navbar-collapse collapse">
        <ul class="nav navbar-nav">
            ...
            {% if current_user.can(Permission.MODERATE_COMMENTS) %}
            <li><a href="{{ url_for('main.moderate') }}">Moderate Comments</a></li>
            {% endif %}
        </ul>
        ...

Add comment moderation route¶

In [ ]:
# app/main/views.py

@main.route('/moderate')
@login_required
@permission_required(Permission.MODERATE_COMMENTS)
def moderate():
    page = request.args.get('page', 1, type=int)
    pagination = Comment.query.order_by(Comment.timestamp.desc()).paginate(
        page, per_page=current_app.config['FLASKY_COMMENTS_PER_PAGE'],
        error_out=False)
    comments = pagination.items
    return render_template('moderate.html', comments=comments,
                           pagination=pagination, page=page)

Add moderate.html¶

  • It uses Jinja2’s set directive to define a moderate template variable.
  • This variable is used by the _comments.html template to determine whether the moderation features need to be rendered.
In [ ]:
<!-- app/templates/moderate.html -->

{% extends "base.html" %}
{% import "_macros.html" as macros %}

{% block title %}Flasky - Comment Moderation{% endblock %}

{% block page_content %}
<div class="page-header"> <h1>Comment Moderation</h1></div>
{% set moderate = True %}
{% include '_comments.html' %}
{% if pagination %}
<div class="pagination">
    {{ macros.pagination_widget(pagination, '.moderate') }}
</div>
{% endif %}
{% endblock %}

Modify _comments.html¶

  • For regular users, any comments that are marked as disabled should be suppressed.
  • For moderators, the body of the comment must be rendered regardless of the disabled state.
In [ ]:
<!-- app/templates/_comments.html -->

...
        <div class="post-content">
            <div class="post-date">{{ moment(comment.timestamp).fromNow() }}</div>
            <div class="post-author"><a href="{{ url_for('.user_profile', username=comment.author.username) }}">{{ comment.author.username }}</a></div>
            <div class="post-body">
                {% if comment.disabled %}
                <p><i>This comment has been disabled by a moderator.</i></p>
                {% endif %}
                {% if moderate or not comment.disabled %}
                    {% if comment.body_html %}
                        {{ comment.body_html | safe }}
                    {% else %}
                        {{ comment.body }}
                    {% endif %}
                {% endif %}
            </div>
            {% if moderate %}
                <br>
                {% if comment.disabled %}
                <a class="btn btn-default btn-xs"
                    href="{{ url_for('.moderate_enable', id=comment.id, page=page) }}">Enable</a>
                {% else %}
                <a class="btn btn-danger btn-xs"
                    href="{{ url_for('.moderate_disable', id=comment.id, page=page) }}">Disable</a>
                {% endif %}
            {% endif %}
        </div>
...

Add comment moderation routes¶

  • The comment enable and disable routes load the comment object, set the disabled field to the proper value, and write it back to the database.
In [ ]:
# app/main/views.py

@main.route('/moderate/enable/<int:id>')
@login_required
@permission_required(Permission.MODERATE_COMMENTS)
def moderate_enable(id):
    comment = Comment.query.get_or_404(id)
    comment.disabled = False
    db.session.add(comment)
    return redirect(url_for('.moderate', page=request.args.get('page', 1, type=int)))


@main.route('/moderate/disable/<int:id>')
@login_required
@permission_required(Permission.MODERATE_COMMENTS)
def moderate_disable(id):
    comment = Comment.query.get_or_404(id)
    comment.disabled = True
    db.session.add(comment)
    return redirect(url_for('.moderate', page=request.args.get('page', 1, type=int)))

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 13b
    • If any error occurs, do "$ git pull origin master" first.

Ch14: Application Programming Interfaces¶

In recent years, there has been a trend in web applications to move more and more of the business logic to the client side, producing an architecture that is known as Rich Internet Application (RIA).¶

In RIAs, the server’s main function is to provide the client application with data retrieval and storage services. In this model, the server becomes a web service or Application Programming Interface (API).¶

Introduction to REST¶

The six defining characteristics of REST architectural style for web services¶

  • Client-Server
    • There must be a clear separation between the clients and the server.
  • 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.
  • 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.
  • Uniform Interface
    • The protocol by which clients access server resources must be consistent, well defined, and standardized. The commonly used uniform interface of REST web services is the HTTP protocol.
  • Layered System
    • Proxy servers, caches, or gateways can be inserted between clients and servers as necessary to improve performance, reliability, and scalability.
  • Code-on-Demand
    • Clients can optionally download code from the server to execute in their context.

Resources Are Everything¶

  • The concept of resources is core to the REST architectural style
    • For example, in the blogging application, users, blog posts, and comments are all resources
  • Each resource must have a unique URL that represents it
    • A blog post could be represented by the URL /api/posts/12345, where 12345 is a unique identifier for the post
  • A collection of all the resources in a class also has an assigned URL
    • The URL for the collection of blog posts could be /api/posts/
  • An API can also define collection URLs that represent logical subsets of all the resources in a class
    • The collection of all comments in blog post 12345 could be represented by the URL /api/posts/12345/comments/
  • It is a common practice to define URLs that represent collections of resources with a trailing slash, as this gives them a “folder” representation
  • In a well-designed RESTful API, the client just knows a short list of top-level resource URLs and then discovers the rest from them

Request Methods¶

  • The client application sends requests to the server at the established resource URLs and uses the request method to indicate the desired operation

The request methods that are commonly used in RESTful APIs with their meanings¶

Request method Target Description HTTP status code
GET Individual resource URL Obtain the resource. 200
GET Resource collection URL Obtain the collection of resources (or one page from it if the server implements pagination). 200
POST Resource collection URL Createa new resource and add it to the collection. The server chooses the URL of the new resource and returns it in a Location header in the response. 201
Request method Target Description HTTP status code
PUT Individual resource URL Modify an existing resource. Alternatively this method can also be used to create a new resource when the client can choose the resource URL. 200
DELETE Individual resource URL Delete a resource. 200
DELETE Resource collection URL Delete all resources in the collection. 200

Note¶

  • The REST architecture does not require that all methods be implemented for a resource.
  • If the client invokes a method that is not supported for a given resource, then a response with the 405 status code for “Method Not Allowed” should be returned.
    • Flask handles this error automatically

Request and Response Bodies¶

  • Resources are sent back and forth between client and server in the bodies of requests and responses
  • REST does not specify the format to use to encode resources
    • The Content-Type header in requests and responses is used to indicate the format in which a resource is encoded in the body
    • The two formats commonly used with RESTful web services are JavaScript Object Notation (JSON) and Extensible Markup Language (XML)

Versioning¶

  • In a traditional server-centric web application, the server has full control of the application
    • When an application is updated, installing the new version in the server is enough
      • Because the parts of the application that run in the user’s web browser are downloaded from the server
  • The situation with RIAs and web services is more complicated, because often clients are developed independently of the server (maybe even by different people)
    • Web services need to be more tolerant and be able to work with old versions of its clients
      • A common way is to version the URLs handled by the web service
        • /api/v1.0/posts/ and /api/v1.1/posts/
        • For a period of time, the server handles all the URLs in their v1.0 and v1.1 variations

RESTful Web Services with Flask¶

Flask makes it very easy to create RESTful web services¶

  • The familiar route() decorator along with its methods optional argument can be used to declare the routes that handle the resource URLs exposed by the service
  • Working with JSON data is also simple
    • JSON data included with a request is automatically exposed as a request.json Python dictionary
    • A response that needs to contain JSON can be easily generated from a Python dictionary using Flask’s jsonify() helper function

The following sections show how blogging application can be extended with a RESTful web service that gives clients access to blog posts and related resources¶

Creating an API Blueprint¶

| -flasky/
   | -app/
      | -api_1_0/
         | -__init__.py
         | -users.py
         | -posts.py
         | -comments.py
         | -authentication.py
         | -errors.py
         | -decorators.py
In [ ]:
# app/api_1_0/__init__.py

from flask import Blueprint

api = Blueprint('api', __name__)

from . import authentication, posts, users, comments, errors
In [ ]:
# app/_init_.py

# ...

def create_app(config_name):
    # ...
    from .api_1_0 import api as api_1_0_blueprint
    app.register_blueprint(api_1_0_blueprint, url_prefix='/api/v1.0')
    # ...

Error Handling¶

  • A RESTful web service informs the client of the status of a request by sending the appropriate HTTP status code in the response plus any additional information in the response body

HTTP response status codes typically returned by APIs¶

HTTP status code Name Description
200 OK The request was completed successfully.
201 Created The request was completed successfully and a new resource was created as a result.
400 Bad request The request is invalid or inconsistent.
401 Unauthorized The request does not include authentication information.
403 Forbidden The authentication credentials sent with the request are insufficient for the request.
404 Not found The resource referenced in the URL was not found.
405 Method not allowed The request method requested is not supported for the given resource.
500 Internal server error An unexpected error has occurred while processing the request.

The handling of status codes 404 and 500 presents a small complication, in that these errors are generated by Flask on its own and will usually return an HTML response, which is likely to confuse an API client.¶

Content Negotiation¶

  • Generate appropriate responses for all clients is to make the error handlers adapt their responses based on the format requested by the client
    • The error handler checks the Accept request header, which Werkzeug decodes into request.accept_mimetypes, to determine what format the client wants the response in
In [ ]:
# app/api_1_0/errors.py

from flask import jsonify, request, render_template

from . import api

@api.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

@api.app_errorhandler(500)
def internal_server_error(e):
    if request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html:
        response = jsonify({'error': 'internal server error'})
        response.status_code = 500
        return response
    return render_template('500.html'), 500

The remaining status codes are generated explicitly by the web service.¶

Show the implementations of the 401 error and 403 error for example, and the others are similar¶

In [ ]:
# app/api_1_0/errors.py

# ...

def unauthorized(message):
    response = jsonify({'error': 'unauthorized', 'message': message})
    response.status_code = 401
    return response

def forbidden(message):
    response = jsonify({'error': 'forbidden', 'message': message})
    response.status_code = 403
    return response

Now view functions in the web service can invoke these auxiliary functions to generate error responses.¶

User Authentication with Flask-HTTPAuth¶

Web services, like regular web applications, need to protect information and ensure that it is not given to unauthorized parties¶

  • For this reason, RIAs must ask their users for login credentials and pass them to the server for verification

It was mentioned earlier that one of the characteristics of RESTful web services is that they are stateless, which means that the server is not allowed to “remember” anything about the client between requests¶

  • Clients need to provide all the information necessary to carry out a request in the request itself, so all requests must include user credentials

The current login functionality implemented with the help of Flask-Login stores data in the user session¶

  • Flask stores by default in a client-side cookie, so the server does not store any user-related information
    • This implementation complies with the stateless requirement of REST, but the use of cookies in RESTful web services is a bad design choice
      • It can be cumbersome for clients that are not web browsers to implement them

Because the RESTful architecture is based on the HTTP protocol, HTTP authentication is the preferred method used to send credentials, either in its Basic or Digest flavors¶

  • With HTTP authentication, user credentials are included in an Authorization header with all requests
  • Flask-HTTPAuth extension provides a convenient wrapper that hides the protocol details in a decorator similar to Flask-Login’s login_required
    • Installation: 
      (venv) $ pip install flask-httpauth

Flask-HTTPAuth initialization¶

  • This type of user authentication will be used only in the API blueprint
    • The FlaskHTTPAuth extension is initialized in the blueprint package, and not in the application package like other extensions
  • The email and password are verified using the existing support in the User model
  • The authentication callback saves the authenticated user in Flask’s g global object so that the view function can access it later
  • When the authentication credentials are invalid, the server returns a 401 error to the client
    • Flask-HTTPAuth generates a response with this status code by default
      • Use the error handler to ensure that the response is consistent with other errors returned by the API
In [ ]:
# app/api_1_0/authentication.py

from flask import g
from flask_httpauth import HTTPBasicAuth

from .errors import unauthorized
from ..models import User

auth = HTTPBasicAuth()

@auth.verify_password
def verify_password(email, password):
    if email == '':
        return False
    
    user = User.query.filter_by(email = email).first()
    if not user:
        return False
    
    g.current_user = user
    return user.verify_password(password)

@auth.error_handler
def auth_error():
    return unauthorized('Invalid credentials')

All the routes in the blueprint need to be protected¶

  • The login_required decorator can be included once in a before_request handler for the blueprint
    • The authentication checks will be done automatically for all the routes in the blueprint
    • As an additional check, also reject authenticated users who have not confirmed their accounts
In [ ]:
# app/api_1_0/authentication.py

# ...

from . import api
from .errors import unauthorized, forbidden

# ...

@api.before_request
@auth.login_required
def before_request():
    if not g.current_user.is_anonymous and not g.current_user.confirmed:
        return forbidden('Unconfirmed account')

Token-Based Authentication¶

  • To avoid having to constantly transfer sensitive information
  • The client sends the login credentials to a special URL that generates authentication tokens
    • The client can use it in place of the login credentials to authenticate requests
    • For security reasons, tokens are issued with an associated expiration
    • When a token expires, the client must reauthenticate to get a new one

Token-based authentication support¶

In [ ]:
# app/models.py

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}).decode('utf-8')

    @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'])
In [ ]:
# app/api_1_0/authentication.py

# ...

@auth.verify_password
def verify_password(email_or_token, password):
    if email_or_token == '':
        return False

    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)

To prevent clients from using an old token to request a new one, the g.token_used variable is checked¶

In [ ]:
# app/api_1_0/authentication.py

from flask import g, jsonify

# ...

@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})

Serializing Resources to and from JSON¶

  • Don't need to be identical to the internal representation of the corresponding database model

Convert a post to a JSON serializable dictionary¶

In [ ]:
# app/models.py

# ...

from flask import current_app, request, url_for
# ...

class Post(db.Model):
    # ...
    def to_json(self):
        json_post = {
            'url': url_for('api.get_post', id=self.id, _external=True),
            'body': self.body,
            'body_html': self.body_html,
            'timestamp': self.timestamp,
            'author': url_for('api.get_user', id=self.author_id, _external=True),
            'comments': url_for('api.get_post_comments', id=self.id, _external=True),
            'comment_count': self.comments.count()
        }
        return json_post

# ...

Convert a user to a JSON serializable dictionary¶

In [ ]:
# app/models.py

# ...

class User(UserMixin, db.Model):
    # ...
    def to_json(self):
        json_user = {
            'url': url_for('api.get_post', id=self.id, _external=True),
            'username': self.username,
            'member_since': self.member_since,
            'last_seen': self.last_seen,
            'posts': url_for('api.get_user_posts', id=self.id, _external=True),
            'followed_posts': url_for('api.get_user_followed_posts', id=self.id, _external=True),
            'post_count': self.posts.count()
        }
        return json_user

# ...

Create a blog post from JSON¶

In [ ]:
# app/models.py

# ...
from .exceptions import ValidationError

# ...

class Post(db.Model):
    # ...
    @staticmethod
    def from_json(json_post):
        body = json_post.get('body')
        if body is None or body == '':
            raise ValidationError('post does not have a body')
        return Post(body=body)

# ...
In [ ]:
# app/exceptions.py

class ValidationError(ValueError):
    pass
In [ ]:
# app/api_1_0/errors.py

# ...
from app.exceptions import ValidationError

# ...

def bad_request(message):
    response = jsonify({'error': 'bad request', 'message': message})
    response.status_code = 400
    return response

@api.errorhandler(ValidationError)
def validation_error(e):
    return bad_request(e.args[0])

Implementing Resource Endpoints¶

  • To implement the routes that handle the different resources

GET resource handlers for posts¶

In [ ]:
# app/api_1_0/posts.py

from flask import jsonify

from . import api
from .authentication import auth
from ..models import Post

@api.route('/posts/')
@auth.login_required
def get_posts():
    posts = Post.query.all()
    return jsonify({'posts': [post.to_json() for post in posts]})

@api.route('/posts/<int:id>')
@auth.login_required
def get_post(id):
    post = Post.query.get_or_404(id)
    return jsonify(post.to_json())

POST resource handler for posts¶

In [ ]:
# app/api_1_0/posts.py

from flask import jsonify, request, g, url_for

from . import api
from .authentication import auth
from .decorators import permission_required
from .. import db
from ..models import Post

# ...

@api.route('/posts/', methods=['POST'])
@permission_required(Permission.WRITE_ARTICLES)
def new_post():
    post = Post.from_json(request.json)
    post.author = g.current_user
    db.session.add(post)
    db.session.commit()
    return jsonify(post.to_json()), 201, \
           {'Location': url_for('api.get_post', id=post.id, _external=True)}

permission_required decorator¶

In [ ]:
# app/api_1_0/decorators.py

from functools import wraps

from flask import g

from .errors import forbidden

def permission_required(permission):
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not g.current_user.can(permission):
                return forbidden('Insufficient permissions')
            return f(*args, **kwargs)
        return decorated_function
    return decorator

PUT resource handler for posts¶

In [ ]:
# app/api_1_0/posts.py

# ...

from .errors import forbidden
from ..models import Post, Permission

# ...

@api.route('/posts/<int:id>', methods=['PUT'])
@permission_required(Permission.WRITE_ARTICLES)
def edit_post(id):
    post = Post.query.get_or_404(id)
    if g.current_user != post.author and not g.current_user.can(Permission.ADMINISTER):
        return forbidden('Insufficient permissions')
    
    post.body = request.json.get('body', post.body)
    db.session.add(post)
    return jsonify(post.to_json())

Since the application does not allow deletion of posts, the handler for the DELETE request method does not need to be implemented.¶

Pagination of Large Resource Collections¶

  • The GET requests that return a collection of resources can be extremely expensive and difficult to manage for very large collections
    • Like web applications, web services can choose to paginate collections

Post pagination¶

In [ ]:
# app/api_1_0/posts.py

from flask import jsonify, request, g, url_for, current_app

# ...

@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
    })

# ...

Finish resource handlers for users and comments during the coding time¶

In [ ]:
# app/api_1_0/users.py

from . import api

@api.route('/users/<int:id>')
def get_user(id):
    # TODO
    pass

@api.route('/users/<int:id>/posts/')
def get_user_posts(id):
    # TODO
    pass

@api.route('/users/<int:id>/timeline/')
def get_user_followed_posts(id):
    # TODO
    pass
In [ ]:
# app/api_1_0/comments.py

from . import api
from .decorators import permission_required
from ..models import Permission

@api.route('/comments/')
def get_comments():
    # TODO
    pass

@api.route('/comments/<int:id>')
def get_comment(id):
    # TODO
    pass

@api.route('/posts/<int:id>/comments/')
def get_post_comments(id):
    # TODO
    pass

@api.route('/posts/<int:id>/comments/', methods=['POST'])
@permission_required(Permission.COMMENT)
def new_post_comment(id):
    # TODO
    pass

Testing Web Services with HTTPie¶

  • The two most used clients for testing web services from the command line are curl and HTTPie

HTTPie¶

  • Installation
    (venv) $ pip install httpie

Send a GET request along with email and password to get blog posts¶

  • FLASKY_POSTS_PER_PAGE = 5
(venv) $ http --auth maomao@pyladies.com:123456 GET http://127.0.0.1:5000/api/v1.0/posts/

Send a GET request along with email and password to get authentication token¶

(venv) $ http --auth maomao@pyladies.com:123456 GET http://127.0.0.1:5000/api/v1.0/token

Send a POST request along with authentication token to add a new blog post¶

(venv) $ http --auth <token>: --json POST http://127.0.0.1:5000/api/v1.0/posts/ "body=I'm adding a post from the *command line*."

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 14a
    • If any error occurs, do "$ git pull origin master" first.

Coding Time¶

  • The resource handlers for users and comments
  • Apply pagination technique to all the routes that return collections

API resources for this application¶

Resource URL Methods Description
/users/<int:id> GET A user
/users/<int:id>/posts/ GET The blog posts written by a user
/users/<int:id>/timeline/ GET The blog posts followed by a user
/posts/ GET, POST All the blog posts
/posts/<int:id> GET, PUT A blog post
/posts/<int:id/>comments/ GET, POST The comments on a blog post
/comments/ GET All the comments
/comments/<int:id> GET A comment
In [ ]:
# app/models.py

# ...

class Comment(db.Model):
    # ...
    def to_json(self):
        json_comment = {
            'url': url_for('api.get_comment', id=self.id),
            'post_url': url_for('api.get_post', id=self.post_id),
            'body': self.body,
            'body_html': self.body_html,
            'timestamp': self.timestamp,
            'author_url': url_for('api.get_user', id=self.author_id),
        }
        return json_comment

    @staticmethod
    def from_json(json_comment):
        body = json_comment.get('body')
        if body is None or body == '':
            raise ValidationError('comment does not have a body')
        return Comment(body=body)

You can get the above code from github¶

  • $ git clone https://github.com/win911/flask_class.git
    • Skip this if you have done it before.
  • $ git checkout 14b
    • If any error occurs, do "$ git pull origin master" first.