SQLAlchemy 映射连接表的列到一个对象

Question

我有三个表：UserTypeMapper、User 和 SystemAdmin。在我的 get_user 方法中，根据 UserTypeMapper 表里的 is_admin 这一行，我会查询 User 表或者 SystemAdmin 表。这里的 user_id 列对应着 User 和 SystemAdmin 表里的主键 id。

class UserTypeMapper(Base):
    __tablename__ = 'user_type_mapper'

    id = Column(BigInteger, primary_key=True)
    is_admin = Column(Boolean, default=False)
    user_id = Column(BigInteger, nullable=False)

class SystemAdmin(Base):
    __tablename__ = 'system_admin'

    id = Column(BigInteger, primary_key=True)
    name = Column(Unicode)
    email = Column(Unicode)

class User(Base):
    __tablename__ = 'user'

    id = Column(BigInteger, primary_key=True)
    name = Column(Unicode)
    email = Column(Unicode)

我想通过一次查询就能获取任何用户，无论是系统管理员还是普通用户，所以我会根据 is_admin 这一行来做连接查询，要么是 User 表，要么是 SystemAdmin 表。例如：

DBSession.query(UserTypeMapper, SystemAdmin).join(SystemAdmin, UserTypeMapper.user_id==SystemAdmin.id).first()

还有

DBSession.query(UserTypeMapper, User).join(User, UserTypeMapper.user_id==User.id).first()

这样做是没问题的；不过，我想要以这样的方式来访问这些数据：

>>> my_admin_obj.is_admin
True
>>> my_admin_obj.name
Bob Smith

而不是

>>> my_user_obj.is_admin
False
>>> my_user_obj.name
Bob Stevens

目前，我必须明确指定： my_user_obj.UserTypeMapper.is_admin 和 my_user_obj.User.name。根据我所了解的，我需要将这些表进行映射，这样我就不需要指定属性属于哪个表。我的问题是，我不明白在有两个可能的表的情况下，如何指定这个属性，比如 name 是从哪个表来的。

这是我提到的例子： 将一个类映射到多个表

我该如何实现这个呢？谢谢。

Answer 1

你发现了为什么“双重用途外键”是一种反模式。

还有一个相关的问题你没有提到，那就是没有办法用外键约束来强制数据保持在有效状态。你想确保在UserTypeMapper的每一行中都有且只有一个东西，但这个“东西”并不是来自某一张特定的表。正式来说，你想要的是一个函数依赖关系：

user_type_mapper → (system_admin× 1) ∪ (user× 0)

但大多数SQL数据库不允许你写出这样的外键约束。

这看起来很复杂，因为它确实复杂。

那么，让我们考虑一下我们真正想表达的意思：“每个system_admin都应该是一个user；或者

system_admin → user

在SQL中，这样写：

CREATE TABLE user (
    id INTEGER PRIMARY KEY,
    name VARCHAR,
    email VARCHAR
);

CREATE TABLE system_admin (
    user_id INTEGER PRIMARY KEY REFERENCES user(id)
);

或者，用SQLAlchemy的声明式风格：

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

class SystemAdmin(Base):
    __tablename__ = 'system_admin'
    user_id = Column(ForeignKey(User.id), primary_key=True)

这个结构允许我们问什么样的问题呢？

• “有没有叫‘john doe’的系统管理员？”

>>> print session.query(User).join(SystemAdmin).filter(User.name == 'john doe').exists()

EXISTS (SELECT 1 
FROM "user" JOIN system_admin ON "user".id = system_admin.user_id 
WHERE "user".name = :name_1)

• “现在有多少用户？有多少系统管理员？”

>>> print session.query(func.count(User.id), func.count(SystemAdmin.user_id)).outerjoin(SystemAdmin)

SELECT count("user".id) AS count_1, count(system_admin.user_id) AS count_2 
FROM "user" LEFT OUTER JOIN system_admin ON "user".id = system_admin.user_id

---

我希望你能明白，以上的设计比你在问题中描述的设计更可取；但如果你真的没有选择（只有在这种情况下，如果你仍然觉得你现有的设计更好，请进一步完善你的问题），你仍然可以把这些数据塞进一个Python对象里，这样会很难处理，因为你需要提供一个替代的表映射；具体来说，就是一个遵循第一个方程式大致结构的映射。

我们需要提到UserTypeMapper两次，分别对应并集的两边，为此，我们需要给它们起别名。

>>> from sqlalchemy.orm import aliased
>>> utm1 = aliased(UserTypeMapper)
>>> utm2 = aliased(UserTypeMapper)

对于并集的主体，将每个别名连接到相应的表：由于SystemAdmin和User的列是相同的且顺序一致，我们不需要详细描述它们，但如果它们有任何不同，我们需要通过明确提到每一列来使它们“兼容并集”；这部分留给你自己练习。

>>> utm_sa = Query([utm1, SystemAdmin]).join(SystemAdmin, (utm1.user_id == SystemAdmin.id) & (utm1.is_admin == True))
>>> utm_u = Query([utm2, User]).join(User, (utm2.user_id == User.id) & (utm2.is_admin == False))

然后我们把它们连接在一起……

>>> print utm_sa.union(utm_u)
SELECT anon_1.user_type_mapper_1_id AS anon_1_user_type_mapper_1_id, anon_1.user_type_mapper_1_is_admin AS anon_1_user_type_mapper_1_is_admin, anon_1.user_type_mapper_1_user_id AS anon_1_user_type_mapper_1_user_id, anon_1.system_admin_id AS anon_1_system_admin_id, anon_1.system_admin_name AS anon_1_system_admin_name, anon_1.system_admin_email AS anon_1_system_admin_email 
FROM (SELECT user_type_mapper_1.id AS user_type_mapper_1_id, user_type_mapper_1.is_admin AS user_type_mapper_1_is_admin, user_type_mapper_1.user_id AS user_type_mapper_1_user_id, system_admin.id AS system_admin_id, system_admin.name AS system_admin_name, system_admin.email AS system_admin_email 
FROM user_type_mapper AS user_type_mapper_1 JOIN system_admin ON user_type_mapper_1.user_id = system_admin.id AND user_type_mapper_1.is_admin = 1 UNION SELECT user_type_mapper_2.id AS user_type_mapper_2_id, user_type_mapper_2.is_admin AS user_type_mapper_2_is_admin, user_type_mapper_2.user_id AS user_type_mapper_2_user_id, "user".id AS user_id, "user".name AS user_name, "user".email AS user_email 
FROM user_type_mapper AS user_type_mapper_2 JOIN "user" ON user_type_mapper_2.user_id = "user".id AND user_type_mapper_2.is_admin = 0) AS anon_1

虽然理论上可以把这一切封装成一个看起来像标准SQLAlchemy ORM的Python类，但我绝对不会这么做。处理非表映射，尤其是当它们不仅仅是简单的连接（这是一个并集）时，工作量很大而回报却几乎为零。