他のユーザーから返信やいいねをもらったときに通知がくるようにしましょう。
モデル
いいねや返信用のAPIはすでに存在するので、それらのAPIのうち通知が必要なものが実行されたときに通知用のデータをDBに保存するようにしましょう。
通知に必要な情報としてはいかのようになると思います
- 通知元のユーザー
- 通知先のユーザー
- 通知種別
- 種別に応じた情報
- 通知先のユーザーが通知を表示したか
- 通知の作成日時
更新日時なんかがあっても良さそうな気もしますが使いみちが思いついていないのでYAGNI的な意味で追加しないでおきます。
server/models/notification.py
from datetime import datetime
from bson.objectid import ObjectId
from database import db
class Notification:
_collection = db["notifications"]
def __init__(
self,
user_from: str,
user_to: str,
notification_type: str,
params: object,
opened: bool = False,
created_at: datetime = None
):
self.user_from = user_from
self.user_to = user_to
self.notification_type = notification_type
self.params = params
self.opened = opened
if not created_at:
created_at = datetime.now()
self.created_at = created_at
def create(self):
data = vars(self)
data["user_from"] = ObjectId(data["user_from"])
data["user_to"] = ObjectId(data["user_to"])
return self._collection.insert_one(data)
notification_type は "like": いいねされた, "reply": 返信された, "retweet": リツイートされた, "follow": フォローされた くらいがあれば良さそうです。
params は 例えば "like" のときはいいねされた投稿のIDのように、通知の表示に必要な付加的な情報を保持するためのフィールドです。
それではそれぞれの操作で通知を保存していきましょう。
いいね
まずいいね用のデータ生成用のクラスメソッドを作っておきましょう。
server/models/notification.py
...
@classmethod
def like(
cls,
user_from,
user_to,
post_id
):
return cls(
user_from,
user_to,
"like",
{
"post_id": post_id
}
)
もちろん Notification のインスタンスを直接生成してもいいですが指定すべきフィールドがわかりやすくてこちらのほうが好みです。
server/apis/posts.py
...
@app.route("/api/posts/<post_id>/like", methods=["POST"])
@login_required()
def like_post(post_id):
post = Post._collection.find_one({"_id": ObjectId(post_id)})
if not post:
return error("投稿が存在しません", 404)
user_id = session["user"]["_id"]
like = Like._collection.find_one(
{"post": ObjectId(post_id), "liked_by": ObjectId(user_id)}
)
if like:
post["liking"] = True
return jsonify(_populate(post))
else:
like = Like(post=post_id, posted_at=post["created_at"], liked_by=user_id)
try:
like.create()
except pymongo.errors.DuplicateKeyError:
pass
like_count = Like._collection.count_documents({"post": ObjectId(post_id)})
post = Post._collection.find_one_and_update(
{"_id": ObjectId(post_id)},
{"$set": {"like_count": like_count}},
return_document=pymongo.ReturnDocument.AFTER,
)
post["liking"] = True
Notification.like(
user_id,
str(post["posted_by"]),
post_id
).create()
return jsonify(_populate(post))
...
いいねの取り消しなどはわざわざ通知しなくていいでしょう。
リツイート
ほぼ同じです
server/models/notification.py
...
@classmethod
def retweet(
cls,
user_from,
user_to,
post_id
):
return cls(
user_from,
user_to,
"retweet",
{
"post_id": post_id
}
)
server/apis/posts.py
...
@app.route("/api/posts/<post_id>/retweet", methods=["POST"])
@login_required()
def retweet_post(post_id):
post = Post._collection.find_one({"_id": ObjectId(post_id)})
if not post:
return error("投稿が存在しません", 404)
user_id = session["user"]["_id"]
retweet = Retweet._collection.find_one(
{"post": ObjectId(post_id), "retweeted_by": ObjectId(user_id)}
)
if retweet:
post["retweeting"] = True
return jsonify(_populate(post))
else:
retweet = Retweet(
post=post_id, posted_at=post["created_at"], retweeted_by=user_id
)
try:
retweet.create()
retweeting_post = Post(
content=None, posted_by=user_id, retweeted_post=post_id
)
retweeting_post.create()
except pymongo.errors.DuplicateKeyError:
pass
retweet_count = Retweet._collection.count_documents({"post": ObjectId(post_id)})
post = Post._collection.find_one_and_update(
{"_id": ObjectId(post_id)},
{"$set": {"retweet_count": retweet_count}},
return_document=pymongo.ReturnDocument.AFTER,
)
post["retweeting"] = True
Notification.retweet(
user_id,
str(post["posted_by"]),
post_id
).create()
return jsonify(_populate(post))
...
返信
server/models/notification.py
...
@classmethod
def reply(
cls,
user_from,
user_to,
post_id
):
return cls(
user_from,
user_to,
"reply",
{
"post_id": post_id
}
)
...
server/apis/posts.py
...
@app.route("/api/posts", methods=["POST"])
@login_required()
def create_post():
body = request.json
error_message = Post.validate(body)
if error_message:
return error(error_message)
if session["user"]["_id"] != body["posted_by"]:
return error("投稿者が不正です")
last = Post._collection.find_one(
{"posted_by": body["posted_by"]}, sort=[("created_at", pymongo.DESCENDING)]
)
if last and last["content"] == body["content"]:
return error("すでに同じ内容のツイートが投稿されています")
reply_to = None
if "reply_to" in body and body["reply_to"]:
reply_to = body["reply_to"]
post = Post(content=body["content"], posted_by=body["posted_by"], reply_to=reply_to)
post.create()
if reply_to:
reply_count = Post._collection.count_documents({"reply_to": ObjectId(reply_to)})
replied_post = Post._collection.find_one_and_update(
{"_id": ObjectId(reply_to)}, {"$set": {"reply_count": reply_count}}
)
Notification.reply(
session["user"]["_id"],
str(replied_post["posted_by"]),
reply_to
).create()
return jsonify(_populate(vars(post)))
...
フォロー
server/models/notification.py
...
@classmethod
def follow(
cls,
user_from,
user_to
):
return cls(
user_from,
user_to,
"follow",
{}
)
server/apis/users.py
...
@app.route("/api/users/<username>/follow", methods=["POST"])
@login_required()
def follow_user(username):
user = User._collection.find_one({"username": username})
if not user:
return error("ユーザーが存在しません。", 404)
own_id = session["user"]["_id"]
user_id = user["_id"]
User._collection.find_one_and_update(
{"username": username},
{"$addToSet": {"followers": ObjectId(own_id)}},
return_document=pymongo.ReturnDocument.AFTER,
)
me = User._collection.find_one_and_update(
{"_id": ObjectId(own_id)},
{"$addToSet": {"following": ObjectId(user_id)}},
return_document=pymongo.ReturnDocument.AFTER,
)
User.set_session_user(me)
Notification.follow(
own_id,
user_id
).create()
return jsonify(me)
...
notification.py の create を修正しておきましょう
server/models/notification.py
...
def create(self):
data = vars(self)
data["user_from"] = ObjectId(data["user_from"])
data["user_to"] = ObjectId(data["user_to"])
if self.notification_type in ["like", "retweet", "reply"]:
self.params["post_id"] = ObjectId(self.params["post_id"])
return self._collection.insert_one(data)
...
各操作で notifications にデータが保存されていることが確認できました。
次回はこの情報を画面に表示していきましょう。