Function build_clock_table

temporal_sqlalchemy/clock.py:198–245

build clock table to be used with a clock model

(entity_table: sa.Table,
                      metadata: sa.MetaData,
                      schema: str,
                      activity_class=None)

def build_clock_table(entity_table: sa.Table,
                      metadata: sa.MetaData,
                      schema: str,
                      activity_class=None) -> sa.Table:
    """ build clock table to be used with a clock model """
    clock_table_name = util.truncate_identifier(
        "%s_clock" % entity_table.name)
    clock_table = sa.Table(
        clock_table_name,
        metadata,
        sa.Column('id',
                  sap.UUID(as_uuid=True),
                  default=uuid.uuid4,
                  primary_key=True),
        sa.Column('tick',
                  sa.Integer,
                  nullable=False,
                  autoincrement=False),
        sa.Column('timestamp',
                  sa.DateTime(True),
                  server_default=sa.func.current_timestamp()),
        schema=schema)

    entity_keys = []
    for fk in util.foreign_key_to(entity_table, nullable=False):
        # this is done to support arbitrary primary key shape on entity
        # We don't add additional indices on the foreign keys here because
        # the uniqueness constraints will add an implicit index.
        clock_table.append_column(fk)
        entity_keys.append(fk.key)

    tick_entity_unique_name = util.truncate_identifier('%s_tick_entity_id_key' % clock_table_name)
    clock_table.append_constraint(
        sa.UniqueConstraint(*(entity_keys + ['tick']), name=tick_entity_unique_name),
    )

    if activity_class:
        activity_keys = []
        # support arbitrary shaped activity primary keys
        for fk in util.foreign_key_to(activity_class.__table__,
                                      prefix='activity',
                                      nullable=False):
            clock_table.append_column(fk)
            activity_keys.append(fk.key)
        # ensure we have DB constraint on clock <> activity uniqueness
        clock_table.append_constraint(sa.UniqueConstraint(*(entity_keys + activity_keys)))

    return clock_table


def build_history_class(