本文最后更新于 1509 天前，其中的信息可能已经有所发展或是发生改变。
# 多行输出结果
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"
1. 连接mysql并创建数据库testdb
官方api: https://pymysql.readthedocs.io/en/latest/user/examples.html#crud
import pymysql
database = 'testdb'
db = pymysql.connect(host='localhost', user='root', password='testpass')
cursor = db.cursor()
cursor.execute('SELECT VERSION()')
data_fetchone = cursor.fetchone() # fetchmany(), fetchall()不能获取到任何字符
print('Database version(fetchone): ', data_fetchone)
sql = "CREATE DATABASE IF NOT EXISTS {database} DEFAULT CHARACTER SET utf8mb4"
cursor.execute(sql.format(database=database))
db.close()
1
Database version(fetchone): ('8.0.12',)
D:\Anaconda3\lib\site-packages\pymysql\cursors.py:170: Warning: (1007, "Can't create database 'testdb'; database exists")
result = self._query(query)
1
通过 PyMySQL 的 connect() 方法声明了一个 MySQL 连接对象，连接对象参数有IP，用户，密码，端口等。
连接成功后需要调用cursor()方法获取mysql的操作游标，利用游标和execute()方法执行相应sql语句。
SELECT VERSION()
是获取mysql版本信息，cursor.fetchone()获取第一条数据，即版本号。
CREATE DATABASE IF NOT EXISTS {database} DEFAULT CHARACTER SET utf8mb4
如果不存在指定名字的数据库则创建，并修改默认字符集是utf8mb4，utf8mb4是utf8的扩充版
2. 创建数据表 — CREATE TABLE
import pymysql
db = pymysql.connect(host='localhost', port=3306,
user='root', password='testpass',
db='testdb',
charset='utf8mb4')
cursor = db.cursor()
然后接下来新创建一个数据表，执行创建表的 SQL 语句即可，创建一个用户表 test_table，在这里指定五个字段，如下:
| 字段名 | 含义 | 类型 |
|---|---|---|
| name | 姓名 | varchar |
| fullname | 全名 | varchar |
| birth | 出生年 | int |
| knownfor | 成就 | varchar |
| nationality | 国籍 | varchar |
# 使用 execute() 方法执行 SQL，如果表存在则删除
cursor.execute("DROP TABLE IF EXISTS test_table")
# 使用预处理语句创建表
sql = """CREATE TABLE test_table (
name VARCHAR(255) NOT NULL,
fullname VARCHAR(255) NOT NULL,
birth INT NOT NULL,
knownfor VARCHAR(255) NOT NULL,
nationality VARCHAR(255) NOT NULL
)"""
cursor.execute(sql)
db.close()
0
0
3. 插入数据 — INSERT INTO
- 下面使用的 test_json.json 是python文件读写-txt, json一中“3.txt与json读写综合例子”生成的文件
- try,except,else,finally.见jupyter notebook与python基础命令
import json
import pymysql
db = pymysql.connect(host='localhost', port=3306,
user='root', password='testpass',
db='testdb',
charset='utf8mb4')
cursor = db.cursor() # 游标最好不要重复利用
with open('test_json.json', 'r') as file:
txt = file.read()
datas = json.loads(txt)
sql = '''INSERT INTO test_table(name, fullname, birth, knownfor, nationality)
VALUES ("{NAME}", "{FULL NAME}", "{BIRTH}", "{KNOWN FOR}", "{NATIONALITY}")
'''
# 把json里的数据上传至服务器
for data in datas:
try:
# 执行sql语句
cursor.execute(sql.format(**data))
# 提交到数据库执行
db.commit()
except pymysql.err.ProgrammingError as e:
print(e)
except:
# 如果发生错误则回滚
db.rollback()
finally:
db.close()
需要执行 db 对象的 commit() 方法才可实现数据插入，这个方法才是真正将语句提交到数据库执行的方法，对于数据插入、更新、删除操作都需要调用该方法才能生效。
接下来我们加了一层异常处理，如果执行失败，则调用rollback() 执行数据回滚，相当于什么都没有发生过一样。
在这里就涉及一个事务的问题，事务机制可以确保数据的一致性，也就是这件事要么发生了，要么没有发生，比如插入一条数据，不会存在插入一半的情况，要么全部插入，要么整个一条都不插入，这就是事务的原子性，另外事务还有另外三个属性，一致性、隔离性、持久性，通常成为 ACID 特性。
事务机制
事务机制属性
| 属性 | 解释 |
|---|---|
| 原子性（atomicity） | 一个事务是一个不可分割的工作单位，事务中包括的诸操作要么都做，要么都不做。 |
| 一致性（consistency） | 事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。 |
| 隔离性（isolation） | 一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的，并发执行的各个事务之间不能互相干扰。 |
| 持久性（durability） | 持续性也称永久性（permanence），指一个事务一旦提交，它对数据库中数据的改变就应该是永久性的。接下来的其他操作或故障不应该对其有任何影响。 |
插入、更新、删除操作都是对数据库进行更改的操作，更改操作都必须为一个事务，所以对于这些操作的标准写法就是：
try:
cursor.execute(sql)
db.commit()
except:
db.rollback()
finally:
db.close()
4. 查询数据
- fetchone(): 该方法获取下一个查询结果集。结果集是一个对象
- fetchall(): 接收全部的返回结果行.
- rowcount: 这是一个只读属性，并返回执行execute()方法后影响的行数。
import pymysql
db = pymysql.connect(host='localhost', port=3306,
user='root', password='testpass',
db='testdb',
charset='utf8mb4')
cursor = db.cursor() # 游标最好不要重复利用
sql = 'SELECT * FROM %s WHERE %s <= %d' % ("test_table", "birth", 1860)
try:
cursor.execute(sql)
print('Count:', cursor.rowcount)
one = cursor.fetchone()
print('One:', one)
results = cursor.fetchall()
print('Results:', results)
print('Results Type:', type(results))
for row in results:
print(row)
except pymysql.err.ProgrammingError as e:
print(e)
except:
print ("Error: unable to fetch data")
finally:
db.close()
Count: 12
One: ('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany')
Results: (('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK'), ('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands'), ('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany'), ('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK'), ('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands'), ('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany'), ('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK'), ('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands'), ('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany'), ('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK'), ('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands'))
Results Type: <class 'tuple'>
('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK')
('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands')
('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany')
('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK')
('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands')
('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany')
('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK')
('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands')
('Planck', 'Max Karl Ernst Ludwig Planck', 1858, 'The Planck–Einstein relation', 'Germany')
('Faraday', 'Michael Faraday', 1791, 'Electromagnetic induction', 'UK')
('Lorentz', 'Hendrik Antoon Lorentz', 1853, 'Lorentz force', 'Netherlands')
5. 更新数据
5.1 简单更新
数据更新操作实际上也是执行 SQL 语句，最简单的方式就是构造一个 SQL 语句然后执行：
sql = 'UPDATE students SET age = %s WHERE name = %s'
try:
cursor.execute(sql, (25, 'Bob'))
db.commit()
except:
db.rollback()
finally:
db.close()
在这里同样是用占位符的方式构造 SQL，然后执行 excute() 方法，传入元组形式的参数，同样执行 commit() 方法执行操作。
如果要做简单的数据更新的话，使用此方法是完全可以的。
5.2 INSERT INTO 边插入边更新（此例需要设置id为主键, 关于主键设置见下文 7. 关于重复数据）：
但是在实际数据抓取过程中，在大部分情况下是需要插入数据的，但是我们关心的是会不会出现重复数据，如果出现了重复数据，我们更希望的做法一般是更新数据而不是重复保存一次，另外就是像上文所说的动态构造 SQL 的问题，所以在这里我们在这里重新实现一种可以做到去重的做法，如果重复则更新数据，如果数据不存在则插入数据，另外支持灵活的字典传值。
data = {
'id': '20120001',
'name': 'Bob',
'age': 20
}
table = 'students'
keys = ', '.join(data.keys())
values = ', '.join(['%s'] * len(data))
sql = 'INSERT INTO {table}({keys}) VALUES ({values})'.format(table=table, keys=keys, values=values)
try:
if cursor.execute(sql, tuple(data.values())):
print('Successful')
db.commit()
except:
print('Failed')
db.rollback()
db.close()
在这里完整的 SQL 构造出来是这样的：
INSERT INTO students(id, name, age) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE id = %s, name = %s, age = %s
6. 删除数据
删除操作相对简单，使用 DELETE 语句即可，需要指定要删除的目标表名和删除条件，而且仍然需要使用 db 的 commit() 方法才能生效，实例如下：
table = 'students'
condition = 'age > 20'
sql = 'DELETE FROM {table} WHERE {condition}'.format(table=table, condition=condition)
try:
cursor.execute(sql)
db.commit()
except:
db.rollback()
db.close()
在这里我们指定了表的名称，删除条件。因为删除条件可能会有多种多样，运算符比如有大于、小于、等于、LIKE等等，条件连接符比如有 AND、OR 等等，所以不再继续构造复杂的判断条件，在这里直接将条件当作字符串来传递，以实现删除操作。
7. 关于重复数据
7.1 重复数据的预防:
在数据表中指定字段为PRIMARY KEY(主键)或者 UNIQUE（唯一） 索引来保证数据的唯一性。
下表中无索引及主键，所以该表允许出现多条重复记录。
CREATE TABLE person_tbl
(
first_name CHAR(20),
last_name CHAR(20),
sex CHAR(10)
);
如果你想设置表中字段first_name，last_name数据不能重复，你可以设置双主键模式来设置数据的唯一性， 如果你设置了双主键，那么那个键的默认值不能为NULL，可设置为NOT NULL。如下所示：
CREATE TABLE person_tbl
(
first_name CHAR(20) NOT NULL,
last_name CHAR(20) NOT NULL,
sex CHAR(10),
PRIMARY KEY (last_name, first_name)
);
如果我们设置了唯一索引，那么在插入重复数据时，SQL语句将无法执行成功,并抛出错。
INSERT IGNORE INTO与INSERT INTO的区别就是INSERT IGNORE会忽略数据库中已经存在的数据，如果数据库没有数据，就插入新的数据，如果有数据的话就跳过这条数据，并发出警告。
REPLACE INTO into如果存在primary 或 unique相同的记录，则先删除掉。再插入新记录。
另一种设置数据的唯一性方法是添加一个UNIQUE索引，如下所示：
CREATE TABLE person_tbl
(
first_name CHAR(20) NOT NULL,
last_name CHAR(20) NOT NULL,
sex CHAR(10),
UNIQUE (last_name, first_name)
);
7.2 重复数据的删除
7.2.1 添加主键
在数据表中添加 INDEX（索引） 和 PRIMAY KEY（主键）这种简单的方法来删除表中的重复记录。方法如下：
mysql> ALTER IGNORE TABLE person_tbl
-> ADD PRIMARY KEY (last_name, first_name);
7.2.2 手动删除
mysql> CREATE TABLE tmp SELECT last_name, first_name, sex FROM person_tbl GROUP BY (last_name, first_name, sex);
mysql> DROP TABLE person_tbl;
mysql> ALTER TABLE tmp RENAME TO person_tbl;
pymysql常用功能打包成类
不推荐，推荐使用元类(metaclass)
import pymysql as mysql # pip install PyMySQL Python3的安装方式
#引入python中的traceback模块，跟踪错误
import traceback
#引入sys模块
import sys
class MysqlUtil(object):
def __init__(self, ip_addr, user, pswd, database, port=3306, charset='utf8mb4'):
"""
初始化
param: host, port, user, pswd, database, charset
"""
self.ip_addr = ip_addr
self.port = port
self.user = user
self.pswd = pswd
self.database = database
self.charset = charset
def getConnect(self):
"""
获取数据库的连接
"""
db = mysql.connect(host=self.ip_addr, port=self.port,
user=self.user, password=self.pswd,
db=self.database, charset=self.charset)
return db
def createTable(self, tableName, sql):
'''
创建数据表
tableName：数据表名称
sql： 数据表的创建sql语句
'''
# 获取数据库连接
db = self.getConnect()
#使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
# 使用execute()方法执行sql ，如果表存在则删除
cursor.execute("drop table if exists %s" %(tableName))
# 使用预处理语句创建表
cursor.execute(sql)
# 关闭数据库连接
db.close()
def insertTable(self,sql):
'''
插入数据库
sql:插入数据库的sql语句
'''
#获取数据库连接
db = self.getConnect()
#使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
try:
#执行sql语句
cursor.execute(sql)
#提交到数据库执行
db.commit()
except Exception: #方法一：捕获所有异常
#如果发生异常，则回滚
print("发生异常",Exception)
db.rollback()
finally:
#最终关闭数据库连接
db.close()
def fetchone(self,sql):
'''
查询数据库：单个结果集
fetchone(): 该方法获取下一个查询结果集。结果集是一个对象
'''
# 获取数据库连接
db = self.getConnect()
# 使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
try:
# 执行sql语句
cursor.execute(sql)
result = cursor.fetchone()
except: #方法二：采用traceback模块查看异常
#输出异常信息
traceback.print_exc()
# 如果发生异常，则回滚
db.rollback()
finally:
# 最终关闭数据库连接
db.close()
return result
def fetchall(self,sql):
'''
查询数据库：多个结果集
fetchall(): 接收全部的返回结果行.
'''
# 获取数据库连接
db = self.getConnect()
# 使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
try:
# 执行sql语句
cursor.execute(sql)
results = cursor.fetchall()
except: #方法三：采用sys模块回溯最后的异常
#输出异常信息
info = sys.exc_info()
print( info[0], ":", info[1])
# 如果发生异常，则回滚
db.rollback()
finally:
# 最终关闭数据库连接
db.close()
return results
def delete(self,sql):
'''
删除结果集
'''
# 获取数据库连接
db = self.getConnect()
# 使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
try:
# 执行sql语句
cursor.execute(sql)
db.commit()
except: #如果你还想把这些异常保存到一个日志文件中，来分析这些异常
#将错误日志输入到目录文件中
f = open("c:log.txt", 'a')
traceback.print_exc(file=f)
f.flush()
f.close()
# 如果发生异常，则回滚
db.rollback()
finally:
# 最终关闭数据库连接
db.close()
def update(self,sql):
'''
更新结果集
'''
# 获取数据库连接
db = self.getConnect()
# 使用cursor() 方法创建一个游标对象 cursor
cursor = db.cursor()
try:
# 执行sql语句
cursor.execute(sql)
db.commit()
except:
# 如果发生异常，则回滚
db.rollback()
finally:
# 最终关闭数据库连接
db.close()
app = MysqlUtil()
#======================创建表结构==============================
#表名称
tableName = "user"
#见表语句
tableSql = """
CREATE table user(
id bigint(20) NOT NULL AUTO_INCREMENT,
name varchar(50) NOT NULL,
sex varchar(2) NOT NULL,
age int,
PRIMARY KEY (`id`)
)
"""
#调用创建表语句
# app.createTable(tableName,tableSql)
#====================================================
#=======================插入数据库=============================
insertSql = "INSERT INTO user(name,sex,age) \
VALUES ('%s', '%s', '%d')" % \
('jack', '女', 18)
# app.insertTable(insertSql)
#=======================查询单个数据库=============================
fetchoneSql = "SELECT * FROM user WHERE id = '%d'" % (1)
user = app.fetchone(fetchoneSql)
print(user)
#=======================查询多个数据库=============================
fetchallSql = "SELECT * FROM user WHERE id > '%d'" % (0)
users = app.fetchall(fetchallSql)
print(users)
#=======================更新数据库=============================
updateSql = "update user set age = age + 1 ,sex = '%s' where id = '%d'" % ('未知',1)
# app.update(updateSql)
#=======================删除数据库=============================
delSql = "delete from user where id = '%d'" % (1)
app.delete(delSql)
# fetchoneSql = "SELECT * FROM user WHERE id = '%d'" % (1)
# user = app.fetchone(fetchoneSql)
# print(user)