一、PyMysql
在使用Python操作MySQL數(shù)據(jù)過的過程中悍及，基本的增刪改查操作如何更加高效優(yōu)雅的執(zhí)行镣煮。這里將以PyMySQL為例义钉，介紹一下如何使用Python操作數(shù)據(jù)庫。 Python對MySQL數(shù)據(jù)庫進行操作氓英，基本思路是先連接數(shù)據(jù)庫 Connection 對象侯勉，建立游標(biāo) Cursor 對象，然后執(zhí)行SQL語句對數(shù)據(jù)庫進行操作铝阐，獲取執(zhí)行結(jié)果址貌，最終斷開連接。大致過程是這樣徘键，在對其進行介紹之前练对，先介紹一些基本的概念。
Connection
Connection 對象即為數(shù)據(jù)庫連接對象吹害，在python中可以使用pymysql.connect()方法創(chuàng)建Connection對象螟凭，該方法的常用參數(shù)如下：
host：IP地址,字符串類型
user：用戶名, 字符串類型
passwd：無默認值；字符串類
db：數(shù)據(jù)庫名稱赠制，無默認值赂摆；字符串類型(可以不傳但是SQL中必須體現(xiàn))
port：端口, 默認為3306, 整型
charset：設(shè)置utf8, 字符串類型
close：關(guān)閉當(dāng)前連接對象
Cursor
Cursor對象即為游標(biāo)對象，用于執(zhí)行查詢和獲取結(jié)果钟些，在python中可以使用connect.cursor()創(chuàng)建
execute()：執(zhí)行數(shù)據(jù)庫單個查詢或命令烟号，將結(jié)果從數(shù)據(jù)庫獲取
executemany(): 對一個查詢運行多個數(shù)據(jù)，其返回是：受影響的行數(shù)(如果有的話)
close()：關(guān)閉當(dāng)前游標(biāo)對象
Transaction
1.事務(wù)是數(shù)據(jù)庫理論中一個比較重要的概念政恍，指訪問和更新數(shù)據(jù)庫的一個程序執(zhí)行單元汪拥，具有ACID特性：
原子性（Atomic）：事務(wù)中的各項操作要么全都做，要么全都不做篙耗，任何一項操作的失敗都會導(dǎo)致整個事務(wù)的失敗
一致性（Consistent）：事務(wù)必須使數(shù)據(jù)庫從一個一致性狀態(tài)變到另一個一致性狀態(tài)
隔離性（Isolated）：并發(fā)執(zhí)行的事務(wù)彼此無法看到對方的中間狀態(tài)迫筑，一個事務(wù)的執(zhí)行不能被其他事務(wù)干擾
持久性（Durable）：事務(wù)一旦提交宪赶，它對數(shù)據(jù)庫的改變就是永久性的，可以通過日志和同步備份在故障發(fā)生后重建數(shù)據(jù)脯燃。
2.常用事務(wù)方法
Connection.commit()：正常事務(wù)提交
Connection.rollback()：事務(wù)異陈蓿回滾
Connection.autocommit()：事務(wù)自動提交機制，默認TRUE辕棚，設(shè)置FALSE則關(guān)閉欲主。
二、Python操作MySQL
1.安裝
$ pip3 install PyMySQL
2.數(shù)據(jù)庫連接
import pymysql
# 打開數(shù)據(jù)庫連接
db = pymysql.connect(host='127.0.0.1',
user='user',
password='123456',
database='demo',
port=3306,
charset='utf8')
# 使用 cursor() 方法創(chuàng)建一個游標(biāo)對象 cursor
cursor = db.cursor()
# 使用 execute() 方法執(zhí)行 SQL 查詢
cursor.execute("SELECT * FROM USER;")
# 使用 fetchone() 方法獲取單條數(shù)據(jù).
data = cursor.fetchone()
print ("Database data : %s " % data)
# 關(guān)閉數(shù)據(jù)庫連接
cursor.close()
db.close()
3.數(shù)據(jù)庫DML操作
import pymysql
# 打開數(shù)據(jù)庫連接
db = pymysql.connect(**config) # 省略連接信息
# 使用cursor()方法獲取操作游標(biāo)
cursor = db.cursor()
# SQL插入語句
sql = """INSERT INTO EMPLOYEE(FIRST_NAME,
LAST_NAME, AGE, SEX, INCOME)
VALUES ('Mac', 'Mohan', 20, 'M', 2000)"""
try:
# 執(zhí)行sql語句
cursor.execute(sql)
# 提交到數(shù)據(jù)庫執(zhí)行
db.commit()
except:
# 如果發(fā)生錯誤則回滾
db.rollback()
# 關(guān)閉當(dāng)前游標(biāo)對象
cursor.close()
# 關(guān)閉數(shù)據(jù)庫連接
db.close()
# 執(zhí)行傳入的SQL也可以更加的靈活逝嚎，可以使用另外一種%s 占位符扁瓢，后續(xù)的參數(shù)依次傳入。
sql2 = """INSERT INTO EMPLOYEE(FIRST_NAME,
LAST_NAME, AGE, SEX, INCOME)
VALUES ('%s', '%s', %s, '%s', %s)"""
cursor.execute(sql2, 'Mac', 'Mohan', 20, 'M', 2000)
4.數(shù)據(jù)庫DQL操作
Python查詢Mysql使用常用幾個方法补君。
fetchone(): 該方法獲取下一個查詢結(jié)果集引几。結(jié)果集是一個對象.
fetchmany()：獲取結(jié)果集的指定幾行.
fetchall(): 接收全部的返回結(jié)果行.
rowcount: 這是一個只讀屬性，并返回執(zhí)行execute()方法后影響的行數(shù)挽铁。
import pymysql
# 打開數(shù)據(jù)庫連接
db = pymysql.connect(**config)
# 使用cursor()方法獲取操作游標(biāo)
cursor = db.cursor()
# SQL 查詢語句
sql = "SELECT * FROM EMPLOYEE WHERE INCOME > %s" % (1000)
try:
# 執(zhí)行SQL語句
cursor.execute(sql)
# 獲取所有記錄列表
result1=cursor.fetchone()
result2=cursor.fetchmany(2)
results = cursor.fetchall()
print(result1)
print(result2)
print(results)
except:
print ("Error: unable to fetch data")
# 關(guān)閉數(shù)據(jù)庫連接
cursor.close()
db.close()
三伟桅、工具類封裝
通過封裝常用方法將會大大降低對數(shù)據(jù)庫操作的成本。接下來分為幾步進行操作：
1.可以通過env文件來存儲數(shù)據(jù)庫的連接信息
2.將env文件數(shù)據(jù)加載進系統(tǒng)環(huán)境變量
3.從系統(tǒng)環(huán)境變量中獲取對應(yīng)連接數(shù)據(jù)
4.連接數(shù)據(jù)庫屿储，操作增刪改查
# .env
DB_INFO={"host": "127.0.0.1","port":3306,"user": "user","passwd": "123456","charset": "utf8"}
import io
import os
class EnvironmentVarUtils(object):
def __init__(self, fileName=None):
self.file_name = fileName
self._load_dot_env_file(self._get_environment_path())
def _get_environment_path(self):
"""
:return: project_path
"""
return os.path.join(os.path.dirname(os.getcwd()), '.env') if self.file_name is None\
else os.path.join(os.path.dirname(os.getcwd()), self.file_name)
def _load_dot_env_file(self, dot_env_path):
""" load .env file.
Args:
dot_env_path (str): .env file path
"""
if not os.path.isfile(dot_env_path):
raise FileNotFoundError(".env file not found Error.")
print("Loading environment variables from 【{}】".format(dot_env_path))
env_variables_mapping = {}
with io.open(dot_env_path, 'r', encoding='utf-8') as fp:
for line in fp:
if "=" in line:
variable, value = line.split("=", 1)
else:
raise Exception(".env format error")
env_variables_mapping[variable.strip()] = value.strip()
self._set_os_environ(env_variables_mapping)
@staticmethod
def _set_os_environ(variables_mapping):
""" set variables mapping to os.environ """
for variable in variables_mapping:
os.environ[variable] = variables_mapping[variable]
print("Set OS environment variable: {}".format(variable))
@staticmethod
def get_os_environ(variable_name):
""" get value of environment variable.
"""
try:
return os.environ[variable_name]
except Exception as e:
raise e
import pymysql
class SqlHelper(object):
def __init__(self, config):
self.connect = pymysql.connect(**eval(config))
self.connect.autocommit(True)
# default return tuple, DictCursor return Json .
self.cursor = self.connect.cursor()
def __enter__(self):
# DictCursor return Json .
self.cursor = self.connect.cursor(cursor=pymysql.cursors.DictCursor)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
self.connect.close()
def queryAll(self, sql, params=None):
"""
:param sql:
:param params:
:return:
"""
self.cursor.execute(sql, params)
return self.cursor.fetchall()
def queryMany(self, sql, num, params=None):
"""
:param sql:
:param num:
:param params:
:return:
"""
self.cursor.execute(sql, params)
return self.cursor.fetchmany(num)
def queryOne(self, sql, params=None):
"""
:param sql:
:param params:
:return:
"""
self.cursor.execute(sql, params)
return self.cursor.fetchone()
def operation(self, sql, params=None, DML=True):
"""
DML: insert / update / delete
DDL: CREATE TABLE/VIEW/INDEX/SYN/CLUSTER
:param DML:
:param sql:
:param params:
:return:
"""
try:
self.cursor.execute(sql, params)
except Exception as e:
if DML:
self.connect.rollback()
raise e
def batch_operation(self, sql_list, params_list=None, DML=True):
"""
Process multiple SQL files in batches .
:param DML:
:param sql_list:
:param params_list:
:return:
"""
for i in range(len(sql_list)):
try:
if params_list is not None:
self.operation(sql_list[i], params_list[i], DML)
else:
self.operation(sql_list[i], params_list, DML)
except Exception as e:
raise e
def batch_processing(self, sql, params_list, DML=True):
"""
The same SQL is executed multiple times in batches.
:param DML:
:param sql:
:param params_list:
:return:
"""
try:
self.cursor.executemany(sql, params_list)
except Exception as e:
if DML:
self.connect.rollback()
raise e
def __del__(self):
"""
Automatic disconnection
:return:
"""
if self.connect.open: # 解決連接重復(fù)關(guān)閉的
self.cursor.close()
self.connect.close()
1.正常方式執(zhí)行
if __name__ == '__main__':
sql = "select age from `demo`.`user` where name= 'Amy';"
env = EnvironmentVarUtils() # 初始化對象加載env文件數(shù)據(jù)
config = env.get_os_environ("DB_INFO") # 獲取指定key數(shù)據(jù)
# 1.正常方式執(zhí)行
db = SqlHelper(config)
result = db.queryOne(sql)
print(result)
-------------------------------------------------------------------------------
控制臺輸出：
Loading environment variables from 【C:\Users\lenovo\PycharmProjects\job\httpRunner_demo\.env】
Set OS environment variable: USERNAME
Set OS environment variable: PASSWORD
Set OS environment variable: BASE_URL
Set OS environment variable: db_info
(18,)
Process finished with exit code 0
2.通過上下文管理的方式執(zhí)行
if __name__ == '__main__':
sql = "select age from `demo`.`user` where name= 'Amy';"
env = EnvironmentVarUtils() # 初始化對象加載env文件數(shù)據(jù)
config = env.get_os_environ("DB_INFO") # 獲取指定key數(shù)據(jù)
# 2.上下文管理方式執(zhí)行
with SqlHelper(config) as db:
result = db.queryOne(sql)
print(result)
-------------------------------------------------------------------------------
控制臺輸出：
Loading environment variables from 【C:\Users\lenovo\PycharmProjects\job\httpRunner_demo\.env】
Set OS environment variable: USERNAME
Set OS environment variable: PASSWORD
Set OS environment variable: BASE_URL
Set OS environment variable: db_info
{'age': 18}
Process finished with exit code 0
以上親測可用贿讹，可以嘗試在自動化項目實踐中操作。有問題歡迎留言討論够掠。
