clock.py 文件源码

def build_history_table(
        cls: declarative.DeclarativeMeta,
        prop: T_PROPS,
        schema: str = None) -> sa.Table:
    """build a sql alchemy table for given prop"""
    if isinstance(prop, orm.RelationshipProperty):
        columns = [util.copy_column(column) for column in prop.local_columns]
    else:
        columns = [util.copy_column(column) for column in prop.columns]

    local_table = cls.__table__
    table_name = util.truncate_identifier(
        _generate_history_table_name(local_table, columns)
    )
    # Build the foreign key(s), specifically adding an index since we may use
    # a casted foreign key in our constraints. See _exclusion_in_uuid
    entity_foreign_keys = list(util.foreign_key_to(local_table, index=True))
    entity_constraints = [
        _exclusion_in(fk.type, fk.key)
        for fk in entity_foreign_keys
    ]

    constraints = [
        sa.Index(
            util.truncate_identifier('%s_effective_idx' % table_name),
            'effective',
            postgresql_using='gist'
        ),
        sap.ExcludeConstraint(
            *itertools.chain(entity_constraints, [('vclock', '&&')]),
            name=util.truncate_identifier('%s_excl_vclock' % table_name)
        ),
        sap.ExcludeConstraint(
            *itertools.chain(entity_constraints, [('effective', '&&')]),
            name=util.truncate_identifier('%s_excl_effective' % table_name)
        ),
    ]

    return sa.Table(
        table_name,
        local_table.metadata,
        sa.Column('id',
                  sap.UUID(as_uuid=True),
                  default=uuid.uuid4,
                  primary_key=True),
        sa.Column('effective',
                  sap.TSTZRANGE,
                  default=util.effective_now,
                  nullable=False),
        sa.Column('vclock', sap.INT4RANGE, nullable=False),
        *itertools.chain(entity_foreign_keys, columns, constraints),
        schema=schema or local_table.schema,
        keep_existing=True
    )  # memoization ftw