importpymysqlfrom scrapy.utils.project import get_project_settings#引入settings配置
classDBHelper():def __init__(self):
self.settings=get_project_settings()#获取settings配置数据
mysql 去重复。self.host=self.settings['MYSQL_HOST']
self.port=self.settings['MYSQL_PORT']
self.user=self.settings['MYSQL_USER']
self.passwd=self.settings['MYSQL_PASSWD']
sql 重复记录、self.db=self.settings['MYSQL_DBNAME']#连接mysql
defconnectMysql(self):
conn=pymysql.connect(host=self.host,
port=self.port,
jdbc连接mysql数据库?user=self.user,
passwd=self.passwd,
charset='utf8')returnconn#连接数据库
defconnectDatabase(self):
mysql删除数据库,conn=pymysql.connect(host=self.host,
port=self.port,
user=self.user,
passwd=self.passwd,
mysql怎么建立数据库。db=self.db,
charset='utf8')returnconn#创建数据库
defcreateDatabase(self):
conn=self.connectMysql()
mysql可重复读。sql="create database if not exists"+self.db
cur=conn.cursor()
cur.execute(sql)
cur.close()
mysql数据库界面、conn.close()#创建数据表
defcreateTable(self,sql):
conn=self.connectDatabase()
cur=conn.cursor()
mysql查看数据库、cur.execute(sql)
cur.close()
conn.close()#插入数据
def insert(self,sql,*params):
sql数据库,conn=self.connectDatabase()
cur=conn.cursor()
cur.execute(sql,params)
conn.commit()
数据库删除重复的数据,cur.close()
conn.close()#更新数据
def update(self,sql,*params):
conn=self.connectDatabase()
cur=conn.cursor()
cur.execute(sql,params)
conn.commit()
cur.close()
conn.close()#删除数据
def delete(self,sql,*params):
conn=self.connectDatabase()
cur=conn.cursor()
cur.execute(sql,params)
conn.commit()
cur.close()
conn.close()#测试数据库操作
classTestDBHelper():def __init__(self):
self.dbHelper=DBHelper()deftestCreateDatebase(self):
self.dbHelper.createDatabase()deftestCreateTable(self):
sql="create table testtable(id int primary key auto_increment,name varchar(50),url varchar(200))"self.dbHelper.createTable(sql)deftestInsert(self, item):
sql="insert into example(hscode, 申报名称, 申报要素, 参考均价, 参考最高价, 参考最低价) values (%s, %s, %s, %s, %s, %s)"params=(item['hscode'], item['申报名称'], item['申报名称'], item['参考均价'], item['参考最高价'], item['参考最低价'])
self.dbHelper.insert(sql,*params)deftestUpdate(self):
sql="update testtable set name=%s,url=%s where id=%s"params=("update","update","1")
self.dbHelper.update(sql,*params)deftestDelete(self):
sql="delete from testtable where id=%s"params=("1")
self.dbHelper.delete(sql,*params)if __name__=="__main__":
testDBHelper=TestDBHelper()#testDBHelper.testCreateDatebase() #
#testDBHelper.testCreateTable() #
testDBHelper.testInsert() # #testDBHelper.testUpdate() #
#testDBHelper.testDelete() #
版权声明:本站所有资料均为网友推荐收集整理而来,仅供学习和研究交流使用。
工作时间:8:00-18:00
客服电话
电子邮件
admin@qq.com
扫码二维码
获取最新动态