def upgrade():
"""Upgrade the database to a newer revision."""
# commands auto generated by Alembic - please adjust! ###
op.create_table('component_gh_usage',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('version', sa.String(length=255), nullable=False),
sa.Column('count', sa.Integer(), nullable=False),
sa.Column('ecosystem_backend', postgresql.ENUM('none', 'npm', 'maven', 'pypi',
'rubygems', 'scm', 'crates',
name='ecosystem_backend_enum',
create_type=False),
nullable=True),
sa.Column('timestamp', sa.DateTime(), server_default=sa.text('LOCALTIMESTAMP'),
nullable=False),
sa.PrimaryKeyConstraint('id'))
# end Alembic commands ###
python类ENUM的实例源码
b4fbc8e04c81_gh_usage_with_versions.py 文件源码
项目：fabric8-analytics-server
作者: fabric8-analytics
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
b4fbc8e04c81_gh_usage_with_versions.py 文件源码
项目：fabric8-analytics-worker
作者: fabric8-analytics
项目源码
文件源码
阅读 26
收藏 0
点赞 0
评论 0
def upgrade():
"""Upgrade the database to a newer revision."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('component_gh_usage',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('version', sa.String(length=255), nullable=False),
sa.Column('count', sa.Integer(), nullable=False),
sa.Column('ecosystem_backend', postgresql.ENUM('none', 'npm', 'maven', 'pypi',
'rubygems', 'scm', 'crates',
name='ecosystem_backend_enum',
create_type=False),
nullable=True),
sa.Column('timestamp', sa.DateTime(), server_default=sa.text('LOCALTIMESTAMP'),
nullable=False), sa.PrimaryKeyConstraint('id'))
# ### end Alembic commands ###
def create(op, name, *values):
enum = ENUM(*values, name=name, create_type=False)
enum.create(op.get_bind(), checkfirst=True)
return enum
def existing(name):
return ENUM(name=name, create_type=False)
def upgrade():
### commands auto generated by Alembic - please adjust! ###
semester = postgresql.ENUM('Fall', 'Spring', 'Neither', name='co_op_enum')
semester.create(op.get_bind())
op.add_column('current_coops', sa.Column('semester', sa.Enum('Fall', 'Spring', 'Neither', name='co_op_enum'), server_default='Neither', nullable=False))
op.drop_column('current_coops', 'active')
### end Alembic commands ###
d616b0a80feb_add_classification_type.py 文件源码
项目：ml-annotate
作者: planbrothers
项目源码
文件源码
阅读 29
收藏 0
点赞 0
评论 0
def upgrade():
classification_type_enum = postgresql.ENUM('binary', 'multi-label', 'multi-class', name='classification_type_enum')
classification_type_enum.create(op.get_bind())
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('problem', sa.Column('classification_type', sa.Enum('binary', 'multi-label', 'multi-class', name='classification_type_enum'), server_default='binary', nullable=True))
op.execute('''
UPDATE "problem" SET classification_type = 'multi-class' WHERE (SELECT COUNT(*) FROM "problem_label" WHERE problem_id = problem.id) >= 2
''')
# ### end Alembic commands ###
d616b0a80feb_add_classification_type.py 文件源码
项目：ml-annotate
作者: planbrothers
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('problem', 'classification_type')
classification_type_enum = postgresql.ENUM('binary', 'multi-label', 'multi-class', name='classification_type_enum')
classification_type_enum.drop(op.get_bind())
# ### end Alembic commands ###
b50c705fea80_migrate_to_new_rules_architecture.py 文件源码
项目：doorman
作者: mwielgoszewski
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column('rule', sa.Column('config', postgresql.JSONB(), autoincrement=False, nullable=True))
op.add_column('rule', sa.Column('type', sa.VARCHAR(), autoincrement=False, nullable=False))
op.add_column('rule', sa.Column('action', postgresql.ENUM(u'added', u'removed', u'both', name='rule_actions'), autoincrement=False, nullable=False))
op.drop_column('rule', 'description')
op.drop_column('rule', 'conditions')
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_table('rule')
### end Alembic commands ###
postgresql.ENUM(name='rule_actions').drop(op.get_bind(), checkfirst=False)
f8bb0efac483_nuget_ecosystem.py 文件源码
项目：fabric8-analytics-worker
作者: fabric8-analytics
项目源码
文件源码
阅读 29
收藏 0
点赞 0
评论 0
def downgrade():
"""Downgrade the database to an older revision."""
# ### commands auto generated by Alembic - please adjust! ###
conn = op.get_bind()
conn.execute("DELETE FROM ecosystems WHERE name = 'nuget'")
# There's no 'ALTER TYPE enum REMOVE VALUE'
op.alter_column('package_gh_usage', 'ecosystem_backend',
existing_type=postgresql.ENUM('none', 'npm', 'maven', 'pypi', 'rubygems',
'scm', 'crates', 'nuget',
name='ecosystem_backend_enum'),
type_=postgresql.ENUM('none', 'npm', 'maven', 'pypi', 'rubygems', 'scm',
'crates', name='ecosystem_backend_enum'),
existing_nullable=True)
op.alter_column('ecosystems', '_backend',
existing_type=sa.Enum('none', 'npm', 'maven', 'pypi', 'rubygems', 'scm',
'crates', 'nuget', name='ecosystem_backend_enum'),
type_=sa.Enum('none', 'npm', 'maven', 'pypi', 'rubygems', 'scm', 'crates',
name='ecosystem_backend_enum'),
existing_nullable=True)
op.alter_column('component_gh_usage', 'ecosystem_backend',
existing_type=postgresql.ENUM('none', 'npm', 'maven', 'pypi', 'rubygems',
'scm', 'crates', 'nuget',
name='ecosystem_backend_enum'),
type_=postgresql.ENUM('none', 'npm', 'maven', 'pypi', 'rubygems', 'scm',
'crates', name='ecosystem_backend_enum'),
existing_nullable=True)
# ### end Alembic commands ###
98860d4e5dfb_add_type_and_weight_to_consent_model.py 文件源码
项目：radar
作者: renalreg
项目源码
文件源码
阅读 32
收藏 0
点赞 0
评论 0
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
consent_type_type = postgresql.ENUM('FORM', 'INFORMATION_SHEET', name='consent_type_type')
consent_type_type.create(op.get_bind())
op.add_column('consents', sa.Column('consent_type', sa.Enum('FORM', 'INFORMATION_SHEET', name="consent_type_type"), nullable=True))
consent_type = sa.table('consents', sa.column('consent_type'))
op.execute(consent_type.update().values(consent_type='FORM'))
op.alter_column('consents', 'consent_type', nullable=False)
op.add_column('consents', sa.Column('weight', sa.Integer(), nullable=True))
# ### end Alembic commands ###
def enum_column(enum_cls, name, metadata):
class EnumColumn(_BaseEnumColumn):
pass
EnumColumn.enum_cls = enum_cls
# Just as tables do, enum columns require metadata to bind to.
EnumColumn.impl = ENUM(*(e.value for e in enum_cls), name=name, metadata=metadata)
EnumColumn.reverse_enum_values = {e.value: e for e in enum_cls}
EnumColumn.__name__ = str('EnumColumn_' + name)
return EnumColumn()
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import config, util
from sqlalchemy.testing.exclusions import against
from sqlalchemy import schema, inspect
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
util.drop_all_tables(e, inspector)
if config.requirements.schemas.enabled_for_config(cfg):
util.drop_all_tables(e, inspector, schema=cfg.test_schema)
if against(cfg, "postgresql"):
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
def _prep_testing_database(options, file_config):
from alembic.testing import config
from alembic.testing.exclusions import against
from sqlalchemy import schema
from alembic import util
if util.sqla_08:
from sqlalchemy import inspect
else:
from sqlalchemy.engine.reflection import Inspector
inspect = Inspector.from_engine
if options.dropfirst:
for cfg in config.Config.all_configs():
e = cfg.db
inspector = inspect(e)
try:
view_names = inspector.get_view_names()
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
try:
view_names = inspector.get_view_names(
schema="test_schema")
except NotImplementedError:
pass
else:
for vname in view_names:
e.execute(schema._DropView(
schema.Table(vname, schema.MetaData(),
schema="test_schema")
))
for tname in reversed(inspector.get_table_names(
order_by="foreign_key")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData())
))
if config.requirements.schemas.enabled_for_config(cfg):
for tname in reversed(inspector.get_table_names(
order_by="foreign_key", schema="test_schema")):
e.execute(schema.DropTable(
schema.Table(tname, schema.MetaData(),
schema="test_schema")
))
if against(cfg, "postgresql") and util.sqla_100:
from sqlalchemy.dialects import postgresql
for enum in inspector.get_enums("*"):
e.execute(postgresql.DropEnumType(
postgresql.ENUM(
name=enum['name'],
schema=enum['schema'])))
