SQLAlchemy：从连接多个表中获取单个对象

Question

DB - SQLAlchemy 设置

考虑两个表的经典设置 - user 和 api_key，由 SQLAlchemy 对象表示为：

class User(Base):
    __tablename__ = 'user'
    user_id = Column(String)
    user_name = Column(String)
    vioozer_api_key = Column(String, ForeignKey("api_key.api_key"))

class ApiKey(Base):
    __tablename__ = 'api_key'    
    api_key = Column(String(37), primary_key=True)

为清楚起见，省略了其他字段。

我的查询

假设我想获取特定用户 id 的用户名和 api 密钥：

user, api_key = database.db_session.query(User, ApiKey)\
    .join(ApiKey, User.vioozer_api_key==ApiKey.api_key)\
    .filter(User.user_id=='user_00000000000000000000000000000000').first()

我得到两个对象 - user 和 api_key，我可以从中获取 user.name 和 api_key.api_key。

问题

我想用一个函数包装这个调用，该函数将返回一个对象，其字段将是 user 字段和 api_key 字段的联合 - 与 SQL join 返回表的方式相同两个表的列都被连接。 有没有办法自动获取一个对象，其字段是两个表的字段的并集？

我尝试了什么

我可以为每个 Join 操作定义一个mapper class，但我想知道映射是否可以自动完成。

Answer 1

我很惊讶在这方面的讨论如此之少...我需要连接两个表并返回一个对象，该对象包含两个表中的所有列。我希望这是一个非常常见的用例 - 无论如何，这是我经过数小时寻找但未能找到好方法后所做的黑客攻击

from sqlalchemy import inspect
def obj_to_dict(obj):
    return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}
def flatten_join(tup_list):
    return [{**obj_to_dict(a), **obj_to_dict(b)} for a,b in tup_list]

这是它的一个例子......

########################## SETUP ##########################
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, ForeignKey, Text
from sqlalchemy.ext.declarative import declarative_base

DeclarativeBase = declarative_base()


class Base(DeclarativeBase):
    __abstract__ = True

    def __repr__(self) -> str:
        items = []
        for key in self.__table__._columns._data.keys():
            val = self.__getattribute__(key)
            items.append(f'{key}={val}')
        key_vals = ' '.join(items)
        name = self.__class__.__name__
        return f"<{name}({key_vals})>"


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(Text)


class Todo(Base):
    __tablename__ = "todo"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(Text)
    user_id = Column(Integer, ForeignKey("user.id"))


engine = create_engine("sqlite:///:memory:")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
db = SessionLocal()
Base.metadata.create_all(bind=engine)

########################## DEMO ##########################

db.add(User(**{"id": 1, "username": "jefmason"}))
db.add(Todo(**{"id": 1, "title": "Make coffee", "user_id": 1}))
db.add(Todo(**{"id": 2, "title": "Learn SQLAlchemy", "user_id": 1}))
db.commit()

result = db.query(Todo, User).join(User).all()

print(result)
# > [(<Todo(id=1 title=Make coffee user_id=1)>, <User(id=1 username=jefmason)>), (<Todo(id=2 title=Learn SQLAlchemy user_id=1)>, <User(id=1 username=jefmason)>)]

print(flatten_join(result))
# > [{'id': 1, 'title': 'Make coffee', 'user_id': 1, 'username': 'jefmason'}, {'id': 1, 'title': 'Learn SQLAlchemy', 'user_id': 1, 'username': 'jefmason'}]

Answer 2

不是查询对象，而是查询字段列表，在这种情况下，SQLAlchemy 返回KeyedTuple 的实例，它提供了KeyedTuple._asdict() 方法，您可以使用它来返回任意字典：

def my_function(user_id):
    row =  database.db_session.query(User.name, ApiKey.api_key)\
        .join(ApiKey, User.vioozer_api_key==ApiKey.api_key)\
        .filter(User.user_id==user_id).first()
    return row._asdict()

my_data = my_function('user_00000000000000000000000000000000')

但是对于您的特定查询，您甚至不需要加入 ApiKey，因为 api_key 字段存在于 User 表中：

row = database.db_session.query(User.name, User.api_key)\
    .filter(User.user_id==user_id).first()

Answer 3

我会在 User 上添加一个 api_key 关系：

class User(Base):
    __tablename__ = 'user'
    user_id = Column(String)
    user_name = Column(String)
    vioozer_api_key = Column(String, ForeignKey("api_key.api_key"))
    api_key = Relationship('ApiKey', uselist=False)

然后你可以这样查询：

>>> user = database.db_session.query(User)\
      .filter(User.user_id=='user_00000000000000000000000000000000').first()
>>> user.api_key
<ApiKey>