python操作mysql

 1 #coding:utf-8
 2 import pymysql
 3 import time
 4 from con_mysql import common_file
 5 import logging
 6 
 7 
 8 
 9 class MysqlHelper():
10     def __init__(self, host, port, db, user, passwd, charset=‘utf8‘):
11         self.host = host
12         self.port = port
13         self.db = db
14         self.user = user
15         self.passwd = passwd
16         self.charset = charset
17         self.conn = pymysql.connect(host=self.host, port=self.port, db=self.db, user=self.user, passwd=self.passwd,
18                                     charset=self.charset)
19         self.cursor = self.conn.cursor()
20 
21 
22 
23     def close(self):
24         self.cursor.close()
25         self.conn.close()
26 
27     def get_one(self, sql):
28         try:
29             self.cursor.execute(sql)
30             result = self.cursor.fetchone()
31             return result
32         except Exception as e:
33             logging.debug(‘get_one：%s‘ % e)
34             logging.info(‘your print：%s‘ % sql)
35 
36 
37     def get_all(self, sql):
38         try:
39             self.cursor.execute(sql)
40             result = self.cursor.fetchall()
41             return result
42         except Exception as e:
43             logging.debug(‘get_all：%s‘ % e)
44             logging.info(‘your print：%s‘ % sql)
45 
46     def do_sql(self, sql):
47         try:
48             rec = self.cursor.execute(sql)
49             self.conn.commit()
50             return rec
51         except Exception as e:
52             logging.debug(‘do_sal：%s‘ % e)
53             logging.info(‘your print：%s‘ % sql)
54 
55 
56     def get_id(self, table):
57         try:
58             self.cursor.execute("select max(id) from %s;" % table)
59             rec = self.cursor.fetchone()
60             return rec[0]
61         except Exception as e:
62             logging.debug(‘get_id：%s‘ % e)
63 
64     def get_time(self):
65         rec = str(time.time())[:10]
66         return int(rec)
67 
68     def get_date(self):
69         rec = time.strftime(‘%Y%m%d‘,time.localtime(time.time()))
70         return int(rec)
71 
72 
73 
74 TASK = MysqlHelper(host=‘127.0.0.1‘, db=‘AutoTestEnv‘, port=3306, user=‘root‘,passwd=‘100100100‘,charset=‘utf8‘)
75 
76 # 获取task id
77 project_id = TASK.get_all("select taskId_id from AutoTestEnv.testTask_project where projectName=‘语文（张三）‘ and status=‘运行中‘")
78 if project_id:
79     pid = project_id[-1][0]
80     print(pid)
81 
82     email = TASK.get_one("select email from AutoTestEnv.testTask_task where id=%d and status=‘运行中‘"%pid)
83     print(email)