无外键约束加入

Question

我正在尝试在两个表之间执行 JOIN，为此我用primaryjoin 属性描述了关系。我预计它将用于构建查询，但事实并非如此，并且sqlalchemy.exc.NoForeignKeysError: Can't find any foreign key relationships 失败。以下是重现错误所需的全部代码。请建议我如何描述我的表以及在这些表之间必须没有 ForeignKey 约束的情况下如何构造查询：

from sqlalchemy import Column, Integer, String, MetaData
from sqlalchemy import create_engine
from sqlalchemy.sql import select, insert
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship


METADATA = MetaData()
SQLAlchemyTable = declarative_base(metadata=METADATA)


class Cities(SQLAlchemyTable):
    __tablename__ = 'cities'

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)


class Orders(SQLAlchemyTable):
    __tablename__ = 'orders'

    id = Column(Integer, primary_key=True)
    city_id = Column(Integer, nullable=True)

    # relation creation code is taken from 
    # https://stackoverflow.com/questions/37806625/sqlalchemy-create-relations-but-without-foreign-key-constraint-in-db
    city = relationship(
        'Cities',
        foreign_keys=[city_id],
        primaryjoin='Orders.city_id==Cities.id'
    )


if __name__ == '__main__':
    connection_string = 'sqlite:///test_join.db'
    db = create_engine(connection_string)
    METADATA.drop_all(db)
    METADATA.create_all(db)
    db.execute(insert(Cities).values([{'id': 1, 'name': 'city_name_1'}]))
    db.execute(insert(Cities).values([{'id': 2, 'name': 'city_name_2'}]))
    db.execute(insert(Orders).values([{'id': 1, 'city_id': 1}]))
    db.execute(insert(Orders).values([{'id': 2, 'city_id': None}]))
    db.execute(insert(Orders).values([{'id': 3, 'city_id': 3}]))
    # i need to get this query:
    print(list(db.execute('SELECT orders.id, orders.city_id FROM orders JOIN cities ON orders.city_id = cities.id WHERE cities.id = 1')))
    # how?
    query = select([Orders]).join(Cities).where(Cities.id == 1)  # not working
    print(query)
    print(list(db.execute(query)))

Answer 1

relationship 是一个 ORM 构造，但您正在使用 Core 来构建您的查询。您必须明确定义如何在两者之间加入：

query = select([Orders]).\
    select_from(join(Orders, Cities, Orders.city_id == Cities.id)).\
    where(Cities.id == 1)

注意使用Select.select_from() 和join() 而不是select([...]).join()。如果使用Query API，您可以根据关系定义连接：

query = session.query(Orders).\
    join(Orders.city).\
    filter(Cities.id == 1)