| 123456789101112131415161718192021222324252627282930313233343536373839404142434445 | from sqlalchemy import create_enginefrom urllib import parseimport jsonimport numpy as np# 创建数据库连接引擎def get_db_engine(db_info):    db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \                 (db_info['username'],                  parse.quote_plus(db_info['password']),                  db_info['host'],                  db_info['port'],                  db_info['database'])    engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})    return engine# insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据def mysql_replace_into(table, conn, keys, data_iter):    from sqlalchemy.dialects.mysql import insert    from sqlalchemy.ext.compiler import compiles    from sqlalchemy.sql.expression import Insert    @compiles(Insert)    def replace_string(insert, compiler, **kw):        s = compiler.visit_insert(insert, **kw)        s = s.replace("INSERT INTO", "REPLACE INTO")        return s    data = [dict(zip(keys, row)) for row in data_iter]    conn.execute(table.table.insert(replace_string=""), data)class NpEncoder(json.JSONEncoder):    def default(self, obj):        if isinstance(obj, np.integer):            return int(obj)        elif isinstance(obj, np.floating):            return float(obj)        elif isinstance(obj, np.ndarray):            return obj.tolist()        else:            return super(NpEncoder, self).default(obj)
 |