【问题标题】:SQLAlchemy ORM NOT EXISTS: table alias scopingSQLAlchemy ORM 不存在:表别名范围
【发布时间】:2020-07-10 17:24:04
【问题描述】:

我有三个实体,ABC,其中CAs 链接到Bs (A-*C-B)。我想找到那些没有未连接到BC 实例的A 实例。

我无法想出一个 SQLAlchemy 查询来为我做这件事,我开始认为编译器有问题。

以下单元测试说明了这一点:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, create_engine, literal

_sql_engine = create_engine('sqlite:///:memory:')
session = sessionmaker(bind=_sql_engine)()

def test_model():
    Base = declarative_base()

    class A(Base):
        __tablename__ = 'a'
        id = Column(Integer, primary_key=True)

    class B(Base):
        __tablename__ = 'b'
        id = Column(Integer, primary_key=True)

    class C(Base):
        __tablename__ = 'c'
        a = Column(Integer, primary_key=True)
        b = Column(Integer, primary_key=True)


    Base.metadata.create_all(_sql_engine)

    a = [A(id=10),      A(id=20)]
    c = [C(a=10, b=11), C(a=20, b=21)]
    b = [B(id=11)]

    session.add_all(a + c + b)
    session.commit()

    q = session.query(A).filter(
        A.id < literal(100),
        ~(
            session.query(C)
                .filter(
                A.id == C.a,
                ~(
                    session.query(B)
                    .filter(
                        B.id == C.b,
                    ).exists()
                )
            ).exists()
        )
    )

    print(q.statement)
    print(len(q.all()))

    assert len(q.all()) == 1

测试需要一个结果,但结果为零。打印出来的 SQL 语句是

SELECT a.id 
FROM a 
WHERE a.id < :param_1 AND NOT (EXISTS (SELECT 1 
FROM c 
WHERE NOT (EXISTS (SELECT 1 
FROM b, a 
WHERE b.id = c.b AND a.id = c.a))))

现在,在我看来问题出在第三个 FROM 语句上。 ba 覆盖上面的别名并与之前的约束断开连接。

这是正确的吗?这是 SQL 中的作用域的工作原理吗?如果是这样,我是在使用 SQLAlchemy 时出错还是这是一个错误?

(单元测试使用 SQLite,但最终结果应该在 PostgreSQL 中运行。)

【问题讨论】:

    标签: python sql postgresql sqlite sqlalchemy


    【解决方案1】:

    感谢 github 上的 zzzeek,我找到了答案:在第二次嵌套之后,别名在子查询中被遮蔽。如果我们不想要这个,我们使用correlated subqueries

    固定单元测试:

    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker, aliased
    from sqlalchemy import Column, Integer, create_engine, literal
    
    _sql_engine = create_engine('sqlite:///:memory:')
    session = sessionmaker(bind=_sql_engine)()
    
    
    def test_model():
        Base = declarative_base()
    
        class A(Base):
            __tablename__ = 'a'
            id = Column(Integer, primary_key=True)
    
        class B(Base):
            __tablename__ = 'b'
            id = Column(Integer, primary_key=True)
    
        class C(Base):
            __tablename__ = 'c'
            a = Column(Integer, primary_key=True)
            b = Column(Integer, primary_key=True)
    
    
        Base.metadata.create_all(_sql_engine)
    
        a = [A(id=10),      A(id=20)]
        c = [C(a=10, b=11), C(a=20, b=21)]
        b = [B(id=11)]
    
        session.add_all(a + c + b)
        session.commit()
    
        q = session.query(A).filter(
            A.id < literal(100),
            ~(
                session.query(C)
                    .filter(
                    A.id == C.a,
                    ~(
                        session.query(B)
                        .filter(
                            B.id == C.b
                        ).exists().correlate(A,C)
                    )
                ).exists()
            )
        )
    
        print(q.statement)
        print(len(q.all()))
    
        assert len(q.all()) == 1
    

    【讨论】:

      猜你喜欢
      • 2011-07-18
      • 1970-01-01
      • 1970-01-01
      • 2021-09-17
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多