end_to_end_test.py 文件源码

python
阅读 21 收藏 0 点赞 0 评论 0

项目:mysql_streamer 作者: Yelp 项目源码 文件源码
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
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号