Python sqlalchemy 模块，CheckConstraint() 实例源码
我们从Python开源项目中，提取了以下6个代码示例，用于说明如何使用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',
# 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', 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',
sa.Column('label', sa.Unicode(length=255),
sa.ForeignKeyConstraint(['problem_id'], ['problem.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('dataset_label_probability',
sa.Column('data_id',
sa.Column('label_id',
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'],
sa.PrimaryKeyConstraint('id')
)
op.add_column('label_event', sa.Column('label_id', nullable=True))
op.add_column('problem', sa.Column('name', 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'), ['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', unique=False)
op.create_foreign_key(None, 'problem_label', ['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),
sa.Column('properties', JSONB(),
sa.PrimaryKeyConstraint('profile_name', name=op.f('profile_pkey'))
)
op.create_table('configuration',
sa.Column('key',
sa.Column('value', sa.String(length=1024),
sa.PrimaryKeyConstraint('key', name=op.f('configuration_pkey'))
)
op.create_table('device',
sa.Column('device_id', sa.Integer(), autoincrement=True),
sa.Column('device_type', sa.String(length=64),
sa.Column('hostname', sa.String(length=256),
sa.Column('ip_address',
sa.Column('mac_address',
sa.Column('deleted', sa.BOOLEAN(), server_default=false_just_for_sqlalchemy(),
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.Column('timestamp', sa.DateTime(timezone=True), server_default=func.Now()),
sa.Column('level',
sa.Column('message', sa.Text(),
sa.ForeignKeyConstraint(['device_id'], ['device.device_id'], name='log_process', 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,
_repeated_value(type_) * (max_length + 1),
_expected_exception(type_)
)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
