因此,我有三个表:
班级定义:
engine = create_engine('sqlite://test.db', echo=False) SQLSession = sessionmaker(bind=engine) Base = declarative_base() class Channel(Base): __tablename__ = 'channel' id = Column(Integer, primary_key = True) title = Column(String) description = Column(String) link = Column(String) pubDate = Column(DateTime) class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key = True) username = Column(String) password = Column(String) sessionId = Column(String) class Subscription(Base): __tablename__ = 'subscription' userId = Column(Integer, ForeignKey('user.id'), primary_key=True) channelId = Column(Integer, ForeignKey('channel.id'), primary_key=True)
注意: 我知道user.username应该是唯一的,需要解决这个问题,而且我不确定SQLalchemy为什么用双引号创建一些行名。
我正在尝试找到一种检索所有频道的方法,以及一个特定用户(由user.sessionId和user.id标识)在哪些频道上进行订阅的指示。
例如,假设我们有四个通道:channel1,channel2,channel3,channel4;用户:user1;在channel1和channel4上有订阅的用户。对user1的查询将返回类似以下内容的内容:
channel.id | channel.title | subscribed --------------------------------------- 1 channel1 True 2 channel2 False 3 channel3 False 4 channel4 True
这是最好的结果,但是由于我完全不知道如何完成已订阅列,因此我一直在尝试获取用户已预订且缺少订阅的行中的特定用户ID。 ,请将其留空。
我与SQLalchemy atm一起使用的数据库引擎。是sqlite3
我已经花了两天时间来解决这个问题,通过订阅表将所有三个通道连接在一起没有问题,但是没有用户订阅的所有通道都会被忽略。
我希望我已经尽力描述了我的问题,在此先感谢。
编辑 :设法以一种涉及子查询的笨拙方式解决了这个问题:
# What a messy SQL query! stmt = query(Subscription).filter_by(userId = uid()).join((User, Subscription.userId == User.id)).filter_by(sessionId = id()).subquery() subs = aliased(Subscription, stmt) results = query(Channel.id, Channel.title, subs.userId).outerjoin((subs, subs.channelId == Channel.id))
但是,我将继续寻找更优雅的解决方案,因此答案仍然非常受欢迎。
选项1:
Subscription只是一个多对多的关系对象,我建议您对它进行建模而不是作为一个单独的类。请参阅的《配置多对多关系》文档SQLAlchemy/declarative。
Subscription
SQLAlchemy/declarative
使用测试代码进行建模将变为:
from sqlalchemy import create_engine, Column, Integer, DateTime, String, ForeignKey, Table from sqlalchemy.orm import relation, scoped_session, sessionmaker, eagerload from sqlalchemy.ext.declarative import declarative_base engine = create_engine('sqlite:///:memory:', echo=True) session = scoped_session(sessionmaker(bind=engine, autoflush=True)) Base = declarative_base() t_subscription = Table('subscription', Base.metadata, Column('userId', Integer, ForeignKey('user.id')), Column('channelId', Integer, ForeignKey('channel.id')), ) class Channel(Base): __tablename__ = 'channel' id = Column(Integer, primary_key = True) title = Column(String) description = Column(String) link = Column(String) pubDate = Column(DateTime) class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key = True) username = Column(String) password = Column(String) sessionId = Column(String) channels = relation("Channel", secondary=t_subscription) # NOTE: no need for this class # class Subscription(Base): # ... Base.metadata.create_all(engine) # ###################### # Add test data c1 = Channel() c1.title = 'channel-1' c2 = Channel() c2.title = 'channel-2' c3 = Channel() c3.title = 'channel-3' c4 = Channel() c4.title = 'channel-4' session.add(c1) session.add(c2) session.add(c3) session.add(c4) u1 = User() u1.username ='user1' session.add(u1) u1.channels.append(c1) u1.channels.append(c3) u2 = User() u2.username ='user2' session.add(u2) u2.channels.append(c2) session.commit() # ###################### # clean the session and test the code session.expunge_all() # retrieve all (I assume those are not that many) channels = session.query(Channel).all() # get subscription info for the user #q = session.query(User) # use eagerload(...) so that all 'subscription' table data is loaded with the user itself, and not as a separate query q = session.query(User).options(eagerload(User.channels)) for u in q.all(): for c in channels: print (c.id, c.title, (c in u.channels))
产生以下输出:
(1, u'channel-1', True) (2, u'channel-2', False) (3, u'channel-3', True) (4, u'channel-4', False) (1, u'channel-1', False) (2, u'channel-2', True) (3, u'channel-3', False) (4, u'channel-4', False)
请注意,使用时eagerload,仅会发出1条SELECT语句,而不是每次要求User时发出1条SELECT语句channels。
eagerload
User
channels
选项2:
但是,如果您想保持模型并只创建一个SA查询,该查询将根据您的要求为您提供列,那么以下查询就可以完成此工作:
from sqlalchemy import and_ from sqlalchemy.sql.expression import case #... q = (session.query(#User.username, Channel.id, Channel.title, case([(Subscription.channelId == None, False)], else_=True) ).outerjoin((Subscription, and_(Subscription.userId==User.id, Subscription.channelId==Channel.id)) ) ) # optionally filter by user q = q.filter(User.id == uid()) # assuming uid() is the function that provides user.id q = q.filter(User.sessionId == id()) # assuming uid() is the function that provides user.sessionId res = q.all() for r in res: print r
输出与上面的选项1完全相同。