SQLAlchemy Core 表之间公共列的重复数据删除答案

【问题标题】：SQLAlchemy Core deduplication of common columns between tablesSQLAlchemy Core 表之间公共列的重复数据删除
【发布时间】：2017-10-31 08:07:54
【问题描述】：

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

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)，但我们认为必须有更好的方法在表之间共享列。

【问题讨论】：

对不起，这周很忙，所以我没有机会测试它。明天试试！
似乎工作，谢谢

标签： python postgresql python-3.x sqlalchemy

【解决方案1】：

任何使用asyncpgsa 和sqlalchemy 核心而不使用deepcopy 的人的解决方法：

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(),
)

【讨论】：

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

【解决方案2】：

据我所知，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)

【讨论】：