本文主要内容python MySQLdb数据库批量插入插入,更新更新的:
1。python MySQLdb的使用,写了一个基类让其他的sqldb继承这样比较方便,数据库的ip、端口等信息使用json配置文件
2。常见的查找,批量插入更新
下面贴出基类代码:
# _ * _编码:utf - 8 _ * _
进口MySQLdb
进口json
进口编解码器
#这个自己改一下啊
从跑龙套。JsonUtil进口get_json_from_file
def byteify(输入):
”“”
的json类型unicode字符串str在python中
这个函数来自堆栈溢出
参数输入:{u 'first_name: u 'Guido, u 'last_name”: u 'jack '}
返回:{“first_name”:“圭多”,“last_name”:“杰克”}
”“”
如果isinstance(输入、dict类型):
返回{byteify(关键):byteify(值)
关键,input.iteritems价值()}
elif isinstance(输入列表):
返回(byteify(元素)元素输入)
unicode elif isinstance(输入):
返回input.encode (“utf - 8”)
其他:
返回输入
def get_json_from_file(文件名):
摩根富林明张开(文件名):
jsondata=https://www.yisu.com/zixun/json.load(摩根富林明)
返回byteify (jsondata)
类DbBase(对象):
def __init__(自我,* * kwargs):
自我。db_config_file=kwargs (“db_config_file”)
self.config_db (self.db_config_file)
def config_db(自我,db_config_file):
data=https://www.yisu.com/zixun/get_json_from_file (db_config_file)
主机=数据(“主机”)
用户=数据(“用户”)
pwd=数据(“pwd”)
db=数据(“数据库”)
端口=数据(“港口”)
自我。tb_audit_mobile=数据(“tb_audit_mobile”)
自我。康涅狄格州=MySQLdb。连接(主机=主机、端口=端口、用户=用户,密码=pwd, db=db, charset=皍tf8 ", use_unicode=True)
自我。光标=self.conn.cursor ()
子类的示例:
类DbAuditTestService (DbBase):
def __init__(自我,* * kwargs):
超级(DbAuditTestService,自我). __init__ (* * kwargs)
def getAdTestURl(自我,乞求,结束):
sql=" "选择url,源从tb_name create_date之间“% s”和“% s”“%(乞讨,结束)
self.cursor.execute (sql)
res=(self.cursor行对行)
返回res
def插入(自我,lst_row):
”““批量插入,使用忽略避免索引唯一问题“”“
试一试:
insert_sql='忽略插入tb_ms_mobile_report_test (appid源)值(% s % s) '
self.cursor。executemany (insert_sql lst_row)
self.conn.commit ()
MySQLdb除外。OperationalError e:
logger.info (' % s ' % e)
self.cursor.close ()
self.conn.close ()
self.config_db (self.db_config_file)
def update_ip_info(自我,ip_info):
”“”
批量更新
[[voilate_right_rate、ip]]
:param ip_info:
返回:
”“”
查询=" "
更新tb_ms_audit_ip_info组
ip_right_rate=% s
submit_ip=% s”“”
self.cursor。ip_info executemany(查询)
self.conn.commit ()
def insert_all ():
”““批量操作的示例“”“
db_audit=DbAuditService (db_config_file=? ./config/mysql_police_audit.json”)
大?db_audit.count ()
db_audit_test=DbAuditTestService (db_config_file=? ./config/mysql_local_audit.json”)
batch_size=2000
对k xrange(100000、大小batch_size):
logger.info(查询限制%年代~ % s % (k, batch_size))
lst_row=db_audit。batch_size query_limit (k)
logger.info (“convert_rows”)
lst_row=convert_rows (lst_row)
db_audit_test.insert (lst_row)
以上所述是小编给大家介绍的python MySQLdb使用教程详解,希望对大家有所帮助,如果大家有任何疑问请给我留的言,小编会及时回复大家的。在此也非常感谢大家对网站的支持!
