SqlAlchemy 过滤器创建一个 where 子句 = to "?"
SqlAlchemy filter creates a where clause = to "?"
我正在学习 Miguel Grinberg 的 flask mega 教程，但我 运行 遇到了一个我无法理解的问题。
目前我在第 8 章：追随者 ( https://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-viii-followers )
我刚刚完成了一些单元测试的部分，发现其中一个测试用例失败了。是 test_follow_posts:
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])
当 f1 被赋值时 运行s 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)
当它到达应该 filter 查询的部分时：
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)
它产生这个：
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 = ?
因此，我的测试失败的原因就很清楚了，因为 where 子句格式错误。
我试图在值中进行硬编码，但它不会更改查询，但会以某种方式更改结果：
这是调试 运行s
中的变量（Post 对象列表）及其旁边的 __repr__ 值
与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>]
与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>]
与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>]
与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 这些结果相互一致 - 有时查询有效，有时则无效
我对 SqlAlchemy 的了解还不够，甚至无法理解导致这种畸形的原因，经过大量谷歌搜索后，我无法弄清楚我认为是导致此问题的语法问题。
另一件需要注意的事情是，我使用 postgreSQL 来实际存储本教程系列的数据：
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
但不想为测试创建另一个数据库，我选择使用教程中使用的内存数据库中的 sqlite：
class UserModelCase(unittest.TestCase):
def setUp(self):
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite://"
db.create_all()
根据我的低估，SqlAlchemy 应该在我与它交互时处理数据库 (Alchemy)，所以我不明白这会如何导致问题 - 但我提到它是为了透明
附加信息：
代码已用PyCharm Professional 2021.3.1
完成
项目结构：
项目中的包：
followed_posts 中的逻辑错误
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)
应该是
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.follower_id == self.id)
follower_id 而不是 followed_id
我正在学习 Miguel Grinberg 的 flask mega 教程，但我 运行 遇到了一个我无法理解的问题。
目前我在第 8 章：追随者 ( https://blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-viii-followers )
我刚刚完成了一些单元测试的部分，发现其中一个测试用例失败了。是 test_follow_posts:
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])
当 f1 被赋值时 运行s 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)
当它到达应该 filter 查询的部分时：
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)
它产生这个：
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 = ?
因此，我的测试失败的原因就很清楚了，因为 where 子句格式错误。 我试图在值中进行硬编码，但它不会更改查询，但会以某种方式更改结果：
这是调试 运行s
中的变量（Post 对象列表）及其旁边的__repr__ 值
与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>]
与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>]
与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>]
与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 这些结果相互一致 - 有时查询有效，有时则无效
我对 SqlAlchemy 的了解还不够，甚至无法理解导致这种畸形的原因，经过大量谷歌搜索后，我无法弄清楚我认为是导致此问题的语法问题。
另一件需要注意的事情是，我使用 postgreSQL 来实际存储本教程系列的数据：
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
但不想为测试创建另一个数据库，我选择使用教程中使用的内存数据库中的 sqlite：
class UserModelCase(unittest.TestCase):
def setUp(self):
app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite://"
db.create_all()
根据我的低估，SqlAlchemy 应该在我与它交互时处理数据库 (Alchemy)，所以我不明白这会如何导致问题 - 但我提到它是为了透明
附加信息：
代码已用PyCharm Professional 2021.3.1
项目结构：
项目中的包：
followed_posts 中的逻辑错误
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.followed_id == self.id)
应该是
followed = Post.query.join(followers,(followers.c.followed_id == Post.user_id)).filter(followers.c.follower_id == self.id)
follower_id 而不是 followed_id