Chapeter 8

app/__init__.py

@@ -2,12 +2,10 @@ import os
 from flask import Flask
 from config import Config
 import logging
-from loggin.handlers import RotatingFileHandler
-from logging.handlers import SMTPHandler
+from logging.handlers import RotatingFileHandler, SMTPHandler
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 from flask_login import LoginManager
-from app import routes, models, errors
 
 app = Flask(__name__)
 app.config.from_object(Config)
@@ -54,3 +52,4 @@ if not app.debug:
 
     app.logger.setLevel(logging.INFO)
     app.logger.info("Microblog startup")
+from app import routes, models, errors

app/forms.py

@@ -57,3 +57,7 @@ class EditProfileForm(FlaskForm):
             user = User.query.filter_by(username=self.username.data).first()
             if user is not None:
                 raise ValidationError("Please use a different username.")
+
+
+class EmptyForm(FlaskForm):
+    submit = SubmitField("Submit")

app/models.py

@@ -10,6 +10,13 @@ def load_user(id):
     return User.query.get(int(id))
 
 
+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):
     id = db.Column(db.Integer, primary_key=True)
     username = db.Column(db.String(64), index=True, unique=True)
@@ -18,6 +25,14 @@ class User(UserMixin, db.Model):
     posts = db.relationship("Post", backref="author", lazy="dynamic")
     about_me = db.Column(db.String(140))
     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)
@@ -34,6 +49,27 @@ class User(UserMixin, db.Model):
             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
+        )
+
+    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())
+
 
 class Post(db.Model):
     id = db.Column(db.Integer, primary_key=True)

app/routes.py

@@ -3,7 +3,7 @@ from flask import render_template, flash, redirect, url_for, request
 from flask_login import current_user, login_user, logout_user, login_required
 from werkzeug.urls import url_parse
 from app import app, db
-from app.forms import LoginForm, RegistrationForm, EditProfileForm
+from app.forms import LoginForm, RegistrationForm, EditProfileForm, EmptyForm
 from app.models import User
 
 
@@ -94,3 +94,43 @@ def edit_profile():
     return render_template(
         "edit_profile.html", title="Edit Profile", form=form
     )
+
+
+@app.route("/follow/<username>", methods=["POST"])
+@login_required
+def follow(username):
+    form = EmptyForm()
+    if form.validate_on_submit():
+        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))
+    else:
+        return redirect(url_for("index"))
+
+
+@app.route("/unfollow/<username>", methods=["POST"])
+@login_required
+def unfollow(username):
+    form = EmptyForm()
+    if form.validate_on_submit():
+        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))
+    else:
+        return redirect(url_for("index"))

app/templates/user.html

@@ -17,10 +17,25 @@
       </td>
     </tr>
   </table>
+  <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>
+      <form action="{{ url_for('follow', username = user.username) }}" method="post">
+        {{ form.hidden_tag() }}
+        {{ form.submit(value = 'Follow') }}
+      </form>
+    </p>
+  {% else %}
+    <p>
+      <form action="{{ url_for('unfollow', username = user.username) }}" method="post">
+        {{ form.hidden_tag() }}
+        {{ form.submit(value = 'Unfollow') }}
+      </form>
+    </p>
   {% endif %}
   <hr />
   {% for post in posts %}

migrations/versions/619062c0486b_followers.py

@@ -0,0 +1,33 @@
+"""followers
+
+Revision ID: 619062c0486b
+Revises: 680d311746e8
+Create Date: 2023-06-03 21:28:40.382514
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '619062c0486b'
+down_revision = '680d311746e8'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('followers',
+    sa.Column('follower_id', sa.Integer(), nullable=True),
+    sa.Column('followed_id', sa.Integer(), nullable=True),
+    sa.ForeignKeyConstraint(['followed_id'], ['user.id'], ),
+    sa.ForeignKeyConstraint(['follower_id'], ['user.id'], )
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('followers')
+    # ### end Alembic commands ###

tests.py

@@ -0,0 +1,114 @@
+import os
+
+os.environ["DATABASE_URL"] = "sqlite://"
+
+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):
+        self.app_context = app.app_context()
+        self.app_context.push()
+        db.create_all()
+
+    def tearDown(self):
+        db.session.remove()
+        db.drop_all()
+        self.app_context.pop()
+
+    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)