The Flask Mega-Tutorial、Chapter 8：Subscribers、Contacts、and Friends（2018エディション）

blog.miguelgrinberg.com

ミゲル・グリンバーグ

<<<前 次>>>

この記事は、Miguel Greenbergによる教科書の新版の第8部の翻訳であり、著者はこの出版物を2018年5月に完成させる予定です。

これは、Flask Mega-Tutorialシリーズの第8部で、Twitterや他のソーシャルネットワークの機能に似た「サブスクライバー」機能の実装方法を説明します。

参考までに、以下はこのシリーズの記事のリストです。

注1：このコースの古いバージョンをお探しの場合は、こちらをご覧ください 。

注2：突然、このブログの私の（ミゲル）の仕事を支持して声を上げたい場合、または単に記事を1週間待つ忍耐がない場合、私（ミゲルグリーンバーグ）はこのガイドの完全版にパッケージ化された電子書籍またはビデオを提供します。 詳細については、 learn.miguelgrinberg.comをご覧ください 。

この章では、データベースの構造についてもう少し説明します。 アプリケーションのユーザーに、興味のあるコンテンツのサブスクリプションを簡単に整理してもらいたい。 したがって、データベースを変更して、誰が誰をフォローしているかを追跡できるようにします。これは、あなたが思っているよりも少し複雑です。

この章のGitHubリンク： Browse 、 Zip 、 Diff 。

再びデータベース接続

上記で述べたように、各ユーザーの「フォロー」ユーザーと「フォロワー」ユーザーのリストを維持したいと思います。 残念ながら、リレーショナルデータベースには、これらのリストに使用できるlistタイプがありません。それは、レコードとこれらのレコード間のリレーションを持つテーブルです。

データベースにはユーザーusersを表すテーブルがありusers 。したがって、 フォロー/フォロワーリンクをモデル化できる適切なタイプの関係を考え出す必要があります。 データベースの基本的な関係のタイプを見てみましょう。

1対多

第4章ですでに1対多の関係を使用しています。 この接続の図は次のとおりです。

この関係に関連付けられている2つのオブジェクトは、ユーザーとメッセージです。 ユーザーには多くのメッセージがあり、メッセージには1人のユーザー（または作成者）がいることがわかります。 接続は、「多」側の外部キーを使用してデータベースに提示されます。 上記の接続では、外部キーはpostsメッセージテーブルに追加されたuser_idフィールドです。
このフィールドは、各メッセージをユーザーテーブルの作成者のレコードに関連付けます。

明らかに、 user_idフィールドはこのメッセージの作成者への直接アクセスを提供しuser_idが、反対方向はどうでしょうか？ 接続を有効にするには、このユーザーが作成したメッセージのリストを取得できる必要があります。
データベースには効率的なクエリを作成できるインデックスがあるため、 postsテーブルのuser_idフィールドもこの質問に答えるのに十分です。そのため、「Xからuser_idを持つすべての投稿を取得します」。

多対多

多対多の関係はやや複雑です。 例として、学生、 students 、教師、教師がいるデータベースを考えます。 生徒にはたくさんの先生がいて、先生にはたくさんの生徒がいると言えます。 これは、両端で相互接続された2つの1対多の関係に似ています。
このタイプの関係では、データベースにクエリを実行し、この学生を教える教師のリストと教師クラスの学生のリストを取得できるはずです。 これは、既存のテーブルに外部キーを追加することではできないため、リレーショナルデータベースで表すのは簡単ではありません。

多値表現を表す多対多では、 関連付けテーブルと呼ばれるヘルパーテーブルを使用する必要があります 。 以下は、データベースで生徒と教師の検索を整理する例です。

おそらく、これは誰かには不明瞭に見えるかもしれませんが、2つの外部キーとの関連付けのテーブルは、関係に対するすべての要求に効果的に答えます。

多対1および1対1

多対1は、1対多の関係のようなものです。 違いは、この関係が「多くの」側から見られることです。

1対1は1対多の特殊なケースです。 ビューは似ていますが、データベースに制限が追加されて多側を防ぎ、複数のリンクを禁止します。
このタイプの関係が役立つ場合もありますが、他のタイプほど一般的ではありません。

サブスクライバー送信

上記のすべてのタイプの関係の分析の合計から、追跡される（フォロー）が多くのユーザー（ユーザー）を監視し、ユーザー（ユーザー）が多くのサブスクライバーを持っているため、 フォロワーを追跡するための正しいデータモデルが多対多の関係であると簡単に判断できます（フォロワー）。 しかし、セットアップがあります。 生徒と教師の例では、多対多の関係で相互接続された2つのオブジェクトがありました。 しかし、フォロワーの場合、他のユーザーをフォローするユーザーがいるので、ユーザーのみがいます。 それでは、2番目の多対多の関係構造とは何ですか？

2番目の関係オブジェクトもユーザーです。
クラスのインスタンスが同じクラスの他のインスタンスに関連付けられている関係は、自己参照関係と呼ばれ、これはまさにここにあるものです。

自己参照型の多対多のサブスクライバトラッキングのチャートを次に示します。

followersテーブルは、関係の関連付けのテーブルまたは相対的な関係のテーブルです。 このテーブルの外部キーは、ユーザーをユーザーに関連付けるため、ユーザーテーブルのエントリを指します。 この表の各エントリは、フォロワーユーザーがサブスクライブしているユーザーと、フォローしているユーザーがサブスクライブしているユーザーとの関係を表します。 学生と教師の例として、このような設定により、データベースは私が必要とする署名者と購読者に関するすべての質問に答えることができます。 きれいです。

データベースモデルの表現

最初にデータベースにフォロワーを追加しましょう。 加入者の関連付けの表は次のとおりです。

 followers = db.Table('followers', db.Column('follower_id', db.Integer, db.ForeignKey('user.id')), db.Column('followed_id', db.Integer, db.ForeignKey('user.id')) ) 

これは、上記のチャートの関連テーブルのライブ変換です。 ユーザーテーブルとメッセージテーブルで行ったように、このテーブルをモデルとして宣言していないことに注意してください。 これは外部キー以外のデータを持たないヘルパーテーブルなので、対応するモデルクラスなしで作成しました。

これで、usersテーブルで多対多の関係を宣言できます。

 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') 

ご注意 翻訳者のフォロワー。 c .follower_id「c」は、モデルとして定義されていないSQLAlchemyテーブルの属性です。 これらのテーブルの場合、テーブル列はこのc属性のサブ属性として表示されます。

関係の設定は簡単ではありません。 1 db.relationship posts関係と同様に、 db.relationship関数を使用して、モデルクラスの関係を定義します。 この関係は、 Userインスタンスを他のUserインスタンスにリンクしUser 。慣例により、この関係に関連付けられたユーザーのペアに対して、左側のユーザーが右側のユーザーを監視していると言えます。 接続を定義します。左側に「 followed 」という名前followed 。この接続を左側から要求すると、後続のユーザー（つまり右側のユーザー）のリストが表示されるためです。 db.relationship()を呼び出すためのすべての引数を1つずつ見てみましょう。

• 'User'はリンクの右側です（左側は親クラスです）。 これは自己参照関係なので、両側で同じクラスを使用する必要があります。
• secondaryは、このクラスのすぐ上で定義したこの関連付けに使用される関連付けテーブルを構成します。
• primaryjoinは、左側のオブジェクト（フォロワーユーザー）を関連付けテーブルに関連付ける条件を示します。 リンクの左側の結合条件は、関連付けテーブルのfollower_idフィールドに対応するユーザー識別子です。 式followers.c.follower_idは、関連付けテーブルのfollower_id列を参照します。
• secondaryjoinは、右側のオブジェクト（フォローしているユーザー）を関連付けテーブルに関連付ける条件を定義します。 この条件はprimaryjoinに似ていますが、唯一の違いは、関連テーブルの別の外部キーであるfollowed_id使用することです。
• backrefは、オブジェクトの右側からこのリンクにアクセスする方法を決定します。 関係の左側では、ユーザーはfollowdと呼ばれるため、右側では、名前followersを使用して、右側のターゲットユーザーに関連付けられているすべての左側のユーザーを表します。 オプションのlazy引数は、このリクエストの実行モードを示します。 dynamic要求調整モードでは、特定の要求が実行されるまで起動できません。これは、1対多の関係がどのように確立されるかに関連しています。
  backrefはbackrefで同じ名前のパラメーターのように見えますが、これは右側ではなく左側を参照しています。

理解するのが難しい場合でも心配しないでください。 これらのリクエストを処理する方法を説明すると、すぐにすべてが明確になります。

データベースへの変更は、新しいデータベース移行で記録する必要があります。

 (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, followers 

「フォロー」の追加と削除（サブスクライバー）

SQLAlchemy ORMのおかげで、別のユーザーを追跡するためにサインアップしているユーザーは、リストのようにデータベースに書き込むことができます。 たとえば、変数user1およびuser2に2人のユーザーが格納されている場合、次の簡単なステートメントで最初のuser1 2番目のuser2追従させることができuser2 。

 user1.followed.append(user2) 

user1がuser2追跡を拒否するには、次のようにします。

 user1.followed.remove(user2) 

サブスクライバーの追加と削除は非常に簡単ですが、コードで再利用しやすくしたいので、「追加」（追加）および「削除」（コードを介して）するつもりはありません。 代わりに、「フォロー」および「フォロー解除」機能をUserモデルのメソッドとして実装しUser 。 アプリケーションロジックを表示機能からモデルまたは他の補助クラスまたはモジュールに移動すると、この章の後半で説明するように、ユニットテストがはるかに簡単になるため、常に優れています。

以下は、関係を追加および削除するためのユーザーモデルの変更点です。

 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 

follow()およびunfollow()メソッドは、上記のようにオブジェクトのappend()およびremove()メソッドを使用しますが、適用される前にis_following() checkメソッドを使用して、要求されたアクションが意味をなすことを確認します。 たとえば、 user1にuser2を監視するように依頼したが、そのようなタスクがデータベースにすでに存在することが判明した場合、なぜそれを複製するのでしょうか。 同じロジックをunfollowingに適用できます。

is_following()メソッドは、2人のユーザーの間に関係が存在するかどうかを確認する要求を生成します。 filter_by()は、SQLAlchemyクエリのfilter_by()メソッドを使用して、たとえばユーザー名でユーザーを検索していました。 ここで使用するfilter()メソッドは似ていますが、より低いレベルです。これは、定数値の等価性のみをチェックできるfilter_by()とは異なり、任意のフィルタリング条件を含めることができるためです。 is_following()使用する条件は、左側の外部キーがユーザーself設定され、右側がuser引数に設定されている関連付けテーブル内の要素を検索します。 要求は、レコード数を返すcount()メソッドで終了します。 このクエリの結果は0または1になるため、カウンターが1以上であるか、0より大きいかを確認することは実際には同等です。 過去に見た他のクエリターミネータは、 all()とfirst()です。

フォローしているユーザーからメッセージを受信する

データベース内のサブスクライバーのサポートはほぼ完了していますが、実際には重要な機能はありません。 アプリケーションのインデックスページで、登録ユーザーがフォローしているすべての人が書いたブログエントリを表示するため、これらのメッセージを返すデータベースクエリを生成する必要があります。

最も明らかな解決策は、フォローしているユーザーのリストを返すクエリを実行することです。これは、ご存じのとおり、 user.followed.all()です。 次に、これらのユーザーごとに、クエリを実行してメッセージを受信できます。 すべてのメッセージが揃ったら、それらを1つのリストにまとめて日付順に並べ替えることができます。 それはいいですね？ まあ、そうでもない。

このアプローチにはいくつかの問題があります。 ユーザーのサブスクリプションが数千人になるとどうなりますか？ すべてのメッセージを収集するには、数千のデータベースクエリを実行する必要があります。 そして、メモリ内の何千ものリストを結合してソートする必要があります。 二次的な問題として、アプリケーションのホームページが最終的にページ分割されるため、利用可能なすべてのメッセージではなく、必要に応じて詳細を取得するためのリンクを含む最初の数ページのみを表示します。 メッセージを日付順にソートして表示する場合、すべてのメッセージを取得して最初にソートしない限り、フォローしている（フォローしている）ユーザーの最後のメッセージを見つけるにはどうすればよいですか？ これは非常に不気味なソリューションであり、適切に拡張できません。

このブログ投稿のプーリングとソートを回避する方法はありませんが、アプリケーションでこれを行うと、非常に非効率的なプロセスにつながります。 この種の作業は、リレーショナルデータベースを異なるものにします。 データベースには、より効率的な方法でクエリとソートを実行できるインデックスがあります。 したがって、受信したい情報を定義する単一のデータベースクエリを作成し、この情報を最も効率的な方法で抽出する方法をデータベースに理解させます。

リクエストは次のとおりです。

 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()) 

これはおそらく、このアプリケーションで使用した最も複雑なクエリです。 一度にこの要求を解読しようとします。 このクエリの構造を見ると、SQLAlchemyクエリオブジェクトのjoin() 、 filter()およびorder_by()メソッドによって開発された3つの主要なセクションがあることがわかります。

 Post.query.join(...).filter(...).order_by(...) 

ユニオン操作-結合

結合操作の機能を理解するために、例を見てみましょう。 次の内容のUserテーブルがあるとします。

簡単にするために、ユーザーモデルのすべてのフィールドではなく、この要求に重要なフィールドのみを示します。

followers関連付けテーブルに、 john susanとsusanを、 susanはmaryを、 maryはsusanいるとします。 上記のデータは次のとおりです。

その結果、 postsテーブルは各ユーザーから1つの投稿を返します。

このリクエストに対して定義したjoin()呼び出しは次のとおりです。

Post.query.join(followers, (followers.c.followed_id == Post.user_id))

postsテーブルで結合操作を呼び出します。 最初の引数はサブスクライバーアソシエーションテーブルで、2番目の引数は結合条件です。 この呼び出しで構築しているのは、投稿テーブルと購読者テーブルのデータを結合する一時テーブルをデータベースで作成することです。 データは、引数として渡した条件に従って結合されます。

使用した条件では、フォロワーテーブルのfollowed_idフィールドuser_id 、 postsテーブルのuser_idと等しくなければなりません。 このマージを実行するために、データベースはメッセージテーブル（接続の左側）から各レコードを取得し、 followersテーブル（接続の右側）から条件に一致するレコードを追加します。 followers複数のレコードが条件を満たしている場合、それぞれのレコードが繰り返されます。 フォロワーにこのメッセージに一致するものがない場合、このエントリは参加の一部ではありません。

結合操作の結果：

これは結合条件であるため、すべての場合でuser_id列とfollowed_id列が等しいことに注意してください。 johnが関心のあるユーザーであるサブスクリプションエントリがないため、つまり、johnメッセージをフォローしているユーザーがいないため、johnからのメッセージは参加テーブルに表示されません。 ただし、2人の異なるユーザーがこのユーザーをフォローしているため、davidエントリが2回表示されます。

このリクエストを完了することで何が得られるかはすぐにはわかりませんが、これは大きなリクエストの一部に過ぎないため、継続しています。

フィルター

参加操作により、一部のユーザーがフォローしているすべてのメッセージのリストが表示され、これは本当に必要なデータ量を超えています。 このリストのサブセット、つまり1人のユーザーのみがフォローしているメッセージにのみ関心があるため、不要なレコードをすべてトリミングする必要があり、 filter()呼び出してこれを行うことができます。

クエリフィルターパーツは次のとおりです。

 filter(followers.c.follower_id == self.id) 

このリクエストはUserクラスのメソッドにあるため、式self.idは、関心のあるユーザーのユーザー識別子を参照します。 filter()呼び出しは、 follower_id列がこのユーザーを指す結合テーブル内のアイテムを選択します。つまり、このユーザーがサブスクライバーであるエントリーのみを保存することを意味します。

idフィールドが1設定されているユーザーjohn興味があるとします。 フィルタリング後のクエリ結果は次のようになります。

そして、これらはまさに私が見たかった投稿です！

クエリはPostクラスに対して送信されたものであるため、このクエリの一部としてデータベースによって作成された一時テーブルを受け取ったとしても、結果はこの一時テーブルに含まれるレコードになり、結合操作によって追加の列が追加されることはありません。

仕分け

最後のステップは、結果をソートすることです。 読み取りのリクエストの部分：

 order_by(Post.timestamp.desc()) 

ここでは、結果をメッセージのタイムスタンプフィールドで降順で並べ替えたいと言います。 この条件下では、最初の結果は最新のブログ投稿になります。

独自のメッセージと購読済みメッセージの組み合わせ

followed_posts ()関数で示したリクエストは非常に便利ですが、1つの制限があります。 人々は自分のメッセージを年表に署名済みのメッセージと一緒に見ることを期待していますが、そうではありませんでした。 リクエストにはこの機能はありません。

独自のユーザーレコードを含めることにより、このクエリを拡張するには2つの方法があります。最も簡単な方法は、リクエストをそのままにすることですが、すべてのユーザーが自分自身を見ていることを確認してください。あなたがあなた自身の加入者である場合、上記のように、リクエストはあなたに興味があるすべての人のリクエストと共にあなた自身のメッセージを見つけます。この方法の欠点は、サブスクライバに関する統計に影響することです。すべてのカウンターが1つ増えるため、表示する前に調整する必要があります。これを行う2番目の方法は、ユーザー自身のメッセージを返す2番目のクエリを作成し、ユニオン演算子を使用して2つのクエリを1つに結合することです。

, . follow_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()) 

, followed , .

UnitTest User Model

, «» , , . , , , , . , , , , — , , .

Python unittest , . User 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='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='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(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='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) 

, , . setUp() tearDown() — , . setUp() , , . sqlite:// SQLAlchemy SQLite . db.create_all() . , . , .

:

 (venv) $ python tests.py 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.494s OK 

, , , , . , , , .

, , , . , , , .

, :

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

, , , , .

, View , . :

  ... <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 %} ... 

, , , . , "" (Edit), :

• , "Edit" , .

• , , "Follow".

• , , "Unfollow".

, . , , URL , , . , susan , http://localhost:5000/user/susan , . , , , .

c index , , , . , .

<<<前 次>>>