搜索到
1
篇与
的结果
-
八、flask博客项目实战-之关注功能 概述这章将学习如何实现类似于Twitter和其他社交网络的“粉丝”或“关注”，比如关注你。接下来，将更多介绍应用程序的数据库。让用户能够轻松选择Ta们想要关注的其他用户。因此，将扩展数据库，以便跟踪谁在关注（粉）谁。重访数据库关系我想每一个用户维护一个“关注”和“关注者”列表。不幸的是，关系数据库没有我能用于这些列表的列表类型，所有都有记录、这些记录之间的关系。数据库中有一个表示用户的表，所以剩下的是提出可以为关注、关注者链接建模的正确关系类型。现在是学习基本数据库关系类型的好时机：一对多使用过一对多关系，下方就是此关系的图表：一个用户可以有多个帖子 通过这个关系连接的两个实体分别是 用户、帖子。一个用户有很多篇帖子，每个帖子有一个用户（作者）。在数据库中，这个关系在多侧使用外键。在上述关系中，外键user_id字段添加到posts表中。这个字段将每个帖子连接到user表中其作者的记录。明显地，user_id字段提供了对给定帖子的作者的直接访问，但反过来呢？为使得关系有用，我们应该可以获得给定用户所编写的帖子列表。posts表中user_id字段也足以回答此问题，因为数据库中具有允许有效查询的索引，如 【检索user_id为X的所有帖子】多对多多对多关系有点复杂。例如，考虑一个拥有students和teachers的数据库，可以说一个学生有很多个老师，一个老师有很多个学生。这就像来自两端的两个重叠的一对多关系。对于这种类型的关系，我们应该能够查询到数据库，并取得教授给定学生的教师列表、教师班级的学生列表。这在关系数据库中表示实际上并不重要，因为无法通过向现有表添加外键来完成。多对多关系的表示需要使用称为 关联表 的辅助表。下方是数据库如何查找学生和教师的案列： 虽然一开始可能看起来不太明显，但是具有两个外键的关联表 能够有效地回答关于多对多关系的所有查询。多对一、一对一多对一关系，类似于一对多关系。不同之处在于从“多”侧看这种关系。一对一关系是一对多关系的特殊情况。表示是类似的，但是得向数据库添加约束以防止“多”侧具有多个链接。虽然在某些情况下，这种关系很有用，但并不像其他关系类型常见。表示粉丝通过上述关系类型的学习，很容易确定跟踪关注者的正确数据模型是多对多关系。因为一个用户可以关注很多用户，而一个用户也可以被很多用户所关注。在学生、教师的案列中，我们通过多对多关系关联这两个实体。但在关注案列下，我有用户关注其他用户，所以只有用户。那么多对多关系的第二个实体是什么？关系的第二个实体也是用户。将类的实例链接到同一个类的其他实例的关系 称为 自引用关系，这将是我们在此所拥有的。下方是跟踪关注者的自引用 多对多关系的图表： followers表是关系的关联表。表中外键都指向user表中的行，因为它将用户链接到用户。这个表中每一个记录表示关注者用户、关注用户之间的一个链接。如同学生、教师的案列一样，像这样的设置允许数据库回答有关我们将要解决的 关注、关注用户的所有问题。用数据库模型表示首先，在数据库中添加关注者 followers，下方是followers关联表： app/models.py：关注者关联表#... from flask_login import UserMixin #关注者关联表 followers = db.Table( 'followers', db.Column('follower_id', db.Integer, db.ForeignKey('user.id')), db.Column('followed_id', db.Integer, db.ForeignKey('user.id')) ) class User(UserMixin, db.Model): #...上述代码是上一节【跟踪关注者的自引用 多对多关系的图表】的的直接翻译了。不过，注意，这里没有声明这个表为模型，即如user表、post表那样。由于这是一个除了外键而没有其他数据的辅助表，因此在没有关联模型类的情况下创建了这个表。现在，可在user表中声明多对多关系： app/models.py：多对多 关注者 关系#... class User(UserMixin, db.Model): #... last_seen = db.Column(db.DateTime, default=datetime.utcnow) 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' ) def __repr__(self): return '<User {}>'.format(self.username) #...这个关系的建立并非易事。如同我们为post表 一对多关系那样，正使用 db.relationship()函数进行定义模型类中的关系。这个关系将User实例 链接到其他User实例，因此作为约定，假设通过此关系链接到一对用户，左侧用户 关注 右侧用户。我们在左侧用户中定义了 followed 的关系，因为当我们从左侧查询这个关系时，将得到已关注的用户列表（即 右侧用户）。下方逐个检查db.relationship()的所有参数：User 是关系的右侧实体（左侧是父类）。由于这是一种自引用关系，我必须在两边使用相同的类。 secondary 配置用于这个关系的关联表，就是在这个类上定义的关联表 followers。 primaryjoin指定了左侧实体（关注者）与关联表链接的条件。关系左侧的连接条件是与关联表中follower_id字段匹配的用户ID。follwer.c.follower_id表达式引用了关联表中follower_id列。 secondaryjoin指定了右侧实体（被关注者）与关联表链接的条件。这个条件与primaryjoin类似，唯一不同的是：现在使用的followed_id，它是关联表中的另一个外键。 backref定义如何右侧实体访问这个关系。从左侧开始，关系被名称为 followed，因此右侧将使用名称followers 来表示链接到右侧目标用户的所有左侧用户。附加lazy 参数表示这个查询的执行模式，设置为dynamic模式的查询在特定请求之前不会运行，这也是我们设置帖子的一对多关系。 lazy类似于同名参数backref，但这个适用于左侧查询而不是右侧。 如果上述很难理解，不必担心。接下来将会展示如何利用这些关系来执行查询，一切将变得清晰。数据库的改变 需要记录在 新的数据库迁移中：(venv) [root@python blog]# flask db migrate -m "followers" [2018-08-15 15:25:39,889] INFO in __init__: Microblog startup 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 /root/blog/migrations/versions/65b4b5c357fa_followers.py ... done(venv) [root@python blog]# flask db upgrade [2018-08-15 15:26:03,154] INFO in __init__: Microblog startup INFO [alembic.runtime.migration] Context impl SQLiteImpl. INFO [alembic.runtime.migration] Will assume non-transactional DDL. INFO [alembic.runtime.migration] Running upgrade 00cd8a8ea68a -> 65b4b5c357fa, followers关注和取消关注由于SQLAlchemy ORM，一个用户关注另一用户 的行为 能被以followed 关系如同是一个列表一样 记录在数据库中，例如，假如我有两个用户存储在 user1 和 user2变量中，我能够用这个简单语句表示 第一个用户关注第二个用户：user1.followed.append(user2)而停止关注这个用户，可以这样做：user1.followed.remove(user2)尽管关注、取消关注 很容易，但我们希望在代码中提升可重用性，因此我们不会在代码中使用 “appends”、“removes”。代替的方法是，我们将在User模型中实现“follow”和“unfollow”方法。最后是将应用程序逻辑从视图函数移到模型或其他辅助类或模块中，因为正如在本章看到的，这让单元测试会变得更容易。下方User模型中添加、删除关系的更改： app/models.py：添加、删除关注者#... class User(UserMixin, db.Model): #... def avatar(self, size): digest = md5(self.email.lower().encode('utf-8')).hexdigest() return 'https://www.gravatar.com/avatar/{}?d=identicon&s={}'.format(digest, size) 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 #...follow()、unfollow()方法用了正如上所展示的关系对象的append()、remove()方法，但是在它们接触这个关系之前，它们用了 is_following()辅助方法以确保请求动作是有道理的。例如，假如要求 user1 关注user2，但事实证明在数据库中已经存在这个关注关系，就不必重复了。相同的逻辑同样运用到 取消关注。is_following()方法在followed关系中发出一个查询去检查两个用户是否已存在链接。在之前看到 使用SQLAlchemy的filter_by()方法去查询对象，例如 查找给定username的用户。在这用到的filter()方法是类似的，但是低水平，因为它能包含任意过滤条件，不像filter_by()只能检查与常量值的相等性。在is_following()中我们正在使用的条件 查找关联表中的项目，左侧外键设置为self 用户，右侧设置为 user参数。查询以一个count()方法终止，该方法返回结果数。这个查询结果将是0 或1，因此检查计数为1 或大于0实际上是等效的。过去使用的前提查询终止符是all()、first（）。获取已关注用户的帖子对数据库中的关注者的支持几乎已经完成，但实际上还缺少一个重要的功能。在应用程序的index页面，将显示已登录用户所关注的其他所有用户的帖子，因此，我们需要提供一个返回这些帖子的数据库查询。最明显解决方案是 运行一个返回已关注用户列表的查询，正如我们已知道的，它就是 user.followed.all()。接着，对这些返回的每个用户，我们运行查询取得帖子。一旦我们有了帖子，就将它们合并到一个列表，并按日期对它们进行排序。听起来是这样，其实不是！这种方法有几个问题。假如一个用户关注了1000人，会发送什么？那么我得执行1000此数据库查询来收集所有帖子。接着需要合并、排序内存中的1000个列表。第二个问题，考虑到应用程序的主页最终将实现分页，因此它不会显示所有可用帖子，仅是前几个，如果需要，可用一个链接去取得更多。第三个问题，如果要按日期排序显示帖子，如何知道哪些用户帖子才是所有用户中最新的？除非我得到所有帖子并先排序。这实际上是一个不能很好扩展的糟糕解决方案。实际上并没有办法避免博客帖子的合并、排序，但在应用程序中执行会导致一个非常低效的过程。这类工作是关系数据库最擅长的。数据库具有索引，允许它以更有效的方式执行查询、排序。因此，我们真正想要的是提出一个简单的数据库查询，它定义了我们想要取得的信息，然后让数据库找出如何以最有效的方式提取信息。下方就是这个查询： app/models.py：已关注用户的帖子的查询#... class User(UserMixin, db.Model): #... def is_following(self, user): return self.followed.filter(followers.c.followed_id==user.id).count()>0 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()) #...这是在这个应用程序中使用的最复杂的查询了。这个查询结构有3个主要部分 join()、filter()、order_by()，都是SQLAlchemy中查询对象的方法。联合查询要理解 join 操作的作用，先看一个例子。假设我有一个User表，包含以下内容： 简单起见，我们不显示 用户模型中的所有字段，只显示对这个查询的重要字段。假设 followers 关联表 表示用户john正在关注用户susan和david，用户susan正在关注用户mary，用户mary正在关注david。表示这个内容的数据是： 最后，posts表 包含每个用户的一条帖子： 这个表还省略了一些不属于本讨论范围的字段。这是我为这个查询再次定义的join()：Post.query.join(followers, (followers.c.followed_id == Post.user_id))上述代码正在posts表上调用join 操作。第一个参数是关注者关联表；第二个参数是连接条件。我希望数据库创建一个临时表，此表结合了posts表、followers表的数据。这个数据将根据我作为参数传递的条件进行合并。使用的条件是followers表的followed_id字段 必须等于 posts表的user_id字段。要执行此合并，数据库将从posts表（join 的左侧）获取每个记录，并附加followers表（join 的右侧）中匹配条件的所有记录。如果followers表中有多个记录符合条件，则每个记录条目将重复。如果对于给定的帖子在followers表没有匹配，那么这个帖子记录不会join操作结果中。使用上述定义的示例数据，join操作的结果是： 注意，在所有情况下，user_id和followed_id列是如何相等的，这是因为连接条件。来自用户john的帖子没有出现在上述连接表中，是因为被关注者中没有包含用户john，换句话说，没有任何人关注john。而来自用户david的帖子出现了两次，因为这个用户有个关注者。虽然创建了join操作，但暂时并未得到想要的结果。请继续，这只是更大查询的一部分。过滤join操作给我一个所有被关注用户的帖子的列表，远超出我真正想要的那部分数据。我只对这个列表的一个子集感兴趣，即某个用户关注的用户们的帖子。因此，我需要调用filter()来去掉不需要的数据。下方是查询的过滤部分：filter(followers.c.follower_id == self.id)由于这个查询是位于User类中的方法，因此self.id 表达式引用了我感兴趣的用户的用户ID。filter()选择join表中follower_id等于这个ID的行，换句话说，我只保留follower是这个用户的数据。假设我感兴趣的用户是 john，其 id 字段设置为1，下方是join表在过滤后的样子： 这些才是我想要的帖子。记住，查询是从Post类发出的，所以即使我最终得到了由数据库创建的一个临时表来作为查询的一部分，但结果是包含在此临时表中的帖子，并没有额外的列是由join操作添加的。排序查询流程的最后一步是 对结果就行排序。这部分的查询语句如下：order_by(Post.timestamp.desc())在此，希望结果是按帖子的时间戳字段按降序排序的。通过这个排序，第一个结果将是最新的博客帖子。结合自己和关注者的帖子在followed_posts()函数中使用的查询非常有用，但有一限制。人们希望看到自己的帖子也包含在Ta所关注的用户帖子的时间线中，不过这个查询目前还未建立。有两种方法可扩展此查询以包含用户自己的帖子。最直接的方法是 保持查询变，但要确保所有用户都关注自己。如果你是自己的关注者，那么上面显示的查询将会找到你自己的帖子，以及你关注的所有人帖子。这种方法的缺点是它会影响关于关注者的统计数据。所有的关注者数量都会被加1，所以必须在显示之前进行调整。第二个方法是通过创建第二个查询，返回用户自己的帖子，然后使用 union 运算符 将这两个查询合并为一个查询。在考虑了这两个选项后，决定选择第二个方法。下方将看到followed_posts()扩展后的功能，通过一个 union 包含用户自己的帖子：修改代码 app/models.py：加上用户自己的帖子#... #... 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()) #...在排序之前，注意followed 和 own 查询结果集是如何合并为一个的。对User模型进行单元测试虽然我不担心建立的这个“复杂的”关注功能的实现，但我认为它并非微不足道。但当编码重要代码时，担心的是确保这部分代码将来继续工作，因为在应用程序的不同部分进行了修改。确保已编写代码将来继续工作的最佳方法是创建一套自动化测试，每次更改时都可以重新运行。Python包含一个非常有用的 unittest包，可轻松编写、执行单元测试。在 tests.py模块中为User类存在的方法编写一些单元测试： microblog/tests.py：用户模型单元测试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='susan2018') 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='john@example.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='john@example.com') u2 = User(username='susan', email='susan@example.com') db.session.add(u1) db.session.add(u2) db.session.commit() self.assertEqual(u1.followed.all(), []) self.assertEqual(u2.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='john@example.com') u2 = User(username='susan', email='susan@example.com') u3 = User(username='mary', email='mary@example.com') u4 = User(username='david', email='david@example.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)上述添加了4个测试，用于在用户模型中执行密码哈希、用户头像、关注者功能。setUp()、tearDown()方法是特殊方法，分别用于单元测试框架之前、每次测试后执行。在setUp()实现了一些小技巧，以防止单元测试使用我们用于开发的常规数据库。通过将应用程序配置更改为 sqlite:// ，在测试期间，让SQLAlchemy使用内存中的SQLite数据库。db.create_all()创建所有数据库表。这是从头开始创建数据库的快速方法，可用于测试。对于开发环境和生产环境的数据库结构管理，之前已展示过数据库迁移的方法。下方命令可运行整个测试组件：(venv) [root@python blog]# python3 tests.py [2018-08-16 11:42:36,534] 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.211s OK (venv) [root@python blog]# 从现在开始，每次对应用程序进行更改时，都可以重新运行测试以确保正在测试的功能不受到影响。此外，每次向应用程序添加另一个功能时，都应为其编写单元测试。将关注功能集成到应用程序数据库、模型中的关注功能已实现，但还没有将它纳入到应用程序中，因此现在添加它。这个操作没有挑战，因为都是基于之前学到的概念、知识。在应用程序中添加两个新路由，以关注、取消关注一个用户：app/routes.py：关注、取消关注用户的路由@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))上述代码较简单，不过要注意当中的 所有错误检查，以防止意外问题，并在发生问题时向用户提供有用的消息。现在有了视图函数，就可以从应用程序中的页面链接到它们了。在每个用户的个人资料页面中添加链接以关注、取消关注用户：app/templates/user.html：在用户个人资料页面中添加 关注、取消关注的链接... {% 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 %} ...对用户个人资料页面更改：一是 在最近访问的时间戳下添加一行，以显示该用户拥有多少关注者和关注用户。二是当查看自己的个人资料页时，出现的“Edit”链接，可能会变成下方3种链接之一：如果用户正在查看自己的个人资料页，那么“Edit”链接跟以前一样显示； 如果用户正在查看当前未关注的用户，则会显示“Follow”链接； 如果用户正在查看当前关注的用户，则会显示“Unfollow”链接。 此时，运行应用程序，创建一些用户并测试 关注、取消关注用户的功能。 唯一需要记住的是，键入要关注 或取消关注的用户的个人资料页面的URL，因为目前无法查看用户列表。例如，如果要关注 susan，则可在浏览器地址栏输入：http://localhost:5000/user/susan 就能访问该用户的个人资料页。确保在测试关注、取消关注时，检查关注、关注者的数量变化情况。(venv) [root@python blog]# flask run * Environment: production WARNING: Do not use the development server in a production environment. Use a production WSGI server instead. * Debug mode: off [2018-08-16 14:02:45,152] INFO in __init__: Microblog startup * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit) 127.0.0.1 - - [16/Aug/2018 14:02:54] "GET /user/susan2018 HTTP/1.1" 200 -目前登录用户为susan2018，输入http://127.0.0.1:5000/user/belen，页面中显示“Follow”链接，点击它，注意页面的变化。 再注册用户john、mary、david，邮箱分别为用户名加@example.com，密码都为123456。 之前有两个用户susan（后改名susan2018，密码cat）、belen 密码Abc123456。 分别用这些用户登录、关注某些用户。进入数据库查看：(venv) [root@python blog]#sqlite3 app.db SQLite version 3.16.2 2017-01-06 16:32:41 Enter ".help" for usage hints. sqlite> select * from user; 1|susan2018|susan@example.com|pbkdf2:sha256:50000$OONOkVyy$8d008c6647ab95a5793cf60bf57eaa3bb1123d6e5b3135c5cc5e42e02eddae32|I rename my name.|2018-08-16 06:16:13.474911 2|belen|belen@example.com|pbkdf2:sha256:50000$PEDt5NxS$cf6c958c97b6ad28d9495d138cb5a310f6f2389534b0cafa3002dd3cec9af9d1|学习Flask超级教程，Python Web开发学习，坚持！|2018-08-16 06:15:38.892932 3|john|john@example.com|pbkdf2:sha256:50000$vdxx4ipx$32ccbde4bc984d459c5a99935adb8c1ce8fc5c0d3e5d7f442815e5005d1a80a4||2018-08-16 06:11:21.665502 4|mary|mary@example.com|pbkdf2:sha256:50000$8q3qPO4V$040967e4481a4882d5277a52b01902b54f3af38736336c08852f2fbae7df61a6||2018-08-16 06:12:21.538626 5|david|david@example.com|pbkdf2:sha256:50000$OhRhtXc2$c03c098ee789ef9229e2f99676f173a30e876f047991744d05f11d6afe296a31||2018-08-16 06:12:50.972967 sqlite> select * from followers; 1|2 2|3 2|4 1|5 1|4 sqlite> .quit (venv) [root@python blog]#followers表 第一列为 follower_id，第二列为 followed_id。上述含义： 1 susan2018关注了 2 belen、5 david、4 mary 2 belen 关注了 3 john、4 mary应该在应用程序 /index页面中显示所有关注帖子的列表，但由于目前用户还不能编写博客帖子（此功能还未完成）。因此，将此推迟更改，直到该功能到位。目前为止，项目结构： microblog/ app/ templates/ _post.html 404.html 500.html base.html edit_profile.html index.html login.html register.html user.html __init__.py errors.py forms.py models.py routes.py logs/ microblog.log migrations/ venv/ app.db config.py microblog.py tests.py
