SQLAlchemy - 从自动加载表的内部连接映射列的子集

Posted 2023-04-12

【中文标题】SQLAlchemy - 从自动加载表的内部连接映射列的子集

【英文标题】：SQLAlchemy - Mapping subset of columns from inner join of autoloaded tables

【发布时间】：2011-10-07 16:23:55

【问题描述】：

我刚刚开始使用 SQLAlchemy 来处理现有的 Postgres 数据库。我想做的是自动加载两个表，执行内部连接并将列的子集映射到对象。为了完成除最后一部分之外的所有内容，我编写了以下代码：

from sqlalchemy.orm import mapper, sessionmaker
from sqlalchemy.sql import join, select

engine = create_engine("postgresql://<username>:@localhost/<DBname>")
metadata = Metadata(engine)
Session = sessionmaker(engine)
profiles = Table('userprofile',metadata,autoload=True)
profilefields = Table('profilefield',metadata,autoload=True)
class DBObj(object):
     pass
j = join(profiles,profilefields,profiles.c.fieldid==profilefields.c.fieldid)
mapper(DBObj,j,properties='fieldid':[profiles.c.fieldid,profilefields.c.fieldid])
q = session.query(DBObj).all()

我应该如何修改此代码以断言我只希望某些指定的列映射到 DBObj？我进行了广泛的搜索，似乎找不到如何完成此任务的示例。

【参考方案1】：

您需要将select() 映射到所需的列。映射一个join() 会导致sqlalchemy 把它变成对应的select * from <join>：

（跳过此块：这只是创建测试夹具）

>>> from sqlalchemy.orm import mapper, sessionmaker
>>> from sqlalchemy.sql import join, select
>>> from sqlalchemy import *
>>> # engine = create_engine("postgresql://<username>:@localhost/<DBname>")
... engine = create_engine("sqlite:///:memory:")
>>> engine.execute(r"""
...     CREATE TABLE userprofile (
...         id integer primary key,
...         fieldid integer,
...         keep integer,
...         discard integer
...     )
... """)
<sqlalchemy.engine.base.ResultProxy object at 0x2390d90>
>>> engine.execute(r"""
...     CREATE TABLE profilefield (
...         id integer primary key,
...         fieldid integer,
...         keep integer,
...         discard integer
...     )
... """)
<sqlalchemy.engine.base.ResultProxy object at 0x2390e90>
>>> metadata = MetaData(engine)
>>> Session = sessionmaker(engine)
>>> profiles = Table('userprofile',metadata,autoload=True)
>>> profilefields = Table('profilefield',metadata,autoload=True)
>>> engine.execute(profiles.insert('fieldid': 1, 'keep': 2, 'discard': 3))
<sqlalchemy.engine.base.ResultProxy object at 0x23e9750>
>>> engine.execute(profilefields.insert('fieldid': 1, 'keep': 2, 'discard': 3))
<sqlalchemy.engine.base.ResultProxy object at 0x23e98d0>
>>> class DBObj(object):
...      def __repr__(self):
...         return "DBobj" + str(self.__dict__)
... 

>>> j = join(profiles,
...          profilefields,
...          profiles.c.fieldid==profilefields.c.fieldid)

这是新的部分：

>>> js = select([profiles.c.id,
...              profiles.c.keep,
...              profilefields.c.keep,
...              profiles.c.fieldid,
...              profilefields.c.fieldid],
...             use_labels=True,
...             from_obj=j)
>>> mapper(DBObj,alias(js),properties='fieldid':[profiles.c.fieldid,profilefields.c.fieldid])
<Mapper at 0x23e9ad0; DBObj>
>>> q = Session().query(DBObj).all()
>>> q
[DBobju'userprofile_keep': 2, u'userprofile_id': 1, '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x23f08d0>, 'fieldid': 1, u'profilefield_keep': 2]

【讨论】：

非常感谢。这确实很有帮助。