如何使用 SQLalchemy 连接三个表并将所有列保留在其中一个表中？

Question

所以，我有三个表：

类定义：

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；谁订阅了频道 1 和频道 4。 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))

但是，我将继续寻找更优雅的解决方案，因此仍然非常欢迎回答。

Answer 1

选项 1：

Subscription 只是一个多对多关系对象，我建议您将其建模为这样，而不是作为一个单独的类。请参阅SQLAlchemy/declarative 的Configuring Many-to-Many Relationships 文档。

你的模型用测试代码变成：

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 的使用，当请求channels 时，它只会为每个User 发出1 个SELECT 语句而不是1 个。

选项 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

输出和上面的option-1完全一样。

Answer 2

为了让这件事更简单一点，我在你的模型中添加了关系，这样你就可以通过 user.subscriptions 来获取所有订阅。

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)
    user = relationship(User, primaryjoin=userId == User.id, backref='subscriptions')
    channelId = Column(Integer, ForeignKey('channel.id'), primary_key=True)
    channel = relationship(channel, primaryjoin=channelId == channel.id, backref='subscriptions')

results = session.query(
    Channel.id,
    Channel.title,
    Channel.subscriptions.any().label('subscribed'),
)

for channel in results:
    print channel.id, channel.title, channel.subscribed

Answer 3

不要向用户查询。从频道查询。

user = query(User).filter_by(id=1).one()
for channel in query(Channel).all():
    print channel.id, channel.title, user in channel.subscriptions.user

这样您可以获得所有频道，而不仅仅是与相关用户相关联的频道。