Python sqlalchemy 模块，CheckConstraint() 实例源码

def _version_table_schema(metadata):
    # NOTE: When modifying this schema, update the ASSET_DB_VERSION value
    return sa.Table(
        'version_info',
        metadata,
        sa.Column(
            'id',
            sa.Integer,
            unique=True,
            nullable=False,
            primary_key=True,
        ),
        sa.Column(
            'version',
            sa.Integer,
            unique=True,
            nullable=False,
        ),
        # This constraint ensures a single entry in this table
        sa.CheckConstraint('id <= 1'),
    )

def User(Base):
    class User(Base):
        __tablename__ = 'user'
        id = sa.Column('_id', sa.Integer, primary_key=True)
        name = sa.Column('_name', sa.String(20))
        age = sa.Column('_age', sa.Integer, nullable=False)
        email = sa.Column(
            '_email', sa.String(200), nullable=False, unique=True
        )
        fav_numbers = sa.Column('_fav_numbers', ARRAY(sa.Integer))

        __table_args__ = (
            sa.CheckConstraint(sa.and_(age >= 0, age <= 150)),
            sa.CheckConstraint(
                sa.and_(
                    sa.func.array_length(fav_numbers, 1) <= 8
                )
            )
        )
    return User

def __table_args__(cls):
        return (sqlalchemy.CheckConstraint('started_at <= ended_at',
                                           name="ck_started_before_ended"),
                COMMON_TABLES_ARGS)

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('problem_label',
        sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
        sa.Column('problem_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False),
        sa.Column('label', sa.Unicode(length=255), nullable=False),
        sa.ForeignKeyConstraint(['problem_id'], ['problem.id'], ),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_table('dataset_label_probability',
        sa.Column('id', sqlalchemy_utils.types.uuid.UUIDType(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
        sa.Column('data_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False),
        sa.Column('label_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=False),
        sa.Column('probability', sa.Float(), nullable=True),
        sa.CheckConstraint('probability >= 0 AND probability <= 1', name='chk_dataset_probability'),
        sa.ForeignKeyConstraint(['data_id'], ['dataset.id'], ondelete='CASCADE'),
        sa.ForeignKeyConstraint(['label_id'], ['problem_label.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )
    op.add_column('label_event', sa.Column('label_id', sqlalchemy_utils.types.uuid.UUIDType(), nullable=True))
    op.add_column('problem', sa.Column('name', sa.Unicode(length=255), nullable=True))

    op.execute('''
        INSERT INTO problem_label(problem_id, label)
        SELECT id, label FROM problem
    ''')
    op.execute('''
        INSERT INTO dataset_label_probability(data_id, label_id, probability)
        SELECT id, (SELECT id FROM problem_label WHERE problem_id = dataset.problem_id), probability FROM dataset
    ''')
    op.execute('''
        UPDATE label_event SET
        label_id = (SELECT id FROM problem_label WHERE label = label_event.label)
    ''')
    op.execute('''
        UPDATE problem SET
        name = (SELECT label FROM problem_label WHERE problem_id = problem.id)
    ''')

    op.alter_column('label_event', 'label_id', nullable=False)
    op.alter_column('problem', 'name', nullable=False)

    op.create_index(op.f('ix_dataset_label_probability_data_id'), 'dataset_label_probability', ['data_id'], unique=False)
    op.create_index(op.f('ix_dataset_label_probability_label_id'), 'dataset_label_probability', ['label_id'], unique=False)
    op.drop_column('dataset', 'probability')
    op.drop_column('problem', 'label')


    op.create_index(op.f('ix_label_event_label_id'), 'label_event', ['label_id'], unique=False)
    op.create_foreign_key(None, 'label_event', 'problem_label', ['label_id'], ['id'], ondelete='CASCADE')
    op.drop_column('label_event', 'label')
    # ### end Alembic commands ###

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###

    op.create_table('profile',
                    sa.Column('profile_name', sa.String(length=128), nullable=False),
                    sa.Column('properties', JSONB(), nullable=False),
                    sa.PrimaryKeyConstraint('profile_name', name=op.f('profile_pkey'))
                    )

    op.create_table('configuration',
                    sa.Column('key', sa.String(length=128), nullable=False),
                    sa.Column('value', sa.String(length=1024), nullable=False),
                    sa.PrimaryKeyConstraint('key', name=op.f('configuration_pkey'))
                    )

    op.create_table('device',
                    sa.Column('device_id', sa.Integer(), nullable=False, autoincrement=True),
                    sa.Column('device_type', sa.String(length=64), nullable=False),
                    sa.Column('properties', JSONB(), nullable=True),
                    sa.Column('hostname', sa.String(length=256), nullable=True),
                    sa.Column('ip_address', sa.String(length=64), nullable=True),
                    sa.Column('mac_address', sa.String(length=64), nullable=True),
                    sa.Column('profile_name', sa.String(length=128), nullable=True),
                    sa.Column('deleted', sa.BOOLEAN(), server_default=false_just_for_sqlalchemy(), nullable=False),
                    sa.PrimaryKeyConstraint('device_id', name=op.f('device_pkey')),
                    sa.ForeignKeyConstraint(['profile_name'], ['profile.profile_name'], name='device_profile',
                                            match='SIMPLE', ondelete='NO ACTION', onupdate='NO ACTION')
                    )

    op.create_table('log',
                    sa.Column('process', sa.String(length=128), nullable=True),
                    sa.Column('timestamp', sa.DateTime(timezone=True), nullable=False, server_default=func.now()),
                    sa.Column('level', sa.Integer(), nullable=False),
                    sa.Column('device_id', sa.Integer(), nullable=True),
                    sa.Column('message', sa.Text(), nullable=False),
                    sa.ForeignKeyConstraint(['device_id'], ['device.device_id'], name='log_process',
                                            match='SIMPLE', ondelete='NO ACTION', onupdate='NO ACTION'),
                    sa.CheckConstraint('level = ANY (ARRAY[0, 10, 15, 20, 30, 40, 50])', name=op.f('valid_log_levels'))
                    )

    creating_functions()
    # ### end Alembic commands ###

def assert_max_length(obj, column, max_length):
    """
    Assert that the given column is of given max length. This function supports
    string typed columns as well as PostgreSQL array typed columns.

    In the following example we add a check constraint that user can have a
    maximum of 5 favorite colors and then test this.::


        class User(Base):
            __tablename__ = 'user'
            id = sa.Column(sa.Integer, primary_key=True)
            favorite_colors = sa.Column(ARRAY(sa.String), nullable=False)
            __table_args__ = (
                sa.CheckConstraint(
                    sa.func.array_length(favorite_colors, 1) <= 5
                )
            )


        user = User(name='John Doe', favorite_colors=['red', 'blue'])
        session.add(user)
        session.commit()


        assert_max_length(user, 'favorite_colors', 5)


    :param obj: SQLAlchemy declarative model object
    :param column: Name of the column
    :param max_length: Maximum length of given column
    """
    type_ = sa.inspect(obj.__class__).columns[column].type
    _expect_successful_update(
        obj,
        column,
        _repeated_value(type_) * max_length,
        _expected_exception(type_)
    )
    _expect_failing_update(
        obj,
        column,
        _repeated_value(type_) * (max_length + 1),
        _expected_exception(type_)
    )