The Flask Mega-Tutorial 之 Chapter 8: Followers
小引
社交网往往有相互关注的特性，本节即添加“Followers”特性。
重点是调整 db，使之能够追踪 who is following whom。
Database Relationships Revisited
- 理想的情况是，对每个 user 都能维护一个 list，里面包括它的两类 users （即 followers 和 followed），但
Relational Database没有这种 list 类型。 -
Database 只储存 users 的表，所以须找到一种 relationship 类型，能够表征（model） User 的
followers / followed之间的 link。
Representing Followers
Relationship
- Many-to-Many： followers - followed
-
Self-referential： followers & followed belong to the same class User.
Association Table: followers
- followers 表中，只存储了两类 ForeignKey，都指向 User 。
Database Model Representation
1、创建辅助表 followers
app / models.py: association table
followers = db.Table('followers',
db.Column('follower_id', db.Integer, db.ForeignKey('user.id')),
db.Column('followed_id', db.Integer, db.ForeignKey('user.id'))
)注：由于作为辅助表的 followers 中，除了外键，没有其他数据，所以没有使用 model class 的方式定义，而是之间采用 db.Table() 的方式定义。
Since this is an auxiliary table that has no data other than the foreign keys, I created it without an associated model class.
2、创建 Many-to-Many 的 relationship
app / models.py: many-to-many followers relationship
class User(UserMixin, db.Model):
# ...
followed = db.relationship(
'User',
secondary=followers,
primaryjoin=(followers.c.follower_id == id),
secondaryjoin=(followers.c.followed_id == id),
backref=db.backref('followers', lazy='dynamic'),
lazy='dynamic')-
User： 因为是 self-referential 关系，故关系的另一边也是 ‘User’（relationship 名字用 Class 名，而非 table 名） -
secondary： configures the association table for this relationship primaryjoin： indicates the condition that links the left side entity (the follower user) with the association table。 其中 followers.c.follower_id 指的是 association table 的 follower_id 列。secondaryjoin： indicates the condition that links the right side entity (the followed user) with the association table.-
backref： 定义逆向关系（defines how this relationship will be accessed from the right side entity）。从左侧看，其追踪的 users 属于followed；从被追踪的 users来看，则追踪者属于followers。 -
lazy： sets up the query to not run until specifically requested, 同于 posts 中设置的 dynamic(one-to-many)。
注： SQLAlchemy tables 如果未定义成 models，则符号 “c” 是这类表的一种属性。 对这类表，其所有的字段或列，均视为属性 “c” 的子属性（sub-attributes）。
The “c” is an attribute of SQLAlchemy tables that are not defined as models. For these tables, the table columns are all exposed as sub-attributes of this “c” attribute.
3、迁移数据库并升级
(venv) $ flask db migrate -m "followers"
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
INFO [alembic.autogenerate.compare] Detected added table 'followers'
Generating /home/miguel/microblog/migrations/versions/ae346256b650_followers.py ... done
(venv) $ flask db upgrade
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
INFO [alembic.runtime.migration] Running upgrade 37f06a334dbf -> ae346256b650, followersAdding and Removing “follows”
- 依赖SQLAlchemy ORM，以及定义的 User 的
followed relationship，则在 user之间（前提是已在db存储）可以像 list 一样实现操作：
user1.followed.append(user2)
user1.followed.remove(user2)- 为提高复用性，封装成 User 的 methods
app / models.py: add and remove followers
class User(UserMixin, db.Model):
#...
def follow(self, user):
if not self.is_following(user):
self.followed.append(user)
def unfollow(self, user):
if self.is_following(user):
self.followed.remove(user)
def is_following(self, user):
return self.followed.filter(
followers.c.followed_id == user.id).count() > 0- 相较于 filter_by()，其中 filter() 更加底层，可以包含任意的 filtering 条件语句；
格式上，filter() 的条件语句更完整严格，须指明 table1.field1== table2.field2，并为”==”。 - filter_by()，can only check for equality to a constant value，并且条件语句更简单。
注：尽可能将app的逻辑从路由函数分离出来，放到 models 或其他辅助classes及模块中，因为这样便于后续的 unit testing。
It is always best to move the application logic away from view functions and into models or other auxiliary classes or modules, because as you will see later in this chapter, that makes unit testing much easier.
Obtaining the Posts from Followed Users
因为最终我们希望在 /index 页展示出 followed 的 users 的 posts。
所以写好 followed & followers 的relationship（即 db 支持已完成设置）后，考虑如何获取 followed 的 users 的 posts。
-
方法一：最容易想到的方法是，首先利用
user.followed.all()来获取 a list of followed users，然后查询每一个 user 的 posts，最终将获取到的所有 posts 并入（merg）到 单个 list 中并按日期排序。 - 弊端一：如果某个 user 追踪的人数 n 过大（followed is very large），则须先执行 n 次 单个 followed user 的 posts 查询，之后须将返回的 n 个 list of posts 进行merge 和 sorting。
- 弊端二：因为最终 Home 页会进行页码编订（pagination），首页只会显示最新的若干条 posts，设置显示更多内容的链接。这种情形下，除非我们先获取所有 posts 并按日期进行排序，否则无法知晓追踪的 users 的最新消息。
As a secondary problem, consider that the application’s home page will eventually have pagination implemented, so it will not display all the available posts but just the first few, with a link to get more if desired. If I’m going to display posts sorted by their date, how can I know which posts are the most recent of all followed users combined, unless I get all the posts and sort them first? This is actually an awful solution that does not scale well.
- 方法二：因为 Relational Database 擅长 merging & sorting，所以不将 query 放到 app 中，而是交给 db 来执行。db 通过索引（index），可以更高效地进行 query & sorting。语句如下：
app / models.py: followed posts query
class User(db.Model):
#...
def followed_posts(self):
return Post.query.join(
followers, (followers.c.followed_id == Post.user_id)).filter(
followers.c.follower_id == self.id).order_by(
Post.timestamp.desc())对 Post 进行 query，结构为 Post.query.join(...).filter(...).order_by(...)。思路为：将 post 表中的 user_id 视为 followers 表中 followed_id 一方，先将所有匹配情况拿到（得到所有被追踪的 users 发布的所有 posts），然后筛选被某个特定 user 追踪的 users 发布的posts，最后按 post 的日期排序。
1、首先，将 post 表与 followers 表进行 join，条件是followers.c.followed_id == Post.user_id（待查询 posts 属于 followed，所以 followed_id；如果待查询 posts 属于 follower，则 follower_id）。
- 若 followed_id 存在，但 user_id 不存在，则此 user 未发布 post。
- 若 followed_id 不存在，但 user_id 存在，则此 user 未被 追踪。
- 若多条 followed_id 匹配，则多人追踪此 user。
- 若多条 user_id 匹配，则此 user 发布多条 posts。
2、然后，进行 filtering。上步获得的是 followers 表中记录的所有 users 追踪的所有 followed 的 users 的 posts，为得到某个 user 的所有 followed 的users 的 posts， 则筛选条件为followers.c.follower_id == self.id。
3、按 Post 的 timestamp 进行 倒序排布，即最新的在最上方。
Combining Own and Followed Posts
app / models.py : followed posts + user’s own posts.
def followed_posts(self):
followed = Post.query.join(
followers, (followers.c.followed_id == Post.user_id)).filter(
followers.c.follower_id == self.id)
own = Post.query.filter_by(user_id=self.id)
return followed.union(own).order_by(Post.timestamp.desc())Unit Testing the User Model
为保证以后改变 app 的其他部分后，某些复杂的功能模块仍能正常工作，最好的方式就是写一组自动的测试 case，以后每次有改动，则 re-run 这组 case，看看是否正常工作。
microblog / tests.py: user model unit tests.
from datetime import datetime, timedelta
import unittest
from app import app, db
from app.models import User, Post
class UserModelCase(unittest.TestCase):
def setUp(self):
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
db.create_all()
def tearDown(self):
db.session.remove()
db.drop_all()
def test_password_hashing(self):
u = User(username='susan')
u.set_password('cat')
self.assertFalse(u.check_password('dog'))
self.assertTrue(u.check_password('cat'))
def test_avatar(self):
u = User(username='john', email='aaa@qq.com')
self.assertEqual(u.avatar(128), ('https://www.gravatar.com/avatar/'
'd4c74594d841139328695756648b6bd6'
'?d=identicon&s=128'))
def test_follow(self):
u1 = User(username='john', email='aaa@qq.com')
u2 = User(username='susan', email='aaa@qq.com')
db.session.add(u1)
db.session.add(u2)
db.session.commit()
self.assertEqual(u1.followed.all(), [])
self.assertEqual(u1.followers.all(), [])
u1.follow(u2)
db.session.commit()
self.assertTrue(u1.is_following(u2))
self.assertEqual(u1.followed.count(), 1)
self.assertEqual(u1.followed.first().username, 'susan')
self.assertEqual(u2.followers.count(), 1)
self.assertEqual(u2.followers.first().username, 'john')
u1.unfollow(u2)
db.session.commit()
self.assertFalse(u1.is_following(u2))
self.assertEqual(u1.followed.count(), 0)
self.assertEqual(u2.followers.count(), 0)
def test_follow_posts(self):
# create four users
u1 = User(username='john', email='aaa@qq.com')
u2 = User(username='susan', email='aaa@qq.com')
u3 = User(username='mary', email='aaa@qq.com')
u4 = User(username='david', email='aaa@qq.com')
db.session.add_all([u1, u2, u3, u4])
# create four posts
now = datetime.utcnow()
p1 = Post(body="post from john", author=u1,
timestamp=now + timedelta(seconds=1))
p2 = Post(body="post from susan", author=u2,
timestamp=now + timedelta(seconds=4))
p3 = Post(body="post from mary", author=u3,
timestamp=now + timedelta(seconds=3))
p4 = Post(body="post from david", author=u4,
timestamp=now + timedelta(seconds=2))
db.session.add_all([p1, p2, p3, p4])
db.session.commit()
# setup the followers
u1.follow(u2) # john follows susan
u1.follow(u4) # john follows david
u2.follow(u3) # susan follows mary
u3.follow(u4) # mary follows david
db.session.commit()
# check the followed posts of each user
f1 = u1.followed_posts().all()
f2 = u2.followed_posts().all()
f3 = u3.followed_posts().all()
f4 = u4.followed_posts().all()
self.assertEqual(f1, [p2, p4, p1])
self.assertEqual(f2, [p2, p3])
self.assertEqual(f3, [p3, p4])
self.assertEqual(f4, [p4])
if __name__ == '__main__':
unittest.main(verbosity=2)- 引入unittest，作为 class UserModelCase 的 基类（unittest.TestCase）
-
setUp()和tearDown()，是 unit testing framework 的两类特殊方法，分别在每个 test 开始之前/结束之后执行。 - setUp()里面，
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'，使SQLAlchemy 在测试中使用 in-memory SQLite database. - 写了四个测试：
test_password_hashing(), test_avatar(), test_follow(), test_follow_posts() -
db 完成操作后，记得
db.session.commit()
(venv)~/Flask_microblog $ python tests.py
[2018-06-13 10:31:14,328] INFO in __init__: Microblog startup
test_avatar (__main__.UserModelCase) ... ok
test_follow (__main__.UserModelCase) ... ok
test_follow_posts (__main__.UserModelCase) ... ok
test_password_hashing (__main__.UserModelCase) ... ok
----------------------------------------------------------------------
Ran 4 tests in 0.245s
OK
Integrating Followers with the Application
1、写路由函数
app / routes.py: follow and unfollow routes.
@app.route('/follow/<username>')
@login_required
def follow(username):
user = User.query.filter_by(username=username).first()
if user is None:
flash('User {} not found.'.format(username))
return redirect(url_for('index'))
if user == current_user:
flash('You cannot follow yourself!')
return redirect(url_for('user', username=username))
current_user.follow(user)
db.session.commit()
flash('You are following {}!'.format(username))
return redirect(url_for('user', username=username))
@app.route('/unfollow/<username>')
@login_required
def unfollow(username):
user = User.query.filter_by(username=username).first()
if user is None:
flash('User {} not found.'.format(username))
return redirect(url_for('index'))
if user == current_user:
flash('You cannot unfollow yourself!')
return redirect(url_for('user', username=username))
current_user.unfollow(user)
db.session.commit()
flash('You are not following {}.'.format(username))
return redirect(url_for('user', username=username))- 均为
@login_required - 首先判断，是否存在
user = User.query.filter_by(username=username).first() - 若 user 不存在，则 flash 提示， 并定向至
‘index’ - 若 user 存在，且为 current_user （即当前登录用户，通过
@login.user_loader导入），则无法执行 follow 或 unfollow，flash 提示后，定向至 ‘user’（即个人界面，url_for('user', username=username)） - 若 user 存在，且不为 curent_user，则可以执行 follow 或者 unfollow （注：均对
user执行，而不是username）；执行后，均需db.session.commit()
2、更新模板 user.html
app / templates / user.html: 在 user profile 页面 添加 follow 、 unfollow 链接
...
<h1>User: {{ user.username }}</h1>
{% if user.about_me %}<p>{{ user.about_me }}</p>{% endif %}
{% if user.last_seen %}<p>Last seen on: {{ user.last_seen }}</p>{% endif %}
<p>{{ user.followers.count() }} followers, {{ user.followed.count() }} following.</p>
{% if user == current_user %}
<p><a href="{{ url_for('edit_profile') }}">Edit your profile</a></p>
{% elif not current_user.is_following(user) %}
<p><a href="{{ url_for('follow', username=user.username) }}">Follow</a></p>
{% else %}
<p><a href="{{ url_for('unfollow', username=user.username) }}">Unfollow</a></p>
{% endif %}
...- 在
last_seen下面添加 数量标定
<p>{{ user.followers.count() }} followers, {{ user.followed.count() }} following.</p>
- 如果 logged-in user 浏览自己的网页，则 仍旧显示“Edit”链接。
- 如果 logged-in user 浏览尚未 follow 的 user 的 ‘user’ （profile），则显示 “Follow” 链接。
- 如果 logged-in user 浏览已经 follow 的 user 的 ‘user’ （profile），则显示 “Unfollow” 链接。
示例： 拿两个已注册的用户 Kungreye 和 Susan （注意大小写），来测试 follow 及 unfollow 。
登录 Kungreye （‘/index’）
进入 Profile 页面 （‘/user/Kungreye’），可以看到是 o followers, 0 following。
更改 addr bar 为 “~/user/Susan”，则进入 Susan 的 Profile。
同样显示 o followers, 0 following，但对于current_user （即Kungreye）来讲，此 user （即 Susan）尚未被 followed，所以显示 “Follow” 链接。追踪 Susan （点击
Follow后）后，Susan 显示有 1 follower, 0 following.-
返回 Kungreye （点击导航条 Profile，链接至
current_user（即logged-in的 Kungreye）的 Profile 页面）。 可以发现 0 follower, 1 following参照 base.html 中 Profile 的源码
<a href="{{ url_for('user', username=current_user.username) }}">Profile</a>
以上。
下一篇： Ubuntu20.04安装