Python源码示例:sqlalchemy.dialects.postgresql.ENUM
示例1
def upgrade():
blockchain_status = postgresql.ENUM('PENDING', 'SUCCESS', 'FAILED', name='blockchainstatus')
blockchain_status.create(op.get_bind())
op.create_table('worker_messages',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('authorising_user_id', sa.Integer(), nullable=True),
sa.Column('created', sa.DateTime(), nullable=True),
sa.Column('updated', sa.DateTime(), nullable=True),
sa.Column('message', sa.String(), nullable=True),
sa.Column('error', sa.String(), nullable=True),
sa.Column('worker_timestamp', sa.DateTime(), nullable=True),
sa.Column('blockchain_task_uuid', sa.String(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.add_column('credit_transfer', sa.Column('blockchain_hash', sa.String(), nullable=True))
op.add_column('credit_transfer', sa.Column('blockchain_status', sa.Enum('PENDING', 'SUCCESS', 'FAILED', name='blockchainstatus'), nullable=True))
op.add_column('credit_transfer', sa.Column('last_worker_update', sa.DateTime(), nullable=True))
op.add_column('exchange', sa.Column('blockchain_hash', sa.String(), nullable=True))
op.add_column('exchange', sa.Column('blockchain_status', sa.Enum('PENDING', 'SUCCESS', 'FAILED', name='blockchainstatus'), nullable=True))
op.add_column('exchange', sa.Column('last_worker_update', sa.DateTime(), nullable=True))
op.create_index(op.f('ix_worker_messages_table_blockchain_task_uuid'), 'worker_messages', ['blockchain_task_uuid'], unique=False)
示例2
def _inline_enum_script(self):
write_script(
self.script,
self.rid,
"""
revision = '%s'
down_revision = None
from alembic import op
from sqlalchemy.dialects.postgresql import ENUM
from sqlalchemy import Column
def upgrade():
op.create_table("sometable",
Column("data", ENUM("one", "two", "three", name="pgenum"))
)
def downgrade():
op.drop_table("sometable")
"""
% self.rid,
)
示例3
def upgrade():
op.create_table('monthly_billing',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('month', sa.String(), nullable=False),
sa.Column('year', sa.Float(), nullable=False),
sa.Column('notification_type',
postgresql.ENUM('email', 'sms', 'letter', name='notification_type', create_type=False),
nullable=False),
sa.Column('monthly_totals', postgresql.JSON(), nullable=False),
sa.Column('updated_at', sa.DateTime, nullable=False),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_monthly_billing_service_id'), 'monthly_billing', ['service_id'], unique=False)
op.create_index(op.f('uix_monthly_billing'), 'monthly_billing', ['service_id', 'month', 'year', 'notification_type'], unique=True)
示例4
def upgrade():
notification_types = postgresql.ENUM('email', 'sms', 'letter', name='notification_type', create_type=False)
op.create_table('rates',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('valid_from', sa.DateTime(), nullable=False),
sa.Column('rate', sa.Numeric(), nullable=False),
sa.Column('notification_type', notification_types, nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_rates_notification_type'), 'rates', ['notification_type'], unique=False)
op.get_bind()
op.execute("INSERT INTO rates(id, valid_from, rate, notification_type) "
"VALUES('{}', '2016-05-18 00:00:00', 1.65, 'sms')".format(uuid.uuid4()))
op.execute("INSERT INTO rates(id, valid_from, rate, notification_type) "
"VALUES('{}', '2017-04-01 00:00:00', 1.58, 'sms')".format(uuid.uuid4()))
示例5
def upgrade():
op.create_table(
'service_contact_list',
sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('original_file_name', sa.String(), nullable=False),
sa.Column('row_count', sa.Integer(), nullable=False),
sa.Column('template_type', postgresql.ENUM(name='template_type', create_type=False), nullable=False),
sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('created_by_id', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['created_by_id'], ['users.id'], ),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_service_contact_list_created_by_id'), 'service_contact_list', ['created_by_id'], unique=False)
op.create_index(op.f('ix_service_contact_list_service_id'), 'service_contact_list', ['service_id'], unique=False)
op.add_column('jobs', sa.Column('contact_list_id', postgresql.UUID(as_uuid=True), nullable=True))
op.create_foreign_key('jobs_contact_list_id_fkey', 'jobs', 'service_contact_list', ['contact_list_id'], ['id'])
示例6
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('monthly_billing',
sa.Column('id', postgresql.UUID(), autoincrement=False, nullable=False),
sa.Column('service_id', postgresql.UUID(), autoincrement=False, nullable=False),
sa.Column('notification_type', postgresql.ENUM('email', 'sms', 'letter', name='notification_type'), autoincrement=False, nullable=False),
sa.Column('monthly_totals', postgresql.JSON(astext_type=sa.Text()), autoincrement=False, nullable=False),
sa.Column('updated_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
sa.Column('start_date', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
sa.Column('end_date', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(['service_id'], ['services.id'], name='monthly_billing_service_id_fkey'),
sa.PrimaryKeyConstraint('id', name='monthly_billing_pkey'),
sa.UniqueConstraint('service_id', 'start_date', 'notification_type', name='uix_monthly_billing')
)
op.create_index('ix_monthly_billing_service_id', 'monthly_billing', ['service_id'], unique=False)
# ### end Alembic commands ###
示例7
def upgrade():
op.alter_column(
'notification_history',
'status',
existing_type=postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
nullable=True
)
op.alter_column(
'notifications',
'status',
existing_type=postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
nullable=True
)
示例8
def downgrade():
op.alter_column(
'notifications',
'status',
existing_type=postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
nullable=False
)
op.alter_column(
'notification_history',
'status',
existing_type=postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
nullable=False
)
示例9
def downgrade():
op.add_column(
'notifications',
sa.Column(
'status',
postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
autoincrement=False,
nullable=True
)
)
op.add_column(
'notification_history',
sa.Column(
'status',
postgresql.ENUM(
'created', 'sending', 'delivered', 'pending', 'failed', 'technical-failure',
'temporary-failure', 'permanent-failure', 'sent', name='notify_status_type'
),
autoincrement=False,
nullable=True
)
)
示例10
def upgrade():
# Change the name and the type of the column resource_availability
# to visibility
visibility_enum = postgresql.ENUM(*visibility_states,
name='visibility_states')
visibility_enum.create(op.get_bind())
for table_name in resource_tables:
op.alter_column(table_name,
'resource_availability',
new_column_name='visibility',
type_=visibility_enum,
postgresql_using='resource_availability::text::'
'visibility_states')
# Remove the enum resource_availability from postgres
op.execute("DROP TYPE resource_availability;")
示例11
def downgrade():
# Change the name and the type of the column visibility back
# to resource_availability
resource_availability = postgresql.ENUM(*visibility_states,
name='resource_availability')
resource_availability.create(op.get_bind())
for table_name in resource_tables:
op.alter_column(table_name,
'visibility',
new_column_name='resource_availability',
type_=resource_availability,
postgresql_using='visibility::text::'
'resource_availability')
# Remove the enum visibility_states from postgres
op.execute("DROP TYPE visibility_states;")
示例12
def fix_postgres_array_of_enum(connection, tbl):
"Change type of ENUM[] columns to a custom type"
for col in tbl.c:
col_str = str(col.type)
if col_str.endswith('[]'): # this is an array
enum_name = col_str[:-2]
try: # test if 'enum_name' is an enum
enum_ranges = connection.execute('''
SELECT enum_range(NULL::%s);
''' % enum_name).fetchone()
enum_values = sql_enum_to_list(enum_ranges[0])
enum = ENUM(*enum_values, name=enum_name)
tbl.c[col.name].type = ArrayOfEnum(enum)
except sa.exc.ProgrammingError as enum_excep:
if 'does not exist' in str(enum_excep):
pass # Must not have been an enum
else:
raise
示例13
def test_generate_multiple_on_metadata(self):
metadata = self.metadata
e1 = Enum("one", "two", "three", name="myenum", metadata=metadata)
t1 = Table("e1", metadata, Column("c1", e1))
t2 = Table("e2", metadata, Column("c1", e1))
metadata.create_all(checkfirst=False)
assert "myenum" in [e["name"] for e in inspect(testing.db).get_enums()]
metadata.drop_all(checkfirst=False)
assert "myenum" not in [
e["name"] for e in inspect(testing.db).get_enums()
]
e1.create() # creates ENUM
t1.create() # does not create ENUM
t2.create() # does not create ENUM
示例14
def test_array_plus_native_enum_create(self):
m = MetaData()
t = Table(
"t",
m,
Column(
"data_1",
self.ARRAY(postgresql.ENUM("a", "b", "c", name="my_enum_1")),
),
Column(
"data_2",
self.ARRAY(types.Enum("a", "b", "c", name="my_enum_2")),
),
)
t.create(testing.db)
eq_(
set(e["name"] for e in inspect(testing.db).get_enums()),
set(["my_enum_1", "my_enum_2"]),
)
t.drop(testing.db)
eq_(inspect(testing.db).get_enums(), [])
示例15
def test_create_drop_enum(self):
# test escaping and unicode within CREATE TYPE for ENUM
typ = postgresql.ENUM(
"val1", "val2", "val's 3", u("méil"), name="myname"
)
self.assert_compile(
postgresql.CreateEnumType(typ),
u(
"CREATE TYPE myname AS "
"ENUM ('val1', 'val2', 'val''s 3', 'méil')"
),
)
typ = postgresql.ENUM("val1", "val2", "val's 3", name="PleaseQuoteMe")
self.assert_compile(
postgresql.CreateEnumType(typ),
'CREATE TYPE "PleaseQuoteMe" AS ENUM '
"('val1', 'val2', 'val''s 3')",
)
示例16
def test_create_type_schema_translate(self):
e1 = Enum("x", "y", "z", name="somename")
e2 = Enum("x", "y", "z", name="somename", schema="someschema")
schema_translate_map = {None: "foo", "someschema": "bar"}
self.assert_compile(
postgresql.CreateEnumType(e1),
"CREATE TYPE foo.somename AS ENUM ('x', 'y', 'z')",
schema_translate_map=schema_translate_map,
render_schema_translate=True,
)
self.assert_compile(
postgresql.CreateEnumType(e2),
"CREATE TYPE bar.somename AS ENUM ('x', 'y', 'z')",
schema_translate_map=schema_translate_map,
render_schema_translate=True,
)
示例17
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('users', sa.Column('promo_codes', sa.TEXT(), autoincrement=False, nullable=True))
op.add_column('users', sa.Column('plan', postgresql.ENUM('free', 'trial', 'monthly', 'yearly', name='plan_enum'), server_default=sa.text("'free'::plan_enum"), autoincrement=False, nullable=False))
op.add_column('users', sa.Column('plan_expiration', postgresql.TIMESTAMP(), autoincrement=False, nullable=True))
op.drop_column('users', 'trial_expiration')
op.drop_table('subscription')
# ### end Alembic commands ###
示例18
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('kernels', 'status',
existing_type=postgresql.ENUM('PENDING', 'PREPARING', 'BUILDING', 'PULLING', 'RUNNING', 'RESTARTING', 'RESIZING', 'SUSPENDED', 'TERMINATING', 'TERMINATED', 'ERROR', 'CANCELLED', name='kernelstatus'),
nullable=False,
existing_server_default=sa.text("'PENDING'::kernelstatus"))
op.alter_column('kernels', 'type',
existing_type=postgresql.ENUM('INTERACTIVE', 'BATCH', name='sessiontypes'),
nullable=False,
existing_server_default=sa.text("'INTERACTIVE'::sessiontypes"))
op.create_index(op.f('ix_kernels_status_changed'), 'kernels', ['status_changed'], unique=False)
op.create_index('ix_kernels_updated_order', 'kernels', ['created_at', 'terminated_at', 'status_changed'], unique=False)
# ### end Alembic commands ###
示例19
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index('ix_kernels_updated_order', table_name='kernels')
op.drop_index(op.f('ix_kernels_status_changed'), table_name='kernels')
op.alter_column('kernels', 'type',
existing_type=postgresql.ENUM('INTERACTIVE', 'BATCH', name='sessiontypes'),
nullable=True,
existing_server_default=sa.text("'INTERACTIVE'::sessiontypes"))
op.alter_column('kernels', 'status',
existing_type=postgresql.ENUM('PENDING', 'PREPARING', 'BUILDING', 'PULLING', 'RUNNING', 'RESTARTING', 'RESIZING', 'SUSPENDED', 'TERMINATING', 'TERMINATED', 'ERROR', 'CANCELLED', name='kernelstatus'),
nullable=True,
existing_server_default=sa.text("'PENDING'::kernelstatus"))
# ### end Alembic commands ###
示例20
def upgrade():
''' Add the column merge_status to the table pull_requests.
'''
enum = ENUM('NO_CHANGE', 'FFORWARD', 'CONFLICTS', 'MERGE',
name='merge_status_enum', create_type=False)
enum.create(op.get_bind(), checkfirst=False)
op.add_column(
'pull_requests',
sa.Column('merge_status', enum, nullable=True)
)
示例21
def downgrade():
''' Remove the column merge_status from the table pull_requests.
'''
ENUM(name="merge_status_enum").drop(op.get_bind(), checkfirst=False)
op.drop_column('pull_requests', 'merge_status')
示例22
def upgrade():
### commands auto generated by Alembic - please adjust! ###
enum = ENUM('unverified', 'valid', 'rejected', name='nu_item_enum', create_type=False)
enum.create(op.get_bind(), checkfirst=True)
op.execute("ALTER TABLE nu_release_item ALTER COLUMN reviewed DROP DEFAULT")
op.alter_column('nu_release_item', 'reviewed',
existing_type=sa.BOOLEAN(),
type_=enum,
existing_nullable=False,
nullable=False,
server_default=sa.text("'unverified'"),
existing_server_default=sa.text('false'),
postgresql_using="CASE WHEN reviewed = false THEN 'unverified'::nu_item_enum ELSE 'valid'::nu_item_enum END"
)
op.execute("ALTER TABLE nu_release_item ALTER COLUMN reviewed SET DEFAULT 'unverified'")
### end Alembic commands ###
示例23
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.alter_column('nu_release_item', 'reviewed',
existing_type=postgresql.ENUM('unverified', 'valid', 'rejected', name='nu_item_enum'),
type_=sa.BOOLEAN(),
existing_nullable=False,
nullable=False,
existing_server_default=sa.text('false'))
ENUM(name="pgenum").drop(op.get_bind(), checkfirst=True)
### end Alembic commands ###
示例24
def downgrade():
op.drop_column('exchange', 'last_worker_update')
op.drop_column('exchange', 'blockchain_status')
op.drop_column('exchange', 'blockchain_hash')
op.drop_column('credit_transfer', 'last_worker_update')
op.drop_column('credit_transfer', 'blockchain_status')
op.drop_column('credit_transfer', 'blockchain_hash')
op.drop_table('worker_messages')
blockchain_status = postgresql.ENUM('PENDING', 'SUCCESS', 'FAILED', name='blockchainstatus')
blockchain_status.drop(op.get_bind())
示例25
def upgrade():
tokentype = postgresql.ENUM('LIQUID', 'RESERVE', name='tokentype')
tokentype.create(op.get_bind())
op.add_column('token', sa.Column('token_type', sa.Enum('LIQUID', 'RESERVE', name='tokentype'), nullable=True))
示例26
def downgrade():
op.drop_column('token', 'token_type')
tokentype = postgresql.ENUM('LIQUID', 'RESERVE', name='tokentype')
tokentype.drop(op.get_bind())
示例27
def test_fancy_coltypes(metadata):
Table(
"simple_items",
metadata,
Column("enum", postgresql.ENUM("A", "B", name="blah")),
Column("bool", postgresql.BOOLEAN),
Column("number", NUMERIC(10, asdecimal=False)),
)
assert (
generate_code(metadata)
== """\
# coding: utf-8
from sqlalchemy import Boolean, Column, Enum, MetaData, Numeric, Table
metadata = MetaData()
t_simple_items = Table(
'simple_items', metadata,
Column('enum', Enum('A', 'B', name='blah')),
Column('bool', Boolean),
Column('number', Numeric(10, asdecimal=False))
)
"""
)
示例28
def setUp(self):
self.conn = conn = config.db.connect()
with conn.begin():
conn.execute(
text("CREATE TYPE mood AS ENUM ('sad', 'ok', 'happy')")
)
示例29
def _distinct_enum_script(self):
write_script(
self.script,
self.rid,
"""
revision = '%s'
down_revision = None
from alembic import op
from sqlalchemy.dialects.postgresql import ENUM
from sqlalchemy import Column
def upgrade():
enum = ENUM("one", "two", "three", name="pgenum", create_type=False)
enum.create(op.get_bind(), checkfirst=False)
op.create_table("sometable",
Column("data", enum)
)
def downgrade():
op.drop_table("sometable")
ENUM(name="pgenum").drop(op.get_bind(), checkfirst=False)
"""
% self.rid,
)
示例30
def test_offline_inline_enum_create(self):
self._inline_enum_script()
with capture_context_buffer() as buf:
command.upgrade(self.cfg, self.rid, sql=True)
assert (
"CREATE TYPE pgenum AS "
"ENUM ('one', 'two', 'three')" in buf.getvalue()
)
assert "CREATE TABLE sometable (\n data pgenum\n)" in buf.getvalue()