def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.add_column('comments', sa.Column('goodsID', sa.Integer(), nullable=True))
op.alter_column('comments', 'commentatorID',
existing_type=mysql.INTEGER(display_width=11),
nullable=True)
op.drop_constraint('comments_ibfk_2', 'comments', type_='foreignkey')
op.create_foreign_key(None, 'comments', 'goods', ['goodsID'], ['goodID'])
op.drop_column('comments', 'goodID')
op.add_column('goods', sa.Column('address', sa.String(length=128), nullable=False))
op.add_column('goods', sa.Column('poster', sa.String(length=128), nullable=False))
op.add_column('goods', sa.Column('times', sa.Integer(), nullable=True))
op.alter_column('orders', 'createDate',
existing_type=mysql.DATETIME(),
nullable=False)
### end Alembic commands ###
python类DATETIME的实例源码
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.alter_column('orders', 'createDate',
existing_type=mysql.DATETIME(),
nullable=True)
op.drop_column('goods', 'times')
op.drop_column('goods', 'poster')
op.drop_column('goods', 'address')
op.add_column('comments', sa.Column('goodID', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False))
op.drop_constraint(None, 'comments', type_='foreignkey')
op.create_foreign_key('comments_ibfk_2', 'comments', 'goods', ['goodID'], ['goodID'])
op.alter_column('comments', 'commentatorID',
existing_type=mysql.INTEGER(display_width=11),
nullable=False)
op.drop_column('comments', 'goodsID')
### end Alembic commands ###
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('dataset', 'collection_date',
existing_type=mysql.DATETIME(),
type_=sa.Text(),
existing_nullable=True)
op.alter_column('dataset', 'latitude',
existing_type=mysql.FLOAT(),
type_=sa.Text(),
existing_nullable=True)
op.alter_column('dataset', 'longitude',
existing_type=mysql.FLOAT(),
type_=sa.Text(),
existing_nullable=True)
op.alter_column('user', 'admin',
existing_type=mysql.TINYINT(display_width=1),
type_=sa.Boolean(),
existing_nullable=True)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.alter_column('user', 'admin',
existing_type=sa.Boolean(),
type_=mysql.TINYINT(display_width=1),
existing_nullable=True)
op.alter_column('dataset', 'longitude',
existing_type=sa.Text(),
type_=mysql.FLOAT(),
existing_nullable=True)
op.alter_column('dataset', 'latitude',
existing_type=sa.Text(),
type_=mysql.FLOAT(),
existing_nullable=True)
op.alter_column('dataset', 'collection_date',
existing_type=sa.Text(),
type_=mysql.DATETIME(),
existing_nullable=True)
# ### end Alembic commands ###
def actual_complex_data(self, complex_table_schema):
res = {'id': 1}
for indx, complex_column_schema in enumerate(complex_table_schema):
if isinstance(complex_column_schema.sqla_obj, mysql.DATE):
data = complex_column_schema.data.strftime('%Y-%m-%d')
elif isinstance(complex_column_schema.sqla_obj, mysql.DATETIME):
data = complex_column_schema.data.strftime('%Y-%m-%d %H:%M:%S.%f')
elif isinstance(complex_column_schema.sqla_obj, mysql.TIMESTAMP):
data = complex_column_schema.data.strftime('%Y-%m-%d %H:%M:%S.%f')
elif isinstance(complex_column_schema.sqla_obj, mysql.TIME):
time = datetime.time(
complex_column_schema.data.seconds / 3600,
(complex_column_schema.data.seconds / 60) % 60,
complex_column_schema.data.seconds % 60,
complex_column_schema.data.microseconds
)
data = time.strftime('%H:%M:%S.%f')
else:
data = complex_column_schema.data
res.update({self._build_sql_column_name(indx): data})
return res
def expected_complex_data(self, actual_complex_data, complex_table_schema):
expected_complex_data_dict = {'id': 1}
for indx, complex_column_schema in enumerate(complex_table_schema):
column_name = self._build_sql_column_name(indx)
if isinstance(complex_column_schema.sqla_obj, mysql.SET):
expected_complex_data_dict[column_name] = \
sorted(actual_complex_data[column_name])
elif isinstance(complex_column_schema.sqla_obj, mysql.DATETIME):
date_time_obj = \
complex_column_schema.data.isoformat()
expected_complex_data_dict[column_name] = date_time_obj
elif isinstance(complex_column_schema.sqla_obj, mysql.TIMESTAMP):
date_time_obj = \
complex_column_schema.data.replace(tzinfo=pytz.utc)
expected_complex_data_dict[column_name] = date_time_obj
elif isinstance(complex_column_schema.sqla_obj, mysql.TIME):
number_of_micros = transform_timedelta_to_number_of_microseconds(
complex_column_schema.data
)
expected_complex_data_dict[column_name] = number_of_micros
else:
expected_complex_data_dict[column_name] = \
complex_column_schema.data
return expected_complex_data_dict
4addfa1236f1_add_fractional_seconds_to_mysql_tables.py 文件源码
项目：incubator-airflow-old
作者: apache
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
def upgrade():
if context.config.get_main_option('sqlalchemy.url').startswith('mysql'):
op.alter_column(table_name='dag', column_name='last_scheduler_run', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag', column_name='last_pickled', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag', column_name='last_expired', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag_pickle', column_name='created_dttm', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag_run', column_name='execution_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag_run', column_name='start_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='dag_run', column_name='end_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='import_error', column_name='timestamp', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='job', column_name='start_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='job', column_name='end_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='job', column_name='latest_heartbeat', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='known_event', column_name='start_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='known_event', column_name='end_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='log', column_name='dttm', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='log', column_name='execution_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='sla_miss', column_name='execution_date', type_=mysql.DATETIME(fsp=6), nullable=False)
op.alter_column(table_name='sla_miss', column_name='timestamp', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_fail', column_name='execution_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_fail', column_name='start_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_fail', column_name='end_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_instance', column_name='execution_date', type_=mysql.DATETIME(fsp=6), nullable=False)
op.alter_column(table_name='task_instance', column_name='start_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_instance', column_name='end_date', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='task_instance', column_name='queued_dttm', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='xcom', column_name='timestamp', type_=mysql.DATETIME(fsp=6))
op.alter_column(table_name='xcom', column_name='execution_date', type_=mysql.DATETIME(fsp=6))
4addfa1236f1_add_fractional_seconds_to_mysql_tables.py 文件源码
项目：incubator-airflow-old
作者: apache
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def downgrade():
if context.config.get_main_option('sqlalchemy.url').startswith('mysql'):
op.alter_column(table_name='dag', column_name='last_scheduler_run', type_=mysql.DATETIME())
op.alter_column(table_name='dag', column_name='last_pickled', type_=mysql.DATETIME())
op.alter_column(table_name='dag', column_name='last_expired', type_=mysql.DATETIME())
op.alter_column(table_name='dag_pickle', column_name='created_dttm', type_=mysql.DATETIME())
op.alter_column(table_name='dag_run', column_name='execution_date', type_=mysql.DATETIME())
op.alter_column(table_name='dag_run', column_name='start_date', type_=mysql.DATETIME())
op.alter_column(table_name='dag_run', column_name='end_date', type_=mysql.DATETIME())
op.alter_column(table_name='import_error', column_name='timestamp', type_=mysql.DATETIME())
op.alter_column(table_name='job', column_name='start_date', type_=mysql.DATETIME())
op.alter_column(table_name='job', column_name='end_date', type_=mysql.DATETIME())
op.alter_column(table_name='job', column_name='latest_heartbeat', type_=mysql.DATETIME())
op.alter_column(table_name='known_event', column_name='start_date', type_=mysql.DATETIME())
op.alter_column(table_name='known_event', column_name='end_date', type_=mysql.DATETIME())
op.alter_column(table_name='log', column_name='dttm', type_=mysql.DATETIME())
op.alter_column(table_name='log', column_name='execution_date', type_=mysql.DATETIME())
op.alter_column(table_name='sla_miss', column_name='execution_date', type_=mysql.DATETIME(), nullable=False)
op.alter_column(table_name='sla_miss', column_name='timestamp', type_=mysql.DATETIME())
op.alter_column(table_name='task_fail', column_name='execution_date', type_=mysql.DATETIME())
op.alter_column(table_name='task_fail', column_name='start_date', type_=mysql.DATETIME())
op.alter_column(table_name='task_fail', column_name='end_date', type_=mysql.DATETIME())
op.alter_column(table_name='task_instance', column_name='execution_date', type_=mysql.DATETIME(), nullable=False)
op.alter_column(table_name='task_instance', column_name='start_date', type_=mysql.DATETIME())
op.alter_column(table_name='task_instance', column_name='end_date', type_=mysql.DATETIME())
op.alter_column(table_name='task_instance', column_name='queued_dttm', type_=mysql.DATETIME())
op.alter_column(table_name='xcom', column_name='timestamp', type_=mysql.DATETIME())
op.alter_column(table_name='xcom', column_name='execution_date', type_=mysql.DATETIME())
def load_dialect_impl(self, dialect):
if dialect.name == 'mysql':
return dialect.type_descriptor(mysql.DATETIME(fsp=6))
return self.impl
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('users',
sa.Column('id', mysql.INTEGER(display_width=11), nullable=False),
sa.Column('username', mysql.VARCHAR(length=100), nullable=False),
sa.Column('password_hash', mysql.TEXT(), nullable=False),
sa.Column('email', mysql.VARCHAR(length=200), nullable=False),
sa.Column('gender', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
sa.Column('desc', mysql.TEXT(), nullable=False),
sa.Column('credit', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
sa.Column('role', mysql.VARCHAR(length=20), nullable=False),
sa.Column('status', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
sa.Column('created_time', mysql.DATETIME(), nullable=False),
sa.Column('updated_time', mysql.DATETIME(), nullable=False),
sa.PrimaryKeyConstraint('id'),
mysql_default_charset='utf8',
mysql_engine='InnoDB'
)
op.create_table('posts',
sa.Column('id', mysql.INTEGER(display_width=11), nullable=False),
sa.Column('user_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=True),
sa.Column('title', mysql.VARCHAR(length=100), nullable=False),
sa.Column('content', mysql.TEXT(), nullable=False),
sa.Column('views', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
sa.Column('status', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False),
sa.Column('created_time', mysql.DATETIME(), nullable=False),
sa.Column('updated_time', mysql.DATETIME(), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], name='posts_ibfk_1'),
sa.PrimaryKeyConstraint('id'),
mysql_default_charset='utf8',
mysql_engine='InnoDB'
)
# ### end Alembic commands ###
