Python数据库编程实践

数据库连接

在Python中进行数据库编程，首先要做的就是建立与数据库的连接。Python提供了多种数据库连接库，不同的数据库有相应适配的库，比如用于SQLite的sqlite3，用于MySQL的mysql - connector - python，用于PostgreSQL的psycopg2等。

SQLite数据库连接

SQLite是一种轻量级的嵌入式数据库，非常适合小型应用程序或快速原型开发。Python标准库中的sqlite3模块提供了操作SQLite数据库的接口。以下是一个简单的示例：

import sqlite3

# 连接到SQLite数据库，如果数据库不存在则会创建
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

在上述代码中，使用sqlite3.connect()方法连接到一个名为example.db的SQLite数据库。如果该数据库不存在，会自动创建。conn.cursor()方法则是创建一个游标对象，通过这个游标对象来执行SQL语句。

MySQL数据库连接

要连接MySQL数据库，需要先安装mysql - connector - python库，可以使用pip install mysql - connector - python进行安装。以下是连接MySQL数据库的示例：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

在这个示例中，mysql.connector.connect()方法用于建立连接。其中host指定数据库服务器地址，user和password分别是数据库的用户名和密码，database指定要连接的数据库名。

PostgreSQL数据库连接

对于PostgreSQL数据库，使用psycopg2库来进行连接。首先通过pip install psycopg2安装该库，然后连接示例如下：

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="test",
    user="postgres",
    password="password"
)
cursor = conn.cursor()

这里psycopg2.connect()方法的参数与MySQL连接有所不同，但基本原理相似，都是通过指定服务器地址、数据库名、用户名和密码来建立连接。

执行SQL语句

连接建立后，就可以通过游标对象执行SQL语句了。SQL语句可以分为查询语句（如SELECT）和非查询语句（如INSERT、UPDATE、DELETE）。

执行查询语句

查询语句用于从数据库中检索数据。以SQLite为例，假设我们有一个users表，包含id、name和age字段，以下是查询所有用户的代码：

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row)
conn.close()

在上述代码中，cursor.execute()方法执行SELECT语句，cursor.fetchall()方法获取查询结果的所有行，然后通过循环打印每一行数据。

如果只想获取一条记录，可以使用cursor.fetchone()方法，示例如下：

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

cursor.execute('SELECT * FROM users')
row = cursor.fetchone()
print(row)
conn.close()

对于MySQL和PostgreSQL，执行查询语句的方式类似，只是连接对象和游标对象的创建方式不同，执行SQL语句和获取结果的方法是一样的。

执行非查询语句

非查询语句用于对数据库进行数据插入、更新和删除操作。

插入数据

以MySQL为例，向users表插入一条新记录：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
val = ("John", 25)
cursor.execute(sql, val)
conn.commit()
print(cursor.rowcount, "记录插入成功。")
conn.close()

在这段代码中，cursor.execute()方法的第一个参数是SQL语句，第二个参数是一个元组，用于替换SQL语句中的占位符%s。执行完插入操作后，需要调用conn.commit()方法来提交事务，使插入操作生效。cursor.rowcount可以获取受影响的行数。

更新数据

假设要更新users表中name为John的用户的年龄，以PostgreSQL为例：

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="test",
    user="postgres",
    password="password"
)
cursor = conn.cursor()

sql = "UPDATE users SET age = %s WHERE name = %s"
val = (30, "John")
cursor.execute(sql, val)
conn.commit()
print(cursor.rowcount, "记录更新成功。")
conn.close()

同样，这里使用占位符%s，通过元组传递实际值。更新操作后也需要提交事务。

删除数据

以SQLite为例，删除users表中age大于30的用户记录：

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

sql = "DELETE FROM users WHERE age >?"
val = (30,)
cursor.execute(sql, val)
conn.commit()
print(cursor.rowcount, "记录删除成功。")
conn.close()

在SQLite中，删除语句中的占位符使用?，同样需要提交事务使删除操作生效。

使用参数化查询

参数化查询是一种在执行SQL语句时，将参数与SQL语句分开传递的技术。这样做的好处是可以防止SQL注入攻击，提高程序的安全性。前面在执行非查询语句时已经展示了简单的参数化查询示例，下面再详细说明其原理。

假设我们要根据用户输入的用户名查询用户信息，如果不使用参数化查询，可能会这样写（以MySQL为例，存在安全风险）：

import mysql.connector

username = input("请输入用户名: ")
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

sql = "SELECT * FROM users WHERE name = '" + username + "'"
cursor.execute(sql)
rows = cursor.fetchall()
for row in rows:
    print(row)
conn.close()

这种方式下，如果用户输入恶意的字符串，比如'; DROP TABLE users; --，就会导致整个users表被删除，这就是SQL注入攻击。

而使用参数化查询，代码如下：

import mysql.connector

username = input("请输入用户名: ")
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

sql = "SELECT * FROM users WHERE name = %s"
val = (username,)
cursor.execute(sql, val)
rows = cursor.fetchall()
for row in rows:
    print(row)
conn.close()

在参数化查询中，数据库驱动会对参数进行适当的转义和处理，确保输入的内容不会干扰SQL语句的逻辑，从而有效防止SQL注入攻击。

事务处理

事务是数据库操作的一个逻辑单元，它包含一组数据库操作语句，这些语句要么全部成功执行，要么全部不执行。在Python数据库编程中，不同的数据库库对事务的处理方式略有不同，但基本原理一致。

自动提交与手动提交

默认情况下，一些数据库连接是自动提交模式，即每条SQL语句执行后立即生效。例如，在SQLite中，默认是自动提交的。如果要手动控制事务，需要关闭自动提交。以SQLite为例：

import sqlite3

conn = sqlite3.connect('example.db')
# 关闭自动提交
conn.isolation_level = None

try:
    cursor = conn.cursor()
    cursor.execute('INSERT INTO users (name, age) VALUES ("Alice", 28)')
    cursor.execute('UPDATE users SET age = age + 1 WHERE name = "Alice"')
    conn.commit()
    print("事务提交成功。")
except Exception as e:
    conn.rollback()
    print("事务回滚，错误信息:", str(e))
finally:
    conn.close()

在上述代码中，通过设置conn.isolation_level = None关闭自动提交。然后在try块中执行一系列数据库操作，成功后调用conn.commit()提交事务。如果出现异常，通过conn.rollback()回滚事务，撤销之前执行的所有操作。

嵌套事务

在一些数据库中支持嵌套事务，但不同数据库的实现方式和语义有所不同。以PostgreSQL为例，虽然PostgreSQL不直接支持传统意义上的嵌套事务，但可以通过保存点（savepoint）来模拟嵌套事务的部分行为。

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="test",
    user="postgres",
    password="password"
)
cursor = conn.cursor()

try:
    # 创建保存点
    cursor.execute('SAVEPOINT my_savepoint')
    cursor.execute('INSERT INTO users (name, age) VALUES ("Bob", 30)')
    cursor.execute('UPDATE users SET age = age + 1 WHERE name = "Bob"')
    # 回滚到保存点
    cursor.execute('ROLLBACK TO SAVEPOINT my_savepoint')
    # 释放保存点
    cursor.execute('RELEASE SAVEPOINT my_savepoint')
    conn.commit()
    print("事务处理完成。")
except Exception as e:
    conn.rollback()
    print("事务回滚，错误信息:", str(e))
finally:
    conn.close()

在这个示例中，通过SAVEPOINT语句创建一个保存点，在保存点之后的操作如果出现问题，可以回滚到保存点，而不会影响保存点之前的操作。最后释放保存点，并提交事务。

处理数据库结果集

当执行查询语句后，会得到一个结果集。在Python中，不同的数据库库对结果集的处理方式有一些共性，也有一些差异。

结果集的遍历

前面已经展示了通过fetchall()和fetchone()方法获取结果集并遍历的方式。除了这两种方法，还可以使用fetchmany()方法获取指定数量的行。以SQLite为例：

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

cursor.execute('SELECT * FROM users')
rows = cursor.fetchmany(3)
for row in rows:
    print(row)
conn.close()

上述代码中，fetchmany(3)表示获取结果集中的前3行数据。

结果集的结构

结果集通常以元组或字典的形式返回，具体取决于使用的数据库库和设置。以MySQL为例，如果使用mysql - connector - python库，默认返回元组形式的结果集：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row[0], row[1], row[2])  # 按索引访问元组元素
conn.close()

如果希望以字典形式返回结果集，可以在连接时设置dictionary = True：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    dictionary=True
)
cursor = conn.cursor()

cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row['id'], row['name'], row['age'])  # 按字段名访问字典元素
conn.close()

这样在遍历结果集时，可以通过字段名来访问数据，使代码更具可读性。

数据库连接池

在实际应用中，如果频繁地创建和销毁数据库连接，会消耗大量的系统资源，影响应用程序的性能。数据库连接池技术可以解决这个问题。连接池在应用程序启动时创建一定数量的数据库连接，并将这些连接保存在池中。当应用程序需要连接数据库时，从连接池中获取一个连接，使用完毕后再将连接放回池中。

使用DBUtils实现连接池

DBUtils是一个流行的Python数据库连接池库，支持多种数据库。以下以MySQL为例，展示如何使用DBUtils创建连接池： 首先通过pip install DBUtils安装该库。

from dbutils.pooled_db import PooledDB
import mysql.connector

# 创建连接池
pool = PooledDB(
    creator=mysql.connector,
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    autocommit=True,
    maxconnections=10
)

# 从连接池获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()
for row in rows:
    print(row)
cursor.close()
conn.close()  # 将连接放回连接池

在上述代码中，PooledDB类用于创建连接池。creator指定使用的数据库连接库，maxconnections指定连接池的最大连接数。通过pool.connection()方法从连接池获取连接，使用完后关闭连接（实际上是将连接放回连接池）。

连接池的优势

连接池的主要优势在于提高性能和资源利用率。它减少了连接创建和销毁的开销，特别是在高并发的应用场景下，可以显著提升系统的响应速度。同时，连接池还可以对连接进行管理和监控，例如设置连接的超时时间、检测连接的有效性等。

数据库编程中的错误处理

在数据库编程过程中，可能会遇到各种错误，如连接错误、SQL语法错误、数据类型不匹配等。正确处理这些错误对于保证程序的稳定性和可靠性至关重要。

捕获连接错误

以PostgreSQL为例，当连接数据库失败时，会抛出psycopg2.OperationalError异常。以下是捕获连接错误的示例：

import psycopg2

try:
    conn = psycopg2.connect(
        host="localhost",
        database="test",
        user="postgres",
        password="wrong_password"
    )
except psycopg2.OperationalError as e:
    print("连接数据库失败，错误信息:", str(e))

在上述代码中，由于密码错误，会抛出OperationalError异常，通过try - except块捕获并打印错误信息。

捕获SQL执行错误

当执行SQL语句出现错误时，不同的数据库库会抛出相应的异常。以SQLite为例，如果SQL语句语法错误，会抛出sqlite3.OperationalError异常：

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()
try:
    cursor.execute('SELECT * FROM non_existent_table')
except sqlite3.OperationalError as e:
    print("执行SQL语句出错，错误信息:", str(e))
conn.close()

在这个示例中，由于查询一个不存在的表，会抛出OperationalError异常，通过try - except块捕获并处理。

数据库元数据操作

数据库元数据是关于数据库结构和对象的信息，例如表名、列名、数据类型等。在Python中，可以通过数据库连接和游标对象获取这些元数据。

获取表信息

以MySQL为例，获取数据库中所有表的名称：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

cursor.execute("SHOW TABLES")
tables = cursor.fetchall()
for table in tables:
    print(table[0])
conn.close()

上述代码中，通过执行SHOW TABLES语句获取所有表的信息，fetchall()方法返回的结果集中每个元素是一个元组，元组的第一个元素就是表名。

获取列信息

获取表中列的信息，同样以MySQL为例：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

cursor.execute("DESCRIBE users")
columns = cursor.fetchall()
for column in columns:
    print(column[0], column[1])  # 列名和数据类型
conn.close()

这里执行DESCRIBE users语句获取users表的列信息，fetchall()返回的结果集中每个元组包含列名、数据类型等信息。

通过对数据库元数据的操作，可以在程序运行时动态获取数据库结构信息，从而实现一些自动化的数据库管理和操作功能。

数据库编程的优化技巧

在进行数据库编程时，优化是提高系统性能的关键。以下介绍一些常见的优化技巧。

合理设计数据库结构

数据库结构设计的合理性直接影响到查询和数据操作的效率。例如，避免数据冗余，通过规范化设计减少重复数据，以提高存储效率和数据一致性。同时，根据应用的查询需求，合理建立索引。比如，如果经常根据name字段查询users表，那么可以在name字段上建立索引：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

cursor.execute('CREATE INDEX idx_name ON users (name)')
conn.commit()
conn.close()

索引可以加快查询速度，但过多的索引也会增加插入、更新和删除操作的开销，所以要根据实际情况权衡。

批量操作数据

在进行插入、更新或删除操作时，如果数据量较大，采用批量操作可以减少与数据库的交互次数，提高效率。以MySQL的插入操作为例：

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='test_db'
)
cursor = conn.cursor()

data = [
    ("Tom", 22),
    ("Jerry", 20),
    ("Mike", 25)
]
sql = "INSERT INTO users (name, age) VALUES (%s, %s)"
cursor.executemany(sql, data)
conn.commit()
print(cursor.rowcount, "条记录插入成功。")
conn.close()

这里使用executemany()方法一次性插入多条数据，相比逐条插入大大减少了数据库交互次数。

优化查询语句

编写高效的查询语句是数据库优化的重要环节。例如，避免使用全表扫描，尽量使用索引覆盖查询。同时，合理使用JOIN操作，确保JOIN条件正确且高效。对于复杂的查询，可以使用查询分析工具（如MySQL的EXPLAIN关键字）来分析查询执行计划，找出性能瓶颈并进行优化。

通过综合运用这些优化技巧，可以显著提升Python数据库编程的性能和效率，使应用程序在处理大量数据时更加稳定和高效。