MySQL适配器 / MySQL Adapter
MySQL是一种关系型数据库，下面主要介绍利用如何利用Python的MySQL适配器来对MySQL进行操作，其余内容可参考文末相关阅读。
1 MySQL环境配置 / Environment Configuration
安装MySQL
首先在官网下载对应版本的MySQL，在安装过程中会要求Visual Studio 201x的安装环境，可前往下载Visual Studio 2015版本，安装完毕后重启完成MySQL安装。
配置MySQL
- 在安装目录下找到配置文件，my-default.ini，重命名为my.ini;
- 在环境变量中添加MySQL路径(eg.:C:\Program Files\MySQL\MySQL Server 5.7\bin);
- 安装MySQL服务，管理员模式打开cmd，执行命令mysqld –install MySQL –defaults-file=”my.ini”，看到提示Service Successfully installed表示安装成功。
Note: 可使用bat完成第3步操作
mysqld --install MySQL --defaults-file="my.ini"
MySQL服务的启动、停止和卸载
在cmd下运行:
启动: net start MySQL
停止: net stop MySQL
卸载: sc delete MySQL
MySQL登录
直接操作MySQL，无需适配器，
cmd登录本地MySQL: MySQL -u username -p[ db_name] (then input your password)
2 MySQLdb适配器 / MySQLdb Adaptor
以MySQLdb适配器为例，连接MySQL数据库。
2.1 安装MySQLdb / Install MySQLdb
安装Python对MySQL的适配器，可使用pip进行安装，Python 2可安装MySQL-python，而对于Python 3需安装mysqlclient。
pip install mysqlclient
2.2 DB-API扩展 / DB-API Extension
这部分内容为在通用标准基础上额外扩展的 API 接口。
2.2.1 query()方法
函数调用: cnx.query(sql)
函数功能:发送SQL语句进行执行
传入参数: sql
sql: str类型，需要执行的SQL语句
返回参数: 无
2.2.2 store_result ()方法
函数调用: rst = cnx.store_result()
函数功能:一般在query()发送SQL的请求命令之后，用于保存返回信息
传入参数: 无
返回参数: rst
rst: obj类型，一个数据结果对象
2.2.3 fetch_row ()方法
函数调用: r = rst.fetch_row(size, type=0)
函数功能:用于从保存的返回信息中获取结果
传入参数: size, type
size: int类型，获取数据数量，0则获取所有数据
type: int类型，0则返回的每条结果为元组，标准DB规定，1为字典，2为老式字典形式，不推荐使用
返回参数: r
r: tuple类型，包含获取的结果信息
3 适配器构建示例 / Adaptor Build Example
下面的例子将以MySQL的操作语句为例，构建一个适用于MySQL数据库的适配器，以root权限登录并进行一系列操作。
1 class SqlConnector():
2 def __init__(self, adaptor):
3 self.adaptor = adaptor
4 self.result = None
5 # Check Adaptor info
6 print(‘Adaptor %s apply DB-API %s, thread safety: %d, parameter style: %s‘ % (adaptor.__name__, adaptor.apilevel, adaptor.threadsafety, adaptor.paramstyle))
7
8 # Login by user name and password
9 def login(self, host=‘localhost‘, user=‘root‘, password=‘root‘, **kwargs):
10 # Create a connection obj
11 self.cnx = self.adaptor.connect(host=host, user=user, password=password, **kwargs)
12
13 def logoff(self):
14 self.cnx.close()
15
16 def query_sql(self, sql, show=True):
17 # Method one: Use Connection
18 ‘‘‘
19 self.cnx.query(sql)
20 self.result = self.cnx.store_result()
21 self.cnx.commit()
22 r = self.result.fetch_row(0, )
23 ‘‘‘
24 # Method two: Use Cursor
25 cur = self.cnx.cursor()
26 cur.execute(sql)
27 self.cnx.commit()
28 r = cur.fetchall()
29 cur.close()
30
31 if show:
32 splt = ‘\n‘ + ((len(sql)+6)*‘-‘)
33 msg = ‘‘.join((‘\n{: ^10} | ‘).format(str(i)) if x.index(i) == 0 else (‘{: ^10} | ‘).format(str(i)) for x in r for i in x)
34 s = (‘{:-^%d}‘%(len(sql)+6)).format(sql) + msg + splt
35 print(s)
36 return (i for x in r for i in x)
37
38 def exec_sql(self, sql):
39 cur = self.cnx.cursor()
40 cur.execute(sql)
41 self.cnx.commit()
42 cur.close()
43
44 def create_try(self, sql_create, sql_drop, name):
45 try:
46 self.exec_sql(sql_create)
47 print(‘%s create success‘ % name)
48 # MySQLdb.ProgrammingError or MySQLdb.OperationalError
49 except:
50 r = input(‘%s may exist, delete and create a new one?[y/n]‘ % name)
51 if r == ‘n‘:
52 print(‘Failed to create %s‘ % name)
53 return False
54 try:
55 self.exec_sql(sql_drop)
56 print(‘%s delete success‘ % name)
57 self.exec_sql(sql_create)
58 print(‘%s create success‘ % name)
59 except:
60 print(‘Failed to create %s‘ % name)
61 return False
62 return True
63
64 # --------------- For user handle -------------------
65 def create_user(self, usn, psw):
66 sql = ‘CREATE USER %s IDENTIFIED BY "%s";‘ % (usn, psw)
67 try:
68 self.exec_sql(sql)
69 except MySQLdb.OperationalError:
70 print(‘Create user %s failed‘ % usn)
71 return False
72 return True
73
74 def show_grants(self, usn):
75 sql = ‘SHOW GRANTS FOR %s;‘ % usn
76 # re = self.query_sql(sql)
77 self.query_sql(sql)
78
79 def grant_for(self, user):
80 pass
81
82 # --------------- For database handle -------------------
83 def create_database(self, db_name):
84 # Try to CREATE a database, DELETE database if create failed
85 # And try CREATE again, if failed again, exit
86 # You can try ‘CREATE DATABASE IF NOT EXISTS db_name;‘ too
87 sql_create = ‘CREATE DATABASE %s;‘ % db_name
88 sql_drop = ‘DROP DATABASE %s;‘ % db_name
89 return self.create_try(sql_create, sql_drop, db_name)
90
91 def drop_database(self, db_name):
92 sql = ‘DROP DATABASE IF EXISTS %s;‘ % db_name
93 self.exec_sql(sql)
94
95 def use_database(self, db_name):
96 sql = ‘USE %s;‘ % db_name
97 self.exec_sql(sql)
98
99 def show_databases(self):
100 sql = ‘SHOW DATABASES;‘
101 re = self.query_sql(sql)
102 # print(list(re))
103
104 # --------------- For table handle -------------------
105 def create_table(self, tb_name, tb_info, db_name=None):
106 if db_name:
107 self.use_database(db_name)
108 tb_info = (‘, ‘.join(‘%s %s‘ % (k, v) for k, v in tb_info.items())).join(‘()‘)
109 sql_create = ‘CREATE TABLE %s %s;‘ % (tb_name, tb_info)
110 sql_drop = ‘DROP TABLE %s;‘ % tb_name
111 return self.create_try(sql_create, sql_drop, tb_name)
112
113 def drop_table(self, tb_name, db_name=None):
114 if db_name:
115 self.use_database(db_name)
116 sql = ‘DROP TABLE IF EXISTS %s;‘ % tb_name
117 self.exec_sql(sql)
118
119 #def use_table(self, tb_name, db_name=None):
120 # if db_name:
121 # self.use_database(db_name)
122 # sql = ‘USE %s;‘ % tb_name
123 # self.exec_sql(sql)
124
125 def show_tables(self, db_name):
126 self.use_database(db_name)
127 sql = ‘SHOW TABLES;‘
128 self.query_sql(sql)
129
130 def describe_table(self, tb_name, db_name=None):
131 if db_name:
132 self.use_database(db_name)
133 sql = ‘DESCRIBE %s;‘ % tb_name
134 self.query_sql(sql)
135
136 # --------------- For column handle -------------------
137 def insert_column(self, tb_name, value):
138 sql = ‘INSERT INTO %s VALUES %s‘ % (tb_name, value)
139 self.query_sql(sql)
140
141 def drop_columns(self, tb_name, cl_name, value):
142 if isinstance(value, str):
143 value = value.join("‘‘")
144 sql = "DELETE FROM %s WHERE %s=%s" % (tb_name, cl_name, str(value))
145 self.query_sql(sql, show=False)
146
147 def insert_columns(self, tb_name, values):
148 n = len(values[0])
149 qn = (‘, ‘.join(‘%s‘ for x in range(n))).join(‘()‘)
150 sql = ‘INSERT INTO %s VALUES%s‘ % (tb_name, qn)
151 print(sql)
152 cur = self.cnx.cursor()
153 cur.executemany(sql, values)
154 cur.close()
155
156 def show_columns(self, tb_name):
157 sql = ‘SHOW COLUMNS in %s;‘ % tb_name
158 self.query_sql(sql)
159
160 def show_all_rows(self, tb_name):
161 sql = ‘SELECT * FROM %s‘ % tb_name
162 self.query_sql(sql)
163
164 def select_row(self, tb_name, cl_name):
165 row_name = ‘, ‘.join(x for x in cl_name)
166 sql = ‘SELECT %s FROM %s‘ % (row_name, tb_name)
167 self.query_sql(sql)
168
169
170 if __name__ == ‘__main__‘:
171 import MySQLdb
172 c = SqlConnector(MySQLdb)
173 #c = SqlConnector(pyodbc)
174 c.login()
175 c.create_database(‘mysql_db‘)
176 c.use_database(‘mysql_db‘)
177 #c.create_user(‘Ericsson‘, ‘123456‘)
178 #c.show_grants(‘Ericsson‘)
179 #c.show_databases()
180 #c.drop_database(‘test_db‘)
181 c.show_databases()
182 from collections import OrderedDict
183 ord_dict = OrderedDict()
184 ord_dict[‘id‘] = ‘TINYINT‘
185 ord_dict[‘name‘] = ‘VARCHAR(8)‘
186 ord_dict[‘age‘] = ‘INT‘
187 c.create_table(‘test_tb‘, ord_dict, db_name=‘mysql_db‘)
188 c.show_tables(‘mysql_db‘)
189 c.describe_table(‘test_tb‘)
190 c.show_all_rows(‘test_tb‘)
191 c.insert_column(‘test_tb‘, (7, ‘ANAY‘, 9))
192 c.insert_columns(‘test_tb‘, [(4, ‘AX‘, 2), (5, ‘SD‘, 2), (6, ‘CSA‘, 5)])
193 c.drop_columns(‘test_tb‘, ‘name‘, ‘1‘)
194 c.show_all_rows(‘test_tb‘)
195 c.show_columns(‘test_tb‘)
196 #c.select_row(‘test_tb‘, (‘age‘, ‘name‘))
197 c.logoff()
首先建立适配器，适配器可以使用继承通用适配器中应用示例的适配器进行构建，以减少代码重复。此处未采用继承，直接进行建立。
第 1-42 行，定义适配器类的基本操作方法。
第 44-62 行，定义一个创建尝试函数，当有需要创建一个新的数据库或表格时，会进行一个创建尝试，若创建失败，则可能是由于创建对象已存在导致的，询问用户是否尝试删除原表，若是则尝试删除后再次重建。
第 64-80 行，定义用于用户管理的函数，如创建用户，查看用户权限等(其余待补充)。
第 82-102 行，定义用于创建、删除、使用及显示数据库的函数。
第 104-134行，定义用于创建、删除、显示表的函数。
第 136-167 行，定义用于插入(多)、删除、显示列的函数，以及显示表内数据的函数。
第 171-181 行，执行函数，首先创建表，并选择使用创建的表，
第 182-183 行，此处由于字典的无序性，会导致列顺序无法确定，因此引入了有序字典。
第 184-197 行，对表进行一系列操作，并最终退出登录。
最终输出结果如下，
Adaptor MySQLdb apply DB-API 2.0, thread safety: 1, parameter style: format
mysql_db may exist, delete and create a new one?[y/n]y
mysql_db delete success
mysql_db create success
---SHOW DATABASES;---
information_schema |
my_database |
my_schema |
mysql |
mysql_db |
performance_schema |
sakila |
sys |
test_db |
world |
---------------------
test_tb create success
---SHOW TABLES;---
test_tb |
------------------
---DESCRIBE test_tb;---
id | tinyint(4) | YES | | None | |
name | varchar(8) | YES | | None | |
age | int(11) | YES | | None | |
-----------------------
---SELECT * FROM test_tb---
---------------------------
---INSERT INTO test_tb VALUES (7, ‘ANAY‘, 9)---
-----------------------------------------------
INSERT INTO test_tb VALUES(%s, %s, %s)
---SELECT * FROM test_tb---
7 | ANAY | 9 |
4 | AX | 2 |
5 | SD | 2 |
6 | CSA | 5 |
---------------------------
---SHOW COLUMNS in test_tb;---
id | tinyint(4) | YES | | None | |
name | varchar(8) | YES | | None | |
age | int(11) | YES | | None | |
------------------------------
至此，利用 Python 完成了一个简单的数据库适配器。
相关阅读
1. 数据库概述
2. DB-API 通用标准
参考链接
http://www.cnblogs.com/mr-wid/archive/2013/05/09/3068229.html
http://jingyan.baidu.com/article/414eccf612e0c16b431f0a95.html
原文地址：https://www.cnblogs.com/stacklike/p/8178748.html
时间： 2024-11-10 14:30:50