Python與數據庫[1] -> 數據庫接口/DB-API -> MySQL 適配器
阿新 • • 發佈：2018-01-02
and eve div 閱讀 == ccf mysql- 直接 類型
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
Python與數據庫[1] -> 數據庫接口/DB-API -> MySQL 適配器