Pet*_*ann 4 python sql orm sqlalchemy
我正在寻找一种方法来告诉sqlalchemy将一些复杂的查询映射到一个自定义类MyResult
而不是默认RowProxy
类.这是一个简单的工作示例
'''
create table foo(id integer, title text);
create table bar(id integer, foo_id integer, name text);
insert into foo values(0, 'null');
insert into foo values(1, 'eins');
insert into bar values(0,0, 'nullnull');
insert into bar values(1,0, 'einsnull');
insert into bar values(2,1, 'zweieins');
'''
Run Code Online (Sandbox Code Playgroud)
和以下代码:
from sqlalchemy import *
from itertools import imap
db = create_engine('sqlite:///test.db')
metadata = MetaData(db)
class MyResult(object):
def __init__(self, id, title, name):
self.id = id
self.title = title
self.name = name
foo = Table('foo', metadata, autoload=True)
bar = Table('bar', metadata, autoload=True)
result = select([foo.c.id, foo.c.title, bar.c.name], foo.c.id == bar.c.foo_id).execute().fetchall()
Run Code Online (Sandbox Code Playgroud)
现在我正在寻找一种方法来告诉sqlalchemy执行从结果行到MyResult的映射.
row = result[0]
print type(row)
#<class 'sqlalchemy.engine.base.RowProxy'>
print row.items()
#[(u'id', 0), (u'title', u'null'), (u'name', u'einsnull')]
Run Code Online (Sandbox Code Playgroud)
我知道我可以通过类似的方式手工绘制
my_result = imap(lambda x: MyResult(**x), result)
Run Code Online (Sandbox Code Playgroud)
但我觉得这不是在sqlalchemy中处理它的方法.
从您的示例中可以看出,返回的Foo将超过1个Foo.id = 0
,这将导致主键的重复值,这反过来只会导致返回结果集的子集.在这种情况下,您可能应该将primary_key
也扩展到其他Bar
列(包括Bar.id
或使用,Bar.name
如果它是唯一的).
然后,您可以使用from_statement
(如使用Literal SQL中所述)来实现此目的:
sql_qry = select([foo.c.id.label("id"),
foo.c.title.label("title"),
bar.c.name.label("name")],
foo.c.id == bar.c.foo_id)
my_result_qry = session.query(MyResult).from_statement(sql_qry)
for x in my_result_qry.all():
print x
Run Code Online (Sandbox Code Playgroud)
但是,MyResult
必须映射模型.您可以将其映射到一些虚拟(不存在)表或视图.label
列的s也很重要,因为它们必须与类的列定义完全匹配(构造函数不会被使用).
通过select
直接调用,您将忽略 ORM 功能。你需要mapper
在你的MyResult
课堂上使用。正如你所拥有的,MyResult
只是一个普通的类。
像这样的东西:
Foo = mapper(MyResult, foo)
Bar = mapper(MyResult, bar) # NOTE: MyResult itself is unchanged by this
session = Session()
# query against the mapper class
result = session.query(Foo).filter(Foo.title == 'xyz').one()
print result.name
Run Code Online (Sandbox Code Playgroud)