Python中操作MySQL/Oracle

• 一、Python操作數據庫介紹
• 二、Python操作MySQL
• • 2.1 PySQL模塊
  • • 2.1.1 安裝PyMySQL
  • 2.2 基本使用
  • 2.3 獲取最新創建的數據庫自增ID
  • 2.4 查詢操作
  • 2.5 防止SQL注入
• 三、數據庫連接池
• • 3.1 DBUtils模塊
  • 3.2 模式一
  • 3.3 模式二
  • 3.3 加鎖
  • 3.4 無鎖（報錯）
• 四、數據庫連接池結合pymysql使用
• 五、MySQL中substr()函數的使用
• 參考資料

一、Python操作數據庫介紹

Python 標準數據庫接口為 Python DB-API，Python DB-API為開發人員提供了數據庫應用編程接口。Python 數據庫接口支持非常多的數據庫，你可以選擇適合你項目的數據庫：

GadFly
mSQL
MySQL
PostgreSQL
Microsoft SQL Server 2000
Informix
Interbase
Oracle
Sybase …

你可以訪問Python數據庫接口及API查看詳細的支持數據庫列表。

不同的數據庫你需要下載不同的DB API模塊，例如你需要訪問Oracle數據庫和Mysql數據，你需要下載Oracle和MySQL數據庫模塊。

DB-API 是一個規范. 它定義了一系列必須的對象和數據庫存取方式, 以便為各種各樣的底層數據庫系統和多種多樣的數據庫接口程序提供一致的訪問接口 。

Python的DB-API，為大多數的數據庫實現了接口，使用它連接各數據庫后，就可以用相同的方式操作各數據庫。

Python DB-API使用流程：

• 引入 API 模塊。
• 獲取與數據庫的連接。
• 執行SQL語句和存儲過程。
• 關閉數據庫連接。

二、Python操作MySQL

Python操作MySQL主要使用兩種方式：

DB模塊（原生SQL）
PyMySQL(支持python2.x/3.x)
MySQLdb(目前僅支持python2.x)

ORM框架
SQLAchemy

2.1 PySQL模塊

本文主要介紹PyMySQL模塊，MySQLdb使用方式類似

2.1.1 安裝PyMySQL

筆者的開發環境：Windows 10 Education、Python3.8

PyMySQL是一個Python編寫的MySQL驅動程序，讓我們可以用Python語言操作MySQL數據庫。

pip install PyMySQL

2.2 基本使用

# 1 基本使用pymysql import pymysql# 創建連接 conn = pymysql.connect(host="127.0.0.1", port=3306, user="root", password="123456", db="springbootdb",charset="utf8mb4") # 創建游標（查詢數據返回的元組格式） cursor = conn.cursor()# 創建游標（查詢數據返回字典格式） # cursor = conn.cursor(pymysql.cursors.DictCursor)# 1.指定SQL,返回受影響的行數，一次插入一行數據 effect_row1 = cursor.execute("insert into city (province_id, city_name, description) value (61, '渭南', '陜西省渭南市')")# 2. 執行SQL,返回受影響的行數 effect_row2 = cursor.execute("select * from city")# 查詢所有數據，返回數據為元組格式 result = cursor.fetchall()# 增刪改需要進行commit提交 conn.commit()# 關閉游標 cursor.close()# 關閉連接 conn.close()print(result)

2.3 獲取最新創建的數據庫自增ID

# 2 獲取最新創建的數據自增ID import pymysqlhost = "127.0.0.1" port = 3306 user = "root" password = "123456" db = "springbootdb" charset = "utf8mb4"# 創建連接 conn = pymysql.connect(host=host, port=port, user=user, password=password, db=db, charset=charset)# 創建游標（查詢數據返回元組格式） cursor = conn.cursor()# 獲取新創建數據自增ID effect_row = cursor.executemany("insert into city (province_id, city_name, description) values (%s, %s, %s)",[("61", "咸陽1", "陜西省咸陽市"), ("61", "咸陽2", "陜西省咸陽市"), ("61", "咸陽3", "陜西省咸陽市")])# 增刪改需要進行commit操作 conn.commit()# 關閉游標 cursor.close()new_id = cursor.lastrowid print(new_id)

2.4 查詢操作

# 3 查詢操作 import pymysqlhost = "127.0.0.1" port = 3306 user = "root" password = "123456" db = "springbootdb" charset = "utf8mb4"# 創建連接 conn = pymysql.connect(host=host, port=port, user=user, password=password, db=db, charset=charset)# 創建游標 cursor = conn.cursor()cursor.execute("select * from city")# 獲取第一行數據 row_1 = cursor.fetchone()# 獲取前N行數據 row_2 = cursor.fetchmany(3)# 獲取所有數據 row_3 = cursor.fetchall()# 關閉游標 cursor.close()# 關閉連接 conn.close()print(row_1) print(row_2) print(row_3)

在fetch數據時按照順序進行，可以使用cursor.scroll(num,mode)來移動游標位置，如：

cursor.scroll(1,mode='relative') # 相對當前位置移動 cursor.scroll(2,mode='absolute') # 相對絕對位置移動

2.5 防止SQL注入

# 4 防止SQL注入 import pymysqlhost = "127.0.0.1" port = 3306 user = "root" password = "123456" db = "springbootdb" charset = "utf8mb4"# 創建連接 conn = pymysql.connect(host=host, port=port, user=user, passwd=password, db=db, charset=charset)# 創建游標 cursor = conn.cursor()# 存在SQL注入情況（不要用格式化字符串的方式拼接SQL） sql = "insert into city(province_id, city_name, description) VALUES (%d,'%s','%s')" % (61, '榆林', '陜西省榆林市') effect_row = cursor.execute(sql) print(effect_row)# 正確方式一 # execute函數接受一個元組/列表作為SQL參數，元素個數只能有一個 sql = "insert into city (province_id, city_name, description) values (%s, %s, %s)" effect_row1 = cursor.executemany(sql, [("61", "榆林", "陜西省榆林市")]) print(effect_row1)# 正確方式二 sql = "insert into city (province_id, city_name, description) VALUES (%(province_id)s,%(city_name)s,%(description)s)" effect_row2 = cursor.execute(sql, {"province_id": "61", "city_name": "延安", "description": "陜西省延安市"}) print(effect_row2)# 寫入多行數據 sql = "insert into city (province_id, city_name, description) values (%s, %s, %s)" effect_row3 = cursor.executemany(sql, [("61", "榆林1", "陜西省榆林市"), ("61", "榆林2", "陜西省榆林市"), ("61", "榆林3", "陜西省榆林市")]) print(effect_row3)# 提交 conn.commit()# 關閉游標 cursor.close()# 關閉連接 conn.close()

這樣，SQL操作就更安全了。如果需要更詳細的文檔參考PyMySQL文檔吧。不過好像這些SQL數據庫的實現還不太一樣，PyMySQL的參數占位符使用%s這樣的C格式化符，而Python自帶的sqlite3模塊的占位符好像是問號(?)。因此在使用其他數據庫的時候還是仔細閱讀文檔吧。Welcome to PyMySQL’s documentation

三、數據庫連接池

上文中的方式存在一個問題,單線程情況下可以滿足,程序需要頻繁的創建釋放連接來完成對數據庫的操作,那么,我們的程序/腳本在多線程情況下會引發什么問題呢?此時,我們就需要使用數據庫連接池來解決這個問題!

3.1 DBUtils模塊

DBUtils是Python的一個用于實現數據庫連接池的模塊。
此連接池有兩種連接模式：

• 為每個線程創建一個連接，線程即使調用了close方法，也不會關閉，只是把連接重新放到連接池，供自己線程再次使用。當線程終止時，連接才會自動關閉
• 創建一批連接到連接池，供所有線程共享使用(推薦使用)

3.2 模式一

• 為每個線程創建一個連接，線程即使調用了close方法，也不會關閉，只是把連接重新放到連接池，供自己線程再次使用。當線程終止時，連接才會自動關閉

from DBUtils.PersistentDB import PersistentDB import pymysqlPOOL = PersistentDB(creator=pymysql, # 使用鏈接數據庫的模塊maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayscloseable=False,# 如果為False時， conn.close() 實際上被忽略，供下次使用，再線程關閉時，才會自動關閉鏈接。如果為True時， conn.close()則關閉鏈接，那么再次調用pool.connection時就會報錯，因為已經真的關閉了連接（pool.steady_connection()可以獲取一個新的鏈接）threadlocal=None, # 本線程獨享值得對象，用于保存鏈接對象，如果鏈接對象被重置host='127.0.0.1',port=3306,user='zff',password='zff123',database='zff',charset='utf8', )def func():conn = POOL.connection(shareable=False)cursor = conn.cursor()cursor.execute('select * from USER')result = cursor.fetchall()cursor.close()conn.close()return resultresult = func() print(result)

3.3 模式二

• 創建一批連接到連接池，供所有線程共享使用(推薦使用)

import time import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnectionPOOL = PooledDB(creator=pymysql, # 使用鏈接數據庫的模塊maxconnections=6, # 連接池允許的最大連接數，0和None表示不限制連接數mincached=2, # 初始化時，鏈接池中至少創建的空閑的鏈接，0表示不創建maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制maxshared=3,# 鏈接池中最多共享的鏈接數量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='127.0.0.1',port=3306,user='zff',password='zff123',database='zff',charset='utf8' )def func():# 檢測當前正在運行連接數的是否小于最大鏈接數，如果不小于則：等待或報raise TooManyConnections異常# 否則# 則優先去初始化時創建的鏈接中獲取鏈接 SteadyDBConnection。# 然后將SteadyDBConnection對象封裝到PooledDedicatedDBConnection中并返回。# 如果最開始創建的鏈接沒有鏈接，則去創建一個SteadyDBConnection對象，再封裝到PooledDedicatedDBConnection中并返回。# 一旦關閉鏈接后，連接就返回到連接池讓后續線程繼續使用。conn = POOL.connection()# print('連接被拿走了', conn._con)# print('池子里目前有', POOL._idle_cache, '\r\n')cursor = conn.cursor()cursor.execute('select * from USER')result = cursor.fetchall()conn.close()return resultresult = func() print(result)

由于pymysql、MySQLdb等threadsafety值為1，所以該模式連接池中的線程會被所有線程共享,因此是線程安全的。如果沒有連接池，使用pymysql來連接數據庫時，單線程應用完全沒有問題，但如果涉及到多線程應用那么就需要加鎖，一旦加鎖那么連接勢必就會排隊等待，當請求比較多時，性能就會降低了。

3.3 加鎖

import pymysql import threading from threading import RLockLOCK = RLock() CONN = pymysql.connect(host='127.0.0.1',port=3306,user='zff',password='zff123',database='zff',charset='utf8')def task(arg):with LOCK:cursor = CONN.cursor()cursor.execute('select * from USER ')result = cursor.fetchall()cursor.close()print(result)for i in range(10):t = threading.Thread(target=task, args=(i,))t.start()

3.4 無鎖（報錯）

import pymysql import threadingCONN = pymysql.connect(host='127.0.0.1',port=3306,user='zff',password='zff123',database='zff',charset='utf8')def task(arg):cursor = CONN.cursor()cursor.execute('select * from USER ')# cursor.execute('select sleep(10)')result = cursor.fetchall()cursor.close()print(result)for i in range(10):t = threading.Thread(target=task, args=(i,))t.start()

此時可以在數據庫中查看連接情況: show status like ‘Threads%’;

四、數據庫連接池結合pymysql使用

import pymysql import threading from DBUtils.PooledDB import PooledDB, SharedDBConnection POOL = PooledDB(creator=pymysql, # 使用鏈接數據庫的模塊maxconnections=20, # 連接池允許的最大連接數，0和None表示不限制連接數mincached=2, # 初始化時，鏈接池中至少創建的空閑的鏈接，0表示不創建maxcached=5, # 鏈接池中最多閑置的鏈接，0和None不限制#maxshared=3, # 鏈接池中最多共享的鏈接數量，0和None表示全部共享。PS: 無用，因為pymysql和MySQLdb等模塊的 threadsafety都為1，所有值無論設置為多少，_maxcached永遠為0，所以永遠是所有鏈接都共享。blocking=True, # 連接池中如果沒有可用連接后，是否阻塞等待。True，等待；False，不等待然后報錯maxusage=None, # 一個鏈接最多被重復使用的次數，None表示無限制setsession=[], # 開始會話前執行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服務端，檢查是否服務可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='192.168.11.38',port=3306,user='root',passwd='apNXgF6RDitFtDQx',db='m2day03db',charset='utf8' )def connect():# 創建連接# conn = pymysql.connect(host='192.168.11.38', port=3306, user='root', passwd='apNXgF6RDitFtDQx', db='m2day03db')conn = POOL.connection()# 創建游標cursor = conn.cursor(pymysql.cursors.DictCursor)return conn,cursordef close(conn,cursor):# 關閉游標cursor.close()# 關閉連接conn.close()def fetch_one(sql,args):conn,cursor = connect()# 執行SQL，并返回收影響行數effect_row = cursor.execute(sql,args)result = cursor.fetchone()close(conn,cursor)return resultdef fetch_all(sql,args):conn, cursor = connect()# 執行SQL，并返回收影響行數cursor.execute(sql,args)result = cursor.fetchall()close(conn, cursor)return resultdef insert(sql,args):"""創建數據:param sql: 含有占位符的SQL:return:"""conn, cursor = connect()# 執行SQL，并返回收影響行數effect_row = cursor.execute(sql,args)conn.commit()close(conn, cursor)def delete(sql,args):"""創建數據:param sql: 含有占位符的SQL:return:"""conn, cursor = connect()# 執行SQL，并返回收影響行數effect_row = cursor.execute(sql,args)conn.commit()close(conn, cursor)return effect_rowdef update(sql,args):conn, cursor = connect()# 執行SQL，并返回收影響行數effect_row = cursor.execute(sql, args)conn.commit()close(conn, cursor)return effect_row

五、MySQL中substr()函數的使用

對于JavaScript的substring() 方法或者String的subString()方法，相信很多人都有用過，最近因為工作的需要，需要在sql中對字段進行切割。SUBSTR函數是用來截取數據庫某一列字段中的一部分，在各個數據庫的函數名稱不一樣。

MySQL: SUBSTR( ), SUBSTRING( ) Oracle: SUBSTR( ) SQL Server: SUBSTRING( ) ;

常用的方式是：

SBUSTR(str,pos); 從pos開始的位置，一直截取到最后。

還有一種比較常用的是：

SUBSTR(str,pos,len);從pos開始的位置，截取len個字符(空白也算字符)。

注：
mysql中的substr()函數和hibernate的substr()參數都一樣，就是含義有所不同。
mysql中的start是從1開始的，而hibernate中的start是從0開始的。

舉個栗子

查詢到的原始數據格式為：

select a.visitdate from src_his_mz_master_info a;

SUBSTR(str,pos,len);
從pos開始的位置，截取len個字符(空白也算字符)。

select substr(a.visitdate,1,4) from src_his_mz_master_info a;

SBUSTR(str,pos);
從pos開始的位置，一直截取到最后。

select substr(a.visitdate,6) from src_his_mz_master_info a;

參考資料

https://mp.weixin.qq.com/s/vXAbJ9KcSDB0CAMWBL0IJg

https://mp.weixin.qq.com/s/8V2puF-QbG6gVI6MIDDFdg

https://mp.weixin.qq.com/s/LAaGDQd-AX2jGkgUjlufLg

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的Python中操作MySQL/Oracle的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。