目录机构如下:
dbutil代码如下:
#!/usr/bin/python # -*- coding:utf-8 -*- import configparser import pymysql class dbutil: # dbsection为配置文件中的section def __init__(self,dbsection): self._conn=self.dbConn(dbsection) if(self._conn): self._cursor=self._conn.cursor() def dbConn(self,dbsection): #读取db.ini文件 cf = configparser.ConfigParser() cf.read("../config/db.ini") dbhost = cf.get(dbsection,"host") dbport = cf.getint(dbsection,"port") dbuser = cf.get(dbsection,"user") dbpassword = cf.get(dbsection,"password") dbname = cf.get(dbsection,"dbname") dbcharset = cf.get(dbsection,"charset") #开始数据库连接 try: conn = pymysql.Connect(host=dbhost,port=dbport,user=dbuser,passwd=dbpassword,db=dbname,charset=dbcharset) except: print("连接失败 host=%s,port=%d,user=%s,dbpassword=%s,db=%s"%(dbhost,dbport,dbuser,dbpassword,dbname)) conn=False return conn # 获取查询结果集 def fetch_all(self, sql): res = ‘‘ if (self._conn): try: self._cursor.execute(sql) res = self._cursor.fetchall() except Exception as data: res = False print("query database exception, %s" % data) return res # 执行更新语句 def update(self, sql): flag = False if (self._conn): try: self._cursor.execute(sql) self._conn.commit() flag = True except Exception as data: flag = False print("update database exception, %s" % data) return flag # 关闭数据库连接 def close(self): if (self._conn): # print(type(self._conn)==‘object‘) try: # if (type(self._cursor) == ‘object‘): self._cursor.close() # if (type(self._conn) == ‘object‘): self._conn.close() except Exception as data: print("close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn))) if __name__=="__main__": dbutil=dbutil("biz_mysql") res=dbutil.fetch_all("select * from wm_people limit %d"%(2)) dbutil.close() print(len(res)) for dd in res: print(dd[0])
对应的db.ini如下
[biz_mysql] host = 10.15.17.xxx port = 3306 user = xxx password = 123456 dbname = xxx charset = utf8 [data_mysql] host = 10.15.17.xxx port = 3306 user = biz password = 123456 dbname = xxx charset = utf8
时间: 2024-10-07 16:47:34