'SqlAlchemy filter creates a where clause = to "?" [closed]

I am working through Miguel Grinberg's flask mega tutorial, and I have run into an issue I cannot understand.

Currently I am on Chapter 8: Followers ( https://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-viii-followers )

I just finished the part where I make some unit tests and found that one of the test cases failed. It is test_follow_posts:

    def test_follow_posts(self):
        # create four users
        u1 = User(username='john', email='[email protected]')
        u2 = User(username='susan', email='[email protected]')
        u3 = User(username='mary', email='[email protected]')
        u4 = User(username='david', email='[email protected]')
        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])

when f1 is assigned it runs followed_posts() from the User class in models

models.py:

from datetime import datetime
from hashlib import md5
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from app import db
from app import login


@login.user_loader
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)
    email = db.Column(db.String(120), index=True, unique=True)
    password_hash = db.Column(db.String(200))
    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)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password, method='pbkdf2:sha512:200000')

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

    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

    def followed_posts(self):
        followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)
        own = Post.query.filter_by(user_id=self.id)
        return followed.union(own).order_by(Post.timestamp.desc())
        #return followed


class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.String(140))
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow())
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

    def __repr__(self):
        return '<Post {}>'.format(self.body)

when it gets to the part where it is supposed to filter the query:

followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)

it produces this:

SELECT post.id AS post_id, post.body AS post_body, post.timestamp AS post_timestamp, post.user_id AS post_user_id 
FROM post JOIN followers ON followers.followed_id = post.user_id 
WHERE followers.followed_id = ?

Thus the reason why my test fails becomes clear because the where clause is malformed. I have attempted to hardcode in values, but it does not change the query, but it does somehow change the result:

Here is the variables (lists of Post objects) with their __repr__ value next to them from the debug runs

With followers.c.followed_id == self.id:

f1 = u1.followed_posts().all() f1: [<Post post from john>]
f2 = u2.followed_posts().all() f2: [<Post post from susan>]
f3 = u3.followed_posts().all() f3: [<Post post from mary>]
f4 = u4.followed_posts().all() f4: [<Post post from david>]

With followers.c.followed_id == 1:

f1 = u1.followed_posts().all() f1: [<Post post from john>]
f2 = u2.followed_posts().all() f2: [<Post post from susan>]
f3 = u3.followed_posts().all() f3: [<Post post from mary>]
f4 = u4.followed_posts().all() f4: [<Post post from david>]

With followers.c.followed_id == 2:

f1 = u1.followed_posts().all() f1: [<Post post from susan>, <Post post from john>]
f2 = u2.followed_posts().all() f2: [<Post post from susan>]
f3 = u3.followed_posts().all() f3: [<Post post from susan>, <Post post from mary>]
f4 = u4.followed_posts().all() f4: [<Post post from susan>, <Post post from david>]

With followers.c.followed_id == 4:

f1 = u1.followed_posts().all() f1: [<Post post from david>, <Post post from john>]
f2 = u2.followed_posts().all() f2: [<Post post from susan>, <Post post from david>]
f3 = u3.followed_posts().all() f3: [<Post post from mary>, <Post post from david>]
f4 = u4.followed_posts().all() f4: [<Post post from david>]

None of these results are consistent with each other - sometimes the query works and other times it does not

I don't know enough about SqlAlchemy to even understand what is happening to cause this malformation and after extensive googling I can't figure out the syntax issue that I assume to be the cause of this problem.

The other thing to note is that I am using postgreSQL for the actual storage of data for this tutorial series:

appConfig.py:

import os
basedir = os.path.abspath(os.path.dirname(__file__))

class Config(object):
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'you-will-never-guess'
    SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://flask:test@localhost/flaskTest'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

but not wishing to make another database for the tests I opt to use the sqlite in memory DB that is used in the tutorial:

class UserModelCase(unittest.TestCase):
    def setUp(self):
        app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite://"
        db.create_all()

From my understating SqlAlchemy is supposed to handle the database while I just interact with it(Alchemy) so I don't see how this would cause the issue - but I mention it for transparency

additional information:
The code has been done with PyCharm Professional 2021.3.1

project structure:
project structure



packages in project:
packages in project



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source