SQLAlchemy核心技术：表之间常见列的去重

Question

假设我想创建以下三个表：

a = db.Table(
    'a', metadata,
    db.Column('id', db.Integer, nullable=False, primary_key=True),
    db.Column('created', server_default=func.now()),
    db.Column('updated', server_default=func.now(), onupdate=func.current_timestamp()),

    db.Column('unique_to_a', db.INTEGER, nullable=True)
)

b = db.Table(
    'b', metadata,
    db.Column('id', db.Integer, nullable=False, primary_key=True),
    db.Column('created', server_default=func.now()),
    db.Column('updated', server_default=func.now(), onupdate=func.current_timestamp()),

    db.Column('unique_to_b', db.INTEGER, nullable=True)
)

c = db.Table(
    'c', metadata,
    db.Column('id', db.Integer, nullable=False, primary_key=True),
    db.Column('created', server_default=func.now()),
    db.Column('updated', server_default=func.now(), onupdate=func.current_timestamp()),


    db.Column('a_id', db.Integer, db.ForeignKey('a.id'), index=True, nullable=True),
    db.Column('unique_to_c', db.INTEGER, nullable=True)
)

这三个表之间有共同的字段(id, created, 和 updated)，也有各自独有的字段。

是否有一种简单的方法来创建一个基础表，用于存储所有共有的字段？

例如：

base_table = db.Table(
    'base', metadata,
    db.Column('id', db.Integer, nullable=False, primary_key=True),
    db.Column('created', server_default=func.now()),
    db.Column('updated', server_default=func.now(), onupdate=func.current_timestamp()),
)

a = db.Table(
    'a', metadata,
    base_table,

    db.Column('unique_to_a', db.INTEGER, nullable=True)
)

我们更倾向于在SQLAlchemy Core中完成所有操作（即不使用ORM）。

我们以前使用过deepcopy（from copy import deepcopy），但我们认为在表之间共享列必须有更好的方法。

- Wesley Bowman

---

抱歉，这周很忙，我还没有机会测试它。明天会尝试！ - Wesley Bowman

似乎可以工作，谢谢。 - Wesley Bowman

Answer 1

据我所知，Table 无法继承，但是 declarative 模型可以继承，您可以利用它来创建表对象。

engine = create_engine("sqlite://")
Base = declarative_base()
Base.metadata.bind = engine

class BaseColumn(object):
    id = Column('id', Integer, nullable=False, primary_key=True)
    created = Column('created', server_default=func.now())
    updated = Column('updated', server_default=func.now(), onupdate=func.current_timestamp())

class A(Base, BaseColumn):
    __tablename__ = "a"
    a_column = Column('unique_to_a', INTEGER, nullable=True)


print `A.__table__`

输出结果:

Table('a', MetaData(bind=Engine(sqlite://)), Column('id', Integer(), table=<a>, primary_key=True, nullable=False), Column('created', NullType(), table=<a>, server_default=DefaultClause(<sqlalchemy.sql.functions.now at 0x105bce490; now>, for_update=False)), Column('updated', NullType(), table=<a>, onupdate=ColumnDefault(<sqlalchemy.sql.functions.current_timestamp at 0x105bce750; current_timestamp>), server_default=DefaultClause(<sqlalchemy.sql.functions.now at 0x105bce610; now>, for_update=False)), Column('unique_to_a', INTEGER(), table=<a>), schema=None)

注：此处为it技术相关内容，展示了一个表的结构。

- georgexsh

Answer 2

如果您在不使用deepcopy的情况下使用sqlalchemy core与asyncpgsa，则可以使用以下解决方法：

def base_columns():
    return {
        Column('created_at',
               DateTime(timezone=True),
               server_default=func.clock_timestamp()),
        Column('modified_at',
               DateTime(timezone=True),
               server_default=func.clock_timestamp(),
               onupdate=func.clock_timestamp()),
    }

company_tbl = Table(
    'company', metadata,
    Column('id', BigInteger, primary_key=True),
    Column('name', Text, nullable=False, unique=True),
    *base_columns(),
)

- scottwernervt

---

应该返回元组而不是集合吗？也许只需 return (Column(...), Column(...))。 - kigawas