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()