def print_create_table(tables):
app.config.from_object('config.default')
database.init_app(app)
engine = database.session.get_bind()
for class_name in tables:
cls = get_class(class_name)
for c in cls.__table__.columns:
if not isinstance(c.type, Enum):
continue
t = c.type
sql = str(CreateEnumType(t).compile(engine))
click.echo(sql.strip() + ';')
for index in cls.__table__.indexes:
sql = str(CreateIndex(index).compile(engine))
click.echo(sql.strip() + ';')
sql = str(CreateTable(cls.__table__).compile(engine))
click.echo(sql.strip() + ';')
python类Enum()的实例源码
def test_querying_table(metadata):
"""
Create an object for test table.
"""
# When using pytest-xdist, we don't want concurrent table creations
# across test processes so we assign a unique name for table based on
# the current worker id.
worker_id = os.environ.get('PYTEST_XDIST_WORKER', 'master')
return Table(
'test_querying_table_' + worker_id, metadata,
Column('id', types.Integer, autoincrement=True, primary_key=True),
Column('t_string', types.String(60)),
Column('t_list', types.ARRAY(types.String(60))),
Column('t_enum', types.Enum(MyEnum)),
Column('t_int_enum', types.Enum(MyIntEnum)),
Column('t_datetime', types.DateTime()),
Column('t_date', types.DateTime()),
Column('t_interval', types.Interval()),
Column('uniq_uuid', PG_UUID, nullable=False, unique=True, default=uuid4),
)
def __init__(self, enum, name):
self.enum = enum
self.name = name
members = (member._value_ for member in enum)
kwargs = {'name': name}
self.impl = SQLAlchemyEnum(*members, **kwargs)
def __init__(self, enum):
self.enum = enum
self.impl = Enum(
*enum.values(),
name="ck%s" % re.sub(
'([A-Z])',
lambda m: "_" + m.group(1).lower(),
enum.__name__)
)
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def sqltype_to_stdtype(sqltype):
import sqlalchemy.types as sqltypes
if isinstance(sqltype, (sqltypes.VARCHAR, sqltypes.CHAR, sqltypes.TEXT, sqltypes.Enum, sqltypes.String)):
return _STRING_TYPE
if isinstance(sqltype, (sqltypes.DATETIME, sqltypes.DATE, sqltypes.TIME, sqltypes.TIMESTAMP)):
return _DATE_TYPE
if isinstance(sqltype, (sqltypes.INTEGER, sqltypes.BIGINT, sqltypes.SMALLINT, sqltypes.Integer)):
return _INTEGER_TYPE
if isinstance(sqltype, (sqltypes.REAL, sqltypes.DECIMAL, sqltypes.NUMERIC, sqltypes.FLOAT)):
return _DECIMAL_TYPE
if isinstance(sqltype, sqltypes.BOOLEAN):
return _BOOLEAN_TYPE
def __init__(self, enum, name=None, default=None):
self.enum = enum
if name is None:
self.impl = Enum(*enum.values(), name='RUCIO_ENUM_' + str(uuid.uuid4())[:6])
else:
self.impl = Enum(*enum.values(), name=name)
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def bind_processor(self, dialect):
enum_proc = sqltypes.Enum.bind_processor(self, dialect)
if util.py2k:
unicode_proc = _NativeUnicodeMixin.bind_processor(self, dialect)
else:
unicode_proc = None
def process(value):
raw_str = enum_proc(value)
if unicode_proc:
raw_str = unicode_proc(raw_str)
return raw_str
return process
def __init__(self, enum, **kwargs):
super(DeclEnumType, self).__init__(**kwargs)
self.enum = enum
self.impl = Enum(
*list(enum.values()),
name="ck%s" % re.sub(
'([A-Z])',
lambda m:"_" + m.group(1).lower(),
enum.__name__)
)
