python入门二（数据库操作）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import pymysql
from test.test_robotparser import PasswordProtectedSiteTestCase
#dicts
dbdict={"testdb":{"host":"127.0.0.1","port":3306,"user":"root","passwd":"admin","db":"test","charset":"utf-8"},
"userdb":{"host":"127.0.0.1","port":3306,"user":"root","passwd":"admin","db":"user","charset":"utf-8"}}
class dball(object):
def __init__(self,name,dicts):
self.name= name
self.host=dicts["host"]
self.port =dicts["port"]
self.user =dicts["user"]
self.passwd =dicts["passwd"]
self.db =dicts["db"]
self.charset = dicts["charset"]
print("init complete")
def getConnection(self):
cnx = pymysql.connect(host=self.host,port =self.port,user=self.user,passwd=self.passwd,db=self.db)
return cnx
def getDicCursor(self,cnx):
cursor = cnx.cursor(pymysql.cursors.DictCursor)
return cursor
def getCursor(self,cnx):
cursor = cnx.cursor()
return cursor
def closeCursor(self,cursor):
if cursor is not None:
cursor.close()
print(str(self.name)+" cursor is closed")
return True
elif cursor is None:
print(str(self.name)+" cursor is None")
return True
else :
print(str(self.name)+" cursor close failed")
return False
def closeCnx(self,cnx):
if cnx is not None:
cnx.close()
print(str(self.name)+" cnx is closed")
return True
elif cnx is None:
print(str(self.name)+" cnx is None")
else:
print(str(self.name)+" cnx closed failed")
return False
def queryDic(self,sqls,cursor):
len = cursor.execute(sqls)
result = cursor.fetchall()
self.closeCursor(cursor)
mydicts={"len":len,"result":result}
return mydicts
def chancelResult(self,dict):
if dict is not None:
nameList =[]
for res in dict:
name = res["name"]
nameList.append(str(name))
count =nameList.count("110")
print("count: "+str(count))
elif dict is None:
print(str(self.name)+" result is none")
else:
print("want to do ?")
if __name__ =="__main__":
dbtest1 = dball("dbtest1",dbdict["userdb"])
con = dbtest1.getConnection()
cursor =dbtest1.getDicCursor(con)
sql ="SELECT id,name FROM user"
dictresult = dbtest1.queryDic(sql, cursor)
len = dictresult["len"]
result =dictresult["result"]
dbtest1.chancelResult(result)
print("len:"+str(len) +" result:"+str(result) )
print("query end")
dbtest1.closeCnx(con)

1
2
3
4
5
6
7
init complete
dbtest1 cursor is closed
count: 3
len:13 result:[{'id': '1', 'name': '12'}, {'id': '110', 'name': '110'}, {'id': '1101', 'name': '110'}, {'id': '1102', 'name': '110'}, {'id': '120', 'name': '120'}, {'id': '121', 'name': '121'}, {'id': '122', 'name': '122'}, {'id': '124', 'name': '124'}, {'id': '125', 'name': '125'}, {'id': 'whd1', 'name': 'whd1'}, {'id': 'whd2', 'name': 'whd2'}, {'id': 'whd3', 'name': 'whd3'}, {'id': 'whd5', 'name': 'whd5'}]
query end
dbtest1 cnx is closed