python-3.x - sqlalchemy ORM 查询对象根据上下文返回不同类型的结果

Question

我正在使用炼金术 ORM 的 Query 对象。
在不同的上下文中，它返回不同的类型。

我在 postgres 中有下表:

select * from test;
 test_id | test_col1 | test_col2 
---------+-----------+-----------
       1 | ABC       | ABC
       2 | XYZ       | XYZ

我在 Python3 中有以下类(class):

class Test (DeclarativeBase):
    __tablename__ = 'test'

    id = sa.Column('test_id', sa.Integer(), primary_key=True)
    col1 = sa.Column('test_col1', sa.String(250), nullable = True)
    col2 = sa.Column('test_col2', sa.String(250), nullable = True)

我的查询和结果:

In [2]: q1 = session.query(Test)                                                                                                                                                                                                                                                         

In [3]: rows1 = q1.all()                                                                                                                                                                                                                                                                 

In [4]: print(type(rows1[0]))                                                                                                                                                                                                                                                            
<class '__main__.Test'>

In [5]:                                                                                                                                                                                                                                                                                  

In [5]: cte = session.query(Test).cte()                                                                                                                                                                                                                                                  

In [6]: q2 = session.query(cte)                                                                                                                                                                                                                                                          

In [7]: rows2 = q2.all()                                                                                                                                                                                                                                                                 

In [8]: print(type(rows2[0]))                                                                                                                                                                                                                                                            
<class 'sqlalchemy.util._collections.result'>

In [9]:                                                                                                                                                                                                                                                                                  

In [9]: q3 = session.query(Test).filter(Test.col1 == 'ABC')                                                                                                                                                                                                                              

In [10]: rows3 = q3.all()                                                                                                                                                                                                                                                                

In [11]: print(type(rows3[0]))                                                                                                                                                                                                                                                           
<class '__main__.Test'>

In [12]:                                                                                                                                                                                                                                                                                 

In [12]:                                                                                                                                                                                                                                                                                 

In [12]: q4 = session.query(Test).filter(Test.col1 == 'ABC').one()                                                                                                                                                                                                                       

In [13]: print(type(q4))                                                                                                                                                                                                                                                                 
<class '__main__.Test'>

In [14]:                                                                                                                                                                                                                                                                                 

In [14]: result = session.execute(session.query(Test).filter(Test.col1 == 'ABC'))                                                                                                                                                                                                        

In [15]: for row in result: 
    ...:         print (type(row)) 
    ...:                                                                                                                                                                                                                                                                                 
<class 'sqlalchemy.engine.result.RowProxy'>

只有键入“sqlalchemy.util._collections.result”才允许在行上使用方法 _asdict()。
因此，如果我在代码中调用 _asdict()，我必须明确检查类型，否则 - 我会得到“AttributeError”异常。

是否可以从炼金术 ORM 查询中只返回一种类型并使用 _asdict() 而不进行额外检查？

Answer 1

第一个例子:

session.query(Test)
# <__main__.Test>

正在使用声明性 orm 的查询生成器。当您保持在它可以处理的范围内时，它最有可能做您想做的事情，例如返回映射类的实例。

第二个例子

cte = session.query(Test).cte()                                                                                                                                                                                                                                                  
session.query(cte)
# <class 'sqlalchemy.util._collections.result'>

转换为 sql

with foo(
    select * from test
)
select * from foo

由于 cte 可以包含任意 sql，因此间接破坏了数据库和 sqlalchemy 对 foo 之间关系的了解。和 test .如果您从单个表中选择所有字段，SQLAlchemy 可以通过显式检查来克服这一问题，但这尚未实现。相反，它返回一个可以处理任何形状数据的泛型类型。

第三种情况使用较低级别的 API session.execute直接地。那是为了从文本中执行 SQL。它将“智能”ORM 查询编译为“愚蠢”的 SQL 字符串，该字符串不知道您的 python 类型是什么，并且再次返回该较低级别 API 的泛型类型。
_asdict() 中的下划线前缀是作者表示这是一个私有(private)方法。这意味着他们可能会随时更改它，恕不另行通知，因此您需要避免它。

您可以添加 .to_dict()方法到您自己的声明性基类，如 in the docs 所述

Here's a reference关于如何实现它

或者，如果你只是在玩，你可以随时调用 .__dict__()