Python数据库查询案例：从入门到精通的完整数据查询指南
目录导读
Python数据库查询基础概念
问：Python查询数据库需要哪些核心组件？
答：需要三要素——数据库驱动（如pymysql、psycopg2）、连接对象和游标对象，其中游标负责执行SQL并获取结果，连接对象管理会话状态。
现代开发中,使用SQLAlchemy或peewee等ORM框架可以简化操作，但底层仍依赖数据库驱动，理解原生查询有助于掌握性能调优。
核心流程：
建立连接 → 2. 创建游标 → 3. 执行SQL → 4. 获取结果 → 5. 关闭游标/连接
主流数据库连接方式对比
案例：连接MySQL、PostgreSQL和SQLite
| 数据库类型 | 推荐驱动 | 连接字符串示例 |
|---|---|---|
| MySQL | pymysql | host=localhost, user=root, password=xxx |
| PostgreSQL | psycopg2 | dbname=test user=postgres |
| SQLite | sqlite3 | 文件路径：test.db |
连接代码示范：
# MySQL
import pymysql
conn = pymysql.connect(host='localhost', user='root', password='123456', database='shop')
# PostgreSQL
import psycopg2
conn = psycopg2.connect(dbname='shop', user='postgres', password='123456')
# SQLite
import sqlite3
conn = sqlite3.connect('shop.db')
问：如何选择连接方式？
答：小型项目用SQLite，生产环境MySQL/PG，安全要求高时优先使用connection_pool防止连接泄露。
Python数据库查询案例：完整数据查询
案例背景：从电商数据库products表中查询价格大于100元的商品，并按价格降序排列。
Step 1：创建表结构
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100),
price DECIMAL(10,2),
stock INT
)
''')
Step 2：插入测试数据
data = [
('无线耳机', 299, 50),
('机械键盘', 199, 30),
('鼠标垫', 29.9, 100)
]
cursor.executemany('INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)', data)
conn.commit()
Step 3：核心查询（带参数化）
min_price = 100
sql = "SELECT id, name, price, stock FROM products WHERE price > %s ORDER BY price DESC"
cursor.execute(sql, (min_price,))
results = cursor.fetchall()
for row in results:
print(f"商品ID：{row[0]} | 名称：{row[1]} | 价格：{row[2]} | 库存：{row[3]}")
输出示例：
商品ID：1 | 名称：无线耳机 | 价格：299.00 | 库存：50
商品ID：2 | 名称：机械键盘 | 价格：199.00 | 库存：30关键点：
- 使用
%s占位符而非直接拼接字符串，防止SQL注入 fetchall()返回元组列表，适合小数据量- 大数据量时使用
fetchmany(size)或游标迭代器
常见查询场景与代码示例
场景1：单条记录查询
cursor.execute("SELECT * FROM products WHERE id = %s", (1,))
product = cursor.fetchone()
# 返回单元素元组，如(1, '无线耳机', 299.00, 50)
场景2：分页查询
page = 1
page_size = 20
offset = (page - 1) * page_size
cursor.execute("SELECT * FROM products LIMIT %s OFFSET %s", (page_size, offset))
场景3：模糊搜索
keyword = "耳机"
cursor.execute("SELECT * FROM products WHERE name LIKE %s", (f"%{keyword}%",))
场景4：聚合查询（带GROUP BY）
cursor.execute("""
SELECT category, AVG(price) as avg_price
FROM products
GROUP BY category
HAVING AVG(price) > 50
""")
问：如何处理查询结果为空的情况？
答：判断cursor.rowcount是否为0，或检查fetchone()是否为None：
if cursor.rowcount == 0:
print("未找到匹配记录")
性能优化与安全最佳实践
1 查询性能优化策略
- 使用索引：对
WHERE和ORDER BY字段建立索引 - **避免SELECT ***：只取需要的列
- 批量操作：用
executemany()代替逐条插入 - 连接池复用：
DBUtils或SQLAlchemy内置连接池 - 流式查询：超大结果集用
sscursor（MySQL）或named cursor（PG）
代码示例：流式查询
# MySQL
cursor = conn.cursor(pymysql.cursors.SSDictCursor)
cursor.execute('SELECT * FROM large_table')
for row in cursor:
process(row)
2 安全最佳实践
- 永远使用参数化查询，避免字符串拼接
- 最小权限原则：数据库用户只赋予
SELECT权限 - 连接加密：生产环境使用SSL连接
- 超时设置：避免长时间占用连接
- 异常处理：使用
try...finally确保连接释放
try:
conn = pymysql.connect(..., connect_timeout=5)
...
except pymysql.Error as e:
print(f"数据库错误：{e}")
finally:
if 'cursor' in locals():
cursor.close()
if 'conn' in locals() and conn.open:
conn.close()
故障排查与问答
Q1：查询返回的结果是None？
原因：SQL语法错误、表名不存在、条件不匹配
解决：打印SQL语句并在数据库客户端验证
Q2：如何查看执行时间？
import time
start = time.time()
cursor.execute(sql)
print(f"查询耗时：{time.time()-start:.3f}秒")
Q3：Python连接数据库时出现"Authentication failed"？
原因：用户名/密码错误，或MySQL8.0使用caching_sha2_password插件
解决：连接参数添加auth_plugin='mysql_native_password'
Q4：多表联查时如何优化？
# 使用JOIN代替子查询
sql = """
SELECT o.order_id, p.name, o.quantity
FROM orders o
INNER JOIN products p ON o.product_id = p.id
WHERE o.create_date > %s
"""
Q5：事务处理中查询结果不一致？
原因：默认的自动提交模式下查询的是未提交数据
解决：使用事务隔离级别
conn.begin()
cursor.execute("SELECT ... LOCK IN SHARE MODE") # 读锁
conn.commit()
Python数据库查询的核心要点
- 连接生命周期管理：使用
with语句自动释放资源 - 参数化查询：安全第一
- 结果集处理：
fetchone/fetchmany/fetchall按需选择 - 异常与超时处理：防止程序崩溃
- 性能监控：通过
EXPLAIN分析SQL计划
掌握这些案例后,你可以轻松应对90%的Python数据库查询需求，当遇到复杂场景时，记得查阅官方文档或使用print(cursor._last_executed)调试SQL。
