Python 第三方模块之 PyMySQL - Python 操作 MySQL

1、PyMySQL 安装

pip install PyMySQL

2、语法

连接数据库

import pymysql
 
 
db = pymysql.connect(
    host='localhost', 
    user='root', 
    password="root",
    database='db', 
    port=3306,        # 数字3306
    charset='utf8',   # 不是utf-8
    autocommit=True   # autocommit=True 让每次提交都去调用 commit 函数 即更新事务
)                                                      # 连接数据库

cursor= db.cursor(cursor=pymysql.cursors.DictCursor) # cursor=pymysql.cursors.DictCursor 表示返回结果以字典形式返回（不加为元祖形式）
cursor.execute("SELECT VERSION()")                     # 使用 execute() 方法执行 SQL 查询
data = cursor.fetchone()                               # 使用 fetchone() 方法获取单条数据
print ("Database version : %s " % data)
db.close()                                             # 关闭数据库连接

创建表

import pymysql
 
 
db = pymysql.connect("localhost","testuser","test123","TESTDB" )
cursor = db.cursor()
 
# 使用 execute() 方法执行 SQL，如果表存在则删除
cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
 
# 使用预处理语句创建表
sql = """CREATE TABLE EMPLOYEE (
         FIRST_NAME  CHAR(20) NOT NULL,
         LAST_NAME  CHAR(20),
         AGE INT,  
         SEX CHAR(1),
         INCOME FLOAT )"""
 
cursor.execute(sql)
db.close()

增

import pymysql
 
 
db = pymysql.connect("localhost","testuser","test123","TESTDB" )
cursor = db.cursor()
 
# SQL 插入语句
sql = "INSERT INTO EMPLOYEE(FIRST_NAME, LAST_NAME, AGE, SEX, INCOME)  VALUES (%s, %s,  %s,  %s,  %s )" % ('Mac', 'Mohan', 20, 'M', 2000)
try:
    cursor.execute(sql)      # 执行sql语句
    print(cursor.lastrowid)  # 最后插入行的主键id,在 conn.commit() 之前执行
    print(db.insert_id())    # 最新插入行的主键id,在 conn.commit() 之前执行
    db.commit()              # 提交到数据库执行
except:
   db.rollback()         # 如果发生错误则回滚
 
db.close()
 
# 最好不要自己拼接sql语句，会有sql注入的漏洞，使用 execute（sql，参数）的方式可以防止
sql = "INSERT INTO EMPLOYEE(FIRST_NAME, LAST_NAME, AGE, SEX, INCOME) VALUES (%s, %s,  %s,  %s,  %s )"
cursor.execute(sql, ('Mac', 'Mohan', 20, 'M', 2000))    # 执行sql语句

源码

def execute(self, query, args=None):
    query = self.mogrify(query, args)

def mogrify(self, query, args=None):
    if args is not None:
        query = query % self._escape_args(args, conn)

def _escape_args(self, args, conn):
    if isinstance(args, (tuple, list)):
        return tuple(conn.literal(arg) for arg in args)

def literal(self, arg):
    # 简化版本
    for arg in args:
        arg = "'" + arg.replace("'", "''") + "'"

查

Python 查询 MySQL 使用 fetchone() 方法获取单条数据, 使用 fetchall() 方法获取多条数据。

• fetchone(): 该方法获取下一个查询结果集。结果集是一个对象
• fetchall(): 接收全部的返回结果行.
• rowcount: 这是一个只读属性，并返回执行execute()方法后影响的行数。

import pymysql
 
 
db = pymysql.connect("localhost","testuser","test123","TESTDB" )
cursor = db.cursor()
 
sql = "SELECT * FROM EMPLOYEE WHERE INCOME > %s" % (1000)
try:
   cursor.execute(sql)          # 执行SQL语句
   results = cursor.fetchall()  # 获取所有记录列表
   for row in results:
      fname = row[0]
      lname = row[1]
      age = row[2]
      sex = row[3]
      income = row[4]
      print ("fname=%s,lname=%s,age=%s,sex=%s,income=%s" % (fname, lname, age, sex, income ))
except:
   print ("Error: unable to fetch data")
 
db.close()

改

import pymysql
 
 
db = pymysql.connect("localhost","testuser","test123","TESTDB" )
cursor = db.cursor()
 
sql = "UPDATE EMPLOYEE SET AGE = AGE + 1 WHERE SEX = '%c'" % ('M')
try:
   cursor.execute(sql)  # 执行SQL语句
   db.commit()          # 提交到数据库执行
except Exception as e:
   db.rollback()        # 发生错误时回滚
 
db.close()

删

import pymysql
 
 
db = pymysql.connect("localhost","testuser","test123","TESTDB" ) 
cursor = db.cursor()
 
sql = "DELETE FROM EMPLOYEE WHERE AGE > %s" % (20)
try:
   cursor.execute(sql)  # 执行SQL语句
   db.commit()          # 提交修改
except:
   db.rollback()        # 发生错误时回滚
db.close()

问题

pymysql.err.InterfaceError: (0, '')

原因：因为这个数据库的连接建立太久了，会自动断开，这个时候我们需要重新建立连接，否则访问接口就会出现异常报错了。

解决办法：我们可以在每次操作SQL之前对连接进行检查，如果发现连接已经断开，则进行重连。在源码中，如果使用了方法 ping(reconnect=True) ，那么可以在每次连接之前，会检查当前连接是否已关闭，如果连接关闭则会重新进行连接，于是我们可以将其用于处理目前报错的问题，改动后的代码如下

import pymysql


class MYBMYSLQ(object):
 
    def __init__(self):
        self.conn = pymysql.connect("localhost","testuser","test123","TESTDB" ) 
        self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
 
    def cmdb_select(self, sql, sql_arg=()):
        self.conn.ping(reconnect=True)
        self.cursor.execute(sql, sql_arg)
        return self.cursor.fetchall()