SQLAlchemy - 将一个类映射到多个表，其中一个是只读的

Question

我有一个由“foobar”和“foobar_data_cache”表表示的 foobar 概念是数据库，而“Foobar”类是 python。 “foobar”表表示有关概念的一些数据，而“foobar_data_cache”表表示有关概念的一些其他数据，这些数据源自数据库中的许多其他信息并由数据库触发器计算。为了保持一致性，从 'foobar_data_cache' 表中撤消了 INSERT、UPDATE 和 DELETE 权限。

使用 SQLAlchemy，我想通过连接将 'Foobar' 类映射到两个表 'foobar' 和 'foobar_data_cache'。没有理由使用另一个类来表示“foobar_data_cache”表中的数据并在这两个类之间建立关系，因为这两个表中的数据是强相关的。确实，从数据库的角度来看，两张表之间存在一对一的关系，由以下方式保证：
- 'foobar_data_cache' 的主键也是一个引用 'foobar' 主键的外键
- 以及一个检查 'foobar' 中的每一行在 'foobar_data_cache' 中是否有对应行的触发器

我的问题是，当我尝试使用 SQLAlchemy ORM 持久化一个新的 Foobar 对象时，它会尝试为 'foobar_data_cache' 表插入一些行，我想阻止这种情况。

那么，是否有可能以将“foobar_data_cache”表视为只读的方式配置 SQLAlchemy？如果是，如何？

这是解释我的问题的代码：

from sqlalchemy import (
    Table,
    Column,
    Integer,
    String,
    ForeignKeyConstraint,
    join,
    create_engine,
    )
from sqlalchemy.orm import (
    column_property,
    sessionmaker,
    )
from sqlalchemy.schema import (
    FetchedValue,
    )
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

_foobar_table = Table('foobar', Base.metadata,
        Column('id', Integer, primary_key=True),
        Column('some_data', String),
    )

_foobar_data_cache_table = Table('foobar_data_cache', Base.metadata,
        Column('foobar_id', Integer, primary_key=True),
        Column('computed_data', String, server_default=FetchedValue()),
        ForeignKeyConstraint(['foobar_id'], ['foobar.id']),
    )

class Foobar(Base):
    __table__ = _foobar_table.join(_foobar_data_cache_table)
    _id = column_property(_foobar_table.c.id, _foobar_data_cache_table.c.foobar_id)

    def __init__(self, some_data):
        self.some_data = some_data

if __name__ == '__main__':
    engine = create_engine('postgresql://tester@localhost:5432/mytestdb')
    Session = sessionmaker(bind=engine)
    session = Session()

    my_foobar = Foobar('Dummy data')
    session.add(my_foobar)
    session.commit()

创建两张表的SQL命令如下：

CREATE TABLE foobar (
  id int NOT NULL DEFAULT -2147483648,
  some_data varchar NOT NULL,
  CONSTRAINT pk_foobar PRIMARY KEY (id)
);

CREATE TABLE foobar_data_cache (
  foobar_id int NOT NULL,
  computed_data varchar NOT NULL,
  CONSTRAINT pk_foobar_data_cache PRIMARY KEY (foobar_id),
  CONSTRAINT fk_foobar_data_cache_foobar_1 FOREIGN KEY (foobar_id)
    REFERENCES foobar (id) MATCH FULL
    ON UPDATE CASCADE ON DELETE CASCADE
);  

注意：
考虑到它们之间是一对一的关系，有些人可能想知道为什么我将数据拆分到两个不同的表中。通过在计算列上使用单个表和 FetchedValue 构造（请参阅How to persist only a subset of the mapped columns on SQLAlchemy?）可以轻松解决该问题。 嗯，这有点复杂，但我会试着解释一下。首先，上面没有解释的其他事情是：
- 我使用 PostgreSQL 8.4，其中不能有可延迟的 UNIQUE 约束
- 我的所有列都不接受 NULL 值
- 'foobar_data_cache' 中的某些列具有唯一约束（不可延迟）
- 计算“foobar_data_cache”中数据的触发器被推迟到事务结束。实际上，这是因为由于外键约束，它从其他表中获取只能在插入 'foobar' 之后插入的信息。

也就是说，在使用单个表的情况下，这意味着由于 NOT-NULL 约束，我必须为我的计算列使用临时虚拟值。我的触发器最终会在事务结束时覆盖它。问题在于并发性。实际上，新事务 Tx 在另一个事务 T1 正在执行时尝试插入新的 'foobar' 会失败，因为唯一列的默认虚拟值已经存在于与正在执行的事务 T1 对应的行中。我可以为具有唯一约束的列生成随机虚拟值，但我不喜欢这种做事风格。

Answer 1

如果没有可用表的主键，ORM 将跳过任何 INSERT/UPDATE/DELETE。在这种情况下exclude_properties 将实现这一点，还要注意复合“PK”列消失了，因为我们不关心“foobar_id”的值：

from sqlalchemy import (
    Table,
    Column,
    Integer,
    String,
    ForeignKeyConstraint,
    join,
    create_engine,
    )
from sqlalchemy.orm import (
    column_property,
    sessionmaker,
    )
from sqlalchemy.schema import (
    FetchedValue,
    )
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

_foobar_table = Table('foobar', Base.metadata,
        Column('id', Integer, primary_key=True),
        Column('some_data', String),
    )

_foobar_data_cache_table = Table('foobar_data_cache', Base.metadata,
        Column('foobar_id', Integer, primary_key=True),
        Column('computed_data', String, server_default=FetchedValue()),
        ForeignKeyConstraint(['foobar_id'], ['foobar.id']),
    )

class Foobar(Base):
    __table__ = _foobar_table.join(_foobar_data_cache_table)
    _id = _foobar_table.c.id

    def __init__(self, some_data):
        self.some_data = some_data

    __mapper_args__ = {"exclude_properties": [_foobar_data_cache_table.c.foobar_id]}

if __name__ == '__main__':
    engine = create_engine('postgresql://scott:tiger@localhost/test', echo=True)

    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    Session = sessionmaker(bind=engine)
    session = Session()

    my_foobar = Foobar('Dummy data')
    session.add(my_foobar)

    # simulate your trigger...
    session.flush()
    assert session.scalar(_foobar_data_cache_table.count()) == 0
    session.execute(
        _foobar_data_cache_table.insert(),
        params=dict(
                foobar_id=my_foobar._id,
                computed_data="some computed data"
                )
    )

    session.commit()

    obj = session.query(Foobar).first()
    assert obj.computed_data == "some computed data"

也就是说，如果您将传统的relationship() 用于_foobar 和_foobar_data_cache_table 之间的链接，而不是组合成一个映射的join()，那么整个映射方式会更简单。