123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- from sqlalchemy import create_engine
- from urllib import parse
- import json
- import 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)
|