'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
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|


