Python sqlalchemy.dialects.mysql 模块，DATETIME 实例源码

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 ###

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

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

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 ###