跳转至

Python 中 SQL 数据库的操作

约 1747 个字 129 行代码 2 张图片 预计阅读时间 7 分钟

Info

同一数据库，操作的模块有多个。这里每个数据库只介绍一个模块。

总论

常用的支持 SQL 的数据库管理系统

数据库管理系统（Database Management System, DBMS），一般也可以叫数据库引擎（Database Engine）

以下只考虑可以自己部署的，不考虑云服务提供商提供的云数据库服务。

企业级 DBMS：企业使用需要购买授权

  • Oracle
  • Microsoft SQL Server
  • IBM DB2

开源的 DBMS

  • MySQL：隶属于 Oracle，最常用的开源 DBMS
  • MariaDB：由 MySQL 分支而来，兼容 MySQL
  • PostgreSQL：最近几年在一些新的项目上常用
  • SQLite：可以在内存、文件上建立本地数据库，常用于单机的数据管理；性能不及普通的数据库

数据库引擎使用量的趋势
数据库引擎使用量的趋势

支持 SQL 的数据库操作基本步骤

Python 数据库 API 的格式由 PEP 249 定义，遵循该标准的模块的操作方式非常相似，故相似的功能不会细讲。

  1. 建立连接（connection）
    connect()
  2. 建立游标（cursor）
    cursor()
  3. 使用游标执行（execute） SQL 语句
    execute(), executemany()
  4. 读取（fetch）游标对应的数据
    fetchone(), fetchmany(), fetchall(), rowcount...
  5. 如果对数据库进行了写操作，可能要对连接进行提交（commit）
    commit(), rollback()
  6. 关闭（close）游标、连接
    close()

防止 SQL 注入

绝对不能直接拼接 SQL 语句。

关于 SQL 注入的一个梗图
关于 SQL 注入的一个梗图

现在的数据库操作模块基本上都会提供一些手段，处理传入的值，防止 SQL 注入。

SQLite

可用 sqlite3 标准库。

建立连接与游标

1
conn = sqlite3.connect(路径)

路径可以是类路径对象（path-like object，包括字符串、bytes、实现 os.PathLike 协议的对象），表示数据库文件（一般扩展名为 .db）。如果路径对应文件不存在，则会自动创建。

如果路径为 ':memory:'，则表示在内存中打开数据库。

建立连接后，就可以以此为基础建立游标：

1
cursor = conn.cursor()

执行 SQL 语句

执行一条 SQL 语句：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
cursor.execute(SQL字符串)
cursor.execute(带占位符的SQL字符串, 变量值构成的元组)

# 例
cursor.execute('select name from emp_info')
# 第二个参数一般来说会写成元组，不过列表也可以
cursor.execute('select name from emp_info where gender = ? and dept = ?', ('M', 'tech'))
cursor.execute(
    'select name from emp_info where gender = :gender and dept = :dept',
    {'gender': 'M', 'dept': 'tech'}
)

批量执行多条类似的 SQL 语句：

1
2
3
4
5
6
cursor.executemany(带占位符的SQL字符串, 变量值构成的元组组成的列表)
# 例
cursor.executemany(
    'select name from emp_info where gender = ? and dept = ?',
    [('M', 'tech'), ('F', 'hr'), ('M', 'test')]
)

提交更改：

1
conn.commit()

获取执行结果

前例的查询的返回值即为一个可迭代对象，迭代每一行；每一行为一个元组，其中的元素为该行每一列的值

1
2
for row in cursor.execute(SQL字符串):
    print(row)

1
2
(1, 'val2', 'val3')
(2, 'val4', 'val5')

也可以返回一行组成的元组，一个一个接着迭代，直到完毕返回 None：

1
cursor.fetchone()

也可以直接返回多行，甚至整个查询结果组成的列表：

1
2
cursor.fetchmany(表示行数的整数，默认为1)
cursor.fetchall()

sqlite3.Row 对象

在建立连接后定义如下，可使执行 SQL 语句后，返回的行（háng）为 Row 对象：

1
conn.row_factory = sqlite3.Row

查看各列列名：返回列表：

1
row.keys()

1
['date', 'trans', 'symbol', 'qty', 'price']

可以通过下标和列名访问其中的值，也可以查看长度等属性：

1
2
3
row[1]
row[列名]
len(row)

可以转换为元组：

1
tuple(row)

关闭连接

1
conn.close()

未提交的更改会丢失。

另外，有打开和关闭操作的对象，都可以试着以 with 语句代替。

MySQL

使用第三方模块 PyMySQL（pymysql）。

例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import pymysql

conn = pymysql.connect( # 建立连接
    host='主机地址', user='用户名', password='密码', database='数据库名',
    cursorclass=pymysql.cursors.DictCursor
)

with conn:
    with conn.cursor() as cursor:   # 建立游标
        sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
        cursor.execute(sql, ('webmaster@python.org', 'very-secret'))    # 执行 SQL
    conn.commit()   # 提交

    with conn.cursor() as cursor:   # 建立游标
        sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
        cursor.execute(sql, ('webmaster@python.org',))  # 执行 SQL
        result = cursor.fetchone()  # 获取结果
        print(result)

1
{'password': 'very-secret', 'id': 1}

连接数据库

1
2
3
4
5
6
7
pymysql.connect(
    host='主机地址', port=端口号,   # 默认为 3306
    user='用户名', password='密码', database='数据库名',    # 可以用 None 表示不选择数据库
    charset='编码'
    # 指定连接使用的游标，默认为 pymysql.cursors.Cursor
    cursorclass=pymysql.cursors.DictCursor
)

MySQL 里面的编码较常用的是 utf8mb4，相比于 utf8，支持更多字符（如 emoji）

执行 SQL 语句

执行 SQL 语句的返回结果，一般是一个整数，表示影响了多少行。

executemany() 方法的返回结果还可能是 None。

参数的格式如下：

1
2
3
4
5
6
# 虽然可以定义占位符的数据类型，但实际上 %s 可以满足绝大多数情况
cursor.execute('select name from emp_info where gender = %d and dept = %s', (1, 'tech'))
cursor.execute(
    'select name from emp_info where gender = %(gender)d and dept = %(dept)s',
    {'gender': 1, 'dept': 'tech'}
)

使用参数时，语句中如有百分号，则双写。

可通过 mogrify() 方法，查看 execute() 方法将会执行的语句（将 execute 替换为 mogrify 即可）。

获取执行结果

根据建立连接时使用的 cursorclass，执行结果的表达方式也不一样。

如果是 pymysql.cursors.Cursor（默认值），则返回的结果，每一行用元组表示。

如果是 pymysql.cursors.DictCursor，则每一行用字典表示。

关闭游标和连接

游标可以关闭，方法和关闭连接一样。

1
cursor.close()

先关闭游标，再关闭连接。

也可以用 with 语句，自动关闭。

Microsoft SQL Server

使用第三方模块 pymssql。

例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import pymssql

conn = pymssql.connect(server='服务器地址', user='用户名', password='密码', database='数据库名')
cursor = conn.cursor()

cursor.execute("""
IF OBJECT_ID('persons', 'U') IS NOT NULL DROP TABLE persons
CREATE TABLE persons (id INT NOT NULL, name VARCHAR(100), salesrep VARCHAR(100), PRIMARY KEY(id))
""")
cursor.executemany(
    "INSERT INTO persons VALUES (%d, %s, %s)",
    [(1, 'John Smith', 'John Doe'), (2, 'Jane Doe', 'Joe Dog'), (3, 'Mike T.', 'Sarah H.')])
conn.commit()

cursor.execute('SELECT * FROM persons WHERE salesrep=%s', 'John Doe')
row = cursor.fetchone()
while row:
    print("ID=%d, Name=%s" % (row[0], row[1]))
    row = cursor.fetchone()

conn.close()

建立连接和游标

和 pymysql 的比较明显的区别：如果想让游标返回结果返回字典格式，既可以定义游标时设置，也可以在定义连接时设置。

1
2
3
4
5
6
7
conn = pymssql.connect(
    server='服务器地址', user='用户名', password='密码', database='数据库名', port=端口号,
    as_dict=布尔值
    # 端口号默认为 1433，数据库默认值为空字符串，as_dict 默认为 False
)
cursor = conn.cursor()  # 如果建立连接时 as_dict 为 False，且想要返回元组
cursor = conn.cursor(as_dict=True)  # 如果想要返回字典

执行 SQL 语句

参数的格式用元组形式：

1
cursor.execute('select name from emp_info where gender = %d and dept = %s', (1, 'tech'))

只有一个参数时，可直接写出：

1
cursor.execute('select name from emp_info where gender = %d', 1)

游标在查询后本身可迭代，可以通过遍历游标，得到结果：

1
2
for row in cursor:
    print('row = %r' % (row,))

不过更建议使用通用的方式读结果，以保证可移植性。

PostgreSQL

使用第三方模块 Psycopg，目前有 2、3 版本之分，支持同步和异步操作，这里只介绍版本 3 的同步操作。

安装

首先确保 pip 的版本至少在 20.3，否则更新：

1
pip install --upgrade pip

安装时，如果只写名称，得到的是纯 Python 版本，需要另外安装 libpq 库；

如果想直接使用，需这么安装：

1
pip install psycopg[binary,pool]

例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import psycopg

with psycopg.connect("dbname=test user=postgres") as conn:
    with conn.cursor() as cur:
        cur.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data text)")
        cur.execute("INSERT INTO test (num, data) VALUES (%s, %s)", (100, "abc'def"))
        cur.execute("SELECT * FROM test")
        cur.fetchone()    # 返回 (1, 100, "abc'def")
        for record in cur:    # 游标可遍历
            print(record)
        conn.commit()

建立连接

连接时，需要写连接字符串：

1
psycopg.connect("连接字符串")

连接字符串格式如下：

  • 直接写 参数=值 的格式，中间用空格隔开，值可用单引号包裹；如值里面有单引号和反斜杠，需用反斜杠转义。如：

    1
    host=主机地址 port=端口号 dbname=数据库名 user=用户名 password=密码 connect_timeout=超时秒数

  • 写成 URI 的形式：

    1
    postgresql://[[用户名][:密码]@[主机地址][:端口号]][/数据库名][?参数名1=参数值1[&参数名2=参数值2]]

    这里面的方括号表示：里面的值是可选的；不在字符串中。

执行 SQL 语句

连接和游标均可以发起执行方法（连接仅能用 execute()，游标可用其他）。

execute() 方法返回自身，自身也可迭代。

参数支持元组（也可使用列表）、字典形式。元组形式的占位符只能用 %s，不可以用其他格式。

上述参数只可用于定义值，不可用于表名等；如果需要，需进行如下拼接：

1
2
3
4
cursor.execute(
    psycopg.sql.SQL("INSERT INTO {} VALUES (%s)").format(psycopg.sql.Identifier('numbers')),
    (10,)
)

回滚

使用 Psycopg 时，每个执行都是一个事务。如果执行失败，则必须要回滚，才可继续（这与其他数据库操作模块不一样）：

1
conn.rollback()

使返回行（háng）为字典

建立连接或游标时，定义 row_factory 为 psycopg.rows.dict_row：

1
2
3
4
5
from psycopg.rows import dict_row

conn = psycopg.connect(DSN, row_factory=dict_row)
conn.execute("select 'John Doe' as name, 33 as age").fetchone()
# {'name': 'John Doe', 'age': 33}

1
2
3
cur = conn.cursor(row_factory=dict_row)
cur.execute("select 'John Doe' as name, 33 as age").fetchone()
# {'name': 'John Doe', 'age': 33}

总结：各数据库操作对比

数据库 SQLite MySQL Microsoft SQL Server PostgreSQL
模块 sqlite3 pymysql pymssql psycopg
连接用的基本参数 表示路径的字符串 主机、端口、用户名、密码、数据库 主机、端口、用户名、密码、数据库 表示连接信息的字符串
支持的替换参数组类型 元组等可迭代对象
字典		 元组等可迭代对象
字典		 元组
只有一个可直接写		 元组、列表
字典
执行 SQL 的参数占位符 ?
:键		 %数据类型（通常为 %s）
%(键)数据类型		 %数据类型 %s
%(键)s
游标可迭代？ 否 否 是 是
执行查询返回 可迭代对象 整数，影响多少行；或 None None 游标
获取字典结果 无 建立连接时定义 cursorclass 建立连接或游标时定义 as_dict 建立连接或游标时定义 row_factory
游标可关闭？ 否 是 是 是

参考资料