如何使用 Python 执行 SQL 查询？

一、常用的Python SQL库

在Python中执行SQL查询，最常用的库包括：

1. sqlite3：用于与SQLite数据库交互，适合小型项目或测试环境。
2. psycopg2：用于与PostgreSQL数据库交互，功能强大，支持复杂查询。
3. mysql-connector-python 或 PyMySQL：用于与MySQL数据库交互。
4. SQLAlchemy：一个ORM（对象关系映射）库，支持多种数据库，提供更高层次的抽象。

二、使用sqlite3执行SQL查询

sqlite3是Python内置的库，无需额外安装，适合快速开发和测试。

示例代码：

import sqlite3

# 连接到SQLite数据库（如果数据库不存在，将会自动创建）
conn = sqlite3.connect('example.db')

# 创建一个游标对象
cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        age INTEGER
    )
''')

# 插入数据
cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30))
cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Bob', 25))

# 提交事务
conn.commit()

# 查询数据
cursor.execute('SELECT * FROM users')
rows = cursor.fetchall()

for row in rows:
    print(row)

# 关闭连接
conn.close()

代码说明：

1. 连接数据库：使用sqlite3.connect()连接到SQLite数据库。
2. 创建游标：通过conn.cursor()创建游标对象，用于执行SQL语句。
3. 执行SQL语句：使用cursor.execute()执行创建表、插入数据和查询数据的SQL语句。
4. 提交事务：对于插入、更新等操作，需要调用conn.commit()提交事务。
5. 获取结果：使用cursor.fetchall()获取查询结果。
6. 关闭连接：操作完成后，调用conn.close()关闭数据库连接。

三、使用 psycopg2 执行 PostgreSQL 查询

psycopg2是Python中常用的PostgreSQL适配器，功能强大，支持异步操作和高级特性。

安装 psycopg2：

pip install psycopg2-binary

示例代码：

import psycopg2
from psycopg2 import sql

# 连接到PostgreSQL数据库
conn = psycopg2.connect(
    host='localhost',
    database='testdb',
    user='yourusername',
    password='yourpassword'
)

# 创建游标
cursor = conn.cursor()

# 创建表
create_table_query = '''
CREATE TABLE IF NOT EXISTS employees (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    position VARCHAR(50),
    salary NUMERIC
);
'''
cursor.execute(create_table_query)

# 插入数据
insert_query = sql.SQL("INSERT INTO employees (name, position, salary) VALUES (%s, %s, %s);")
cursor.execute(insert_query, ('Charlie', 'Developer', 70000))
cursor.execute(insert_query, ('Diana', 'Designer', 65000))

# 提交事务
conn.commit()

# 查询数据
cursor.execute('SELECT * FROM employees;')
rows = cursor.fetchall()

for row in rows:
    print(row)

# 关闭连接
cursor.close()
conn.close()

代码说明：

1. 连接数据库：使用psycopg2.connect()连接到PostgreSQL数据库，需提供主机、数据库名、用户名和密码。
2. 创建游标：通过conn.cursor()创建游标对象。
3. 执行SQL语句：使用参数化查询（如%s）防止SQL注入。
4. 提交事务：插入数据后，调用conn.commit()提交事务。
5. 获取结果：使用cursor.fetchall()获取查询结果。
6. 关闭连接：操作完成后，关闭游标和数据库连接。

四、最佳实践与注意事项

1. 使用参数化查询防止SQL注入

   直接拼接SQL字符串容易导致SQL注入攻击。应使用参数化查询或预编译语句。

   # 不安全的做法（可能导致SQL注入）
   user_input = "Alice'; DROP TABLE users;--"
   cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")
   
   # 安全的做法
   cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

2. 上下文管理器（with语句）管理连接和游标

   使用with语句可以确保连接和游标在使用完毕后自动关闭，避免资源泄漏。

   import sqlite3
   
   with sqlite3.connect('example.db') as conn:
       with conn.cursor() as cursor:
           cursor.execute('SELECT * FROM users')
           rows = cursor.fetchall()
           for row in rows:
               print(row)

3. 处理事务

   对于需要保证数据一致性的操作，应显式管理事务，使用commit()和rollback()。

   try:
       with conn.cursor() as cursor:
           cursor.execute("BEGIN;")
           # 执行多个SQL操作
           cursor.execute("INSERT INTO table1 ...")
           cursor.execute("UPDATE table2 ...")
           conn.commit()
   except Exception as e:
       conn.rollback()
       print("事务回滚:", e)

4. 使用ORM（如SQLAlchemy）简化数据库操作

   ORM提供了更高层次的抽象，可以减少手动编写SQL语句的需求，提高代码可维护性。

   安装SQLAlchemy：

   pip install SQLAlchemy

   示例代码：

   from sqlalchemy import create_engine, Column, Integer, String
   from sqlalchemy.orm import declarative_base, sessionmaker
   
   # 创建引擎
   engine = create_engine('sqlite:///example.db')
   
   # 定义基类
   Base = declarative_base()
   
   # 定义模型
   class User(Base):
       __tablename__ = 'users'
       id = Column(Integer, primary_key=True)
       name = Column(String)
       age = Column(Integer)
   
   # 创建表
   Base.metadata.create_all(engine)
   
   # 创建会话
   Session = sessionmaker(bind=engine)
   session = Session()
   
   # 添加数据
   new_user = User(name='Eve', age=28)
   session.add(new_user)
   session.commit()
   
   # 查询数据
   users = session.query(User).all()
   for user in users:
       print(user.id, user.name, user.age)
   
   # 关闭会话
   session.close()

   优点：

   • 提高代码的可读性和可维护性。
   • 自动处理连接和事务管理。
   • 支持复杂的查询和关系映射。

5. 错误处理

   在数据库操作中，应捕获并处理可能的异常，确保程序的健壮性。

   try:
       cursor.execute("SELECT * FROM non_existent_table;")
       rows = cursor.fetchall()
   except sqlite3.OperationalError as e:
       print("表不存在:", e)

6. 性能优化

   • 批量操作：对于大量数据的插入或更新，使用批量操作（如executemany）提高效率。
   • 索引：在频繁查询的字段上创建索引，加快查询速度。
   • 连接池：对于高并发应用，使用连接池管理数据库连接，避免频繁建立和关闭连接的开销。

   批量插入示例：

   data = [('Frank', 35), ('Grace', 29)]
   cursor.executemany('INSERT INTO users (name, age) VALUES (?, ?)', data)
   conn.commit()

五、实际开发中的建议

1. 选择合适的库

   根据项目需求选择合适的数据库驱动或ORM。如果项目较小且使用SQLite，sqlite3足够；如果需要与PostgreSQL或MySQL交互，选择相应的适配器；对于复杂项目，考虑使用SQLAlchemy等ORM。

2. 配置管理

   数据库连接信息（如主机、用户名、密码）应通过配置文件或环境变量管理，避免硬编码在代码中，提升安全性和灵活性。

3. 日志记录

   记录数据库操作的日志，有助于调试和监控。可以使用Python的logging模块记录SQL语句和异常信息。

4. 测试

   编写单元测试和集成测试，确保数据库操作的正确性。使用测试数据库进行测试，避免影响生产数据。

5. 安全性

   除了使用参数化查询防止SQL注入，还应确保数据库用户权限最小化，定期更新数据库驱动和依赖库，防范潜在的安全漏洞。

使用Python执行SQL查询是后端开发中的基本技能。通过选择合适的库、遵循最佳实践、注意安全性和性能优化，可以高效地进行数据库交互。

同时，结合ORM等高级工具，可以进一步提升开发效率和代码质量。在实际开发中，务必重视错误处理、配置管理和安全性，确保系统的稳定和安全运行。