Mysql驅(qū)動介紹
MySQL-python(棄用)：也就是MySQLdb，是對C語言操作MySQL數(shù)據(jù)庫的一個簡單封裝。遵循了Python DB API v2。但是只支持Python2，目前還不支持Python3。
mysqlclient(安裝難,但django官方默認(rèn))：是MySQL-python的另外一個分支，支持Python3并且修復(fù)了一些bug。
pymysql：純Python實現(xiàn)的一個驅(qū)動，因為是純Python編寫的，因此執(zhí)行效率不如MySQL-python。并且也因為是純Python編寫的，因此可以和Python代碼無縫銜接。
MySQL Connector/Python：MySQL官方推出的使用純Python連接MySQL的驅(qū)動。因為是純Python開發(fā)的，效率不高。
推薦參考
Python MySQL 數(shù)據(jù)庫連接 (一) mysql-connector-python
Python MySQL 數(shù)據(jù)庫連接 (三) mysqlclient django默認(rèn)使用
安裝
pip install pymysql
使用案例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import logging
import pymysql
class DBQuery:
"""
1
sqluser = DBQuery().ment("SELECT * FROM `user` ") # 直接使用
2
sqldb = DBQuery() # sql 初始化
fimes = sqldb.ment("SELECT * FROM tables WHERE username=%s AND id=%s", (usernmae, 1)) # 自動關(guān)閉當(dāng)前鏈接
3
sqldb = DBQuery() # sql 初始化
fimes = sqldb.ment("SELECT * FROM WHERE username=%s AND id=%s", (usernmae, 1), close=False) # 不關(guān)當(dāng)前鏈接
4
"""
def __init__(self, **kwargs):
self.mysql_config = {
'user': 'user',
'password': 'password',
'host': '127.0.0.1',
'database': 'database',
'charset': 'utf8',
'connect_timeout': 10,
}
self.kwargs = kwargs
self.replaceconfig()
self.config = self.mysql_config.copy()
self.conn = None
self.cur = None
def replaceconfig(self):
""" 替換參數(shù) """
for k, v in self.kwargs.items():
self.mysql_config[k] = v
def connect(self):
""" 數(shù)據(jù)庫連接函數(shù),建立連接和游標(biāo) """
if self.conn and self.cur:
return True
try:
self.conn = pymysql.connect(self.config['host'], self.config['user'], self.config['password'],
self.config['database'], charset=self.config['charset'],
connect_timeout=self.config['connect_timeout'],
cursorclass=pymysql.cursors.DictCursor)
self.cur = self.conn.cursor()
except Exception:
logging.error("數(shù)據(jù)庫連接失敗, %s" , exc_info=True)
return
return True
def ment(self, statement, parameter=None, close=True):
""" 數(shù)據(jù)執(zhí)行語句函數(shù) ,statement:語句,parameter:參數(shù),必須是 元組 或者 列表,返回執(zhí)行結(jié)果"""
# print(statement,parameter)
# query = self.cur.mogrify(statement, parameter) # 返回格式化好的sql語句
# print(query)
if not self.connect():
return False
try:
self.cur.execute(statement, parameter)
if 'SELECT' == statement.strip().upper()[0:6] or 'SHOW' == statement.strip().upper()[0:4]:
result = self.cur.fetchall()
elif 'CREATE TABLE' == statement.strip().upper()[0:12]:
result = True
else:
result = self.conn.commit()
if statement.upper()[0:6] in ['INSERT', 'UPDATE']:
result = self.rowcount()
except Exception:
logging.error('SQL_ERROR: statement:%s' % statement, exc_info=True)
self.conn.rollback()
result = False
finally:
self.mclose(close)
return result
def lastd(self):
""" 返回 INSERT 語句的id, 應(yīng)該在插入后立即使用 """
try:
return self.cur.lastrowid
except:
pass
def rowcount(self):
""" 返回 INSERT or UPDATE 語句的更新數(shù)目 """
try:
return self.cur.rowcount
except:
pass
def mclose(self, close=True):
if close:
self.dbclose()
def dbclose(self):
try:
if self.cur:
self.cur.close()
if self.conn:
self.conn.close()
return True
except:
pass
其他
-
參數(shù)轉(zhuǎn)義,避免sql注入
import pymysql
parameter = "I am a parameter, '# I want to inject!'"
pymysql.escape_string(parameter)
# 輸出 I am a parameter, \'# I want to inject!\'
-
返回格式化好的sql語句
query = self.cur.mogrify(statement, parameter)
print(query)
-
官網(wǎng)
