当前位置：首页 > 数据库 > 正文

Python如何查询数据库操作？

在Python中查询数据库通常使用DB-API兼容库（如sqlite3/pymysql）或ORM工具（如SQLAlchemy），基本流程：建立连接→创建游标→执行SQL语句→fetch获取结果→关闭连接，关键方法包括execute()和fetchall()/fetchone()。

Python数据库查询操作详解

在Python中执行数据库查询是开发Web应用、数据分析工具和自动化系统的核心技能，下面详细介绍从连接数据库到安全查询的完整流程,涵盖主流数据库操作。

准备工作：安装驱动与导入模块

Python通过专用驱动连接不同数据库,常用库如下：

数据库类型 安装命令 导入模块
SQLite 无需安装（内置） import sqlite3
MySQL pip install mysql-connector-python import mysql.connector
PostgreSQL pip install psycopg2 import psycopg2

连接数据库（以MySQL为例）

import mysql.connector
# 创建数据库连接
db = mysql.connector.connect(
    host="localhost",
    user="your_username",
    password="your_password",
    database="your_database"
)
# 创建游标对象
cursor = db.cursor()

执行SQL查询操作

基础查询（SELECT）

# 执行查询语句
cursor.execute("SELECT id, name, email FROM users WHERE age > %s", (18,))
# 获取全部结果
results = cursor.fetchall()
for row in results:
    print(f"ID: {row[0]}, Name: {row[1]}, Email: {row[2]}")

参数化查询（防止SQL注入）

# 安全传参方式（使用占位符）
user_id = 5
cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user_id,))

分页查询

page = 2
limit = 10
offset = (page - 1) * limit
cursor.execute("SELECT * FROM products LIMIT %s OFFSET %s", (limit, offset))

处理查询结果

方法 描述 使用场景
fetchone() 获取下一行记录 逐行处理大数据
fetchmany(size=5) 获取指定数量记录 分批次读取
fetchall() 获取所有记录（返回元组列表） 小型数据集
dictcursor 返回字典格式结果 需字段名访问时

字典游标示例：

cursor = db.cursor(dictionary=True)  # MySQL
cursor.execute("SELECT * FROM employees")
for row in cursor:
    print(row["name"], row["salary"])  # 通过字段名访问

事务管理与异常处理

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id=1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id=2")
    db.commit()  # 提交事务
except Exception as e:
    db.rollback()  # 回滚事务
    print(f"操作失败: {e}")
finally:
    cursor.close()
    db.close()  # 必须关闭连接！

高级实践建议

  1. 使用上下文管理器（自动关闭连接）

    Python如何查询数据库操作？ 第1张

    with sqlite3.connect("mydb.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM logs")
    # 自动提交事务并关闭连接

  2. ORM工具（如SQLAlchemy）

    from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///mydb.db")
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users"))
    for row in result:
        print(row)

  3. 异步查询（aiomysql/asyncpg）

    import asyncio
import aiomysql
async def fetch_data():
    conn = await aiomysql.connect(host='localhost', user='root', db='test')
    cursor = await conn.cursor()
    await cursor.execute("SELECT * FROM posts")
    print(await cursor.fetchall())
    conn.close()
asyncio.run(fetch_data())

安全注意事项

  1. 禁止字符串拼接SQL

    # 危险示例（易受SQL注入攻击）
query = f"SELECT * FROM users WHERE name='{user_input}'"
# 正确做法
cursor.execute("SELECT * FROM users WHERE name=%s", (user_input,))

  2. 最小权限原则
    数据库账号仅授予必要权限（如只读权限）。

性能优化技巧

  • 使用EXPLAIN分析SQL执行计划
  • 建立索引加速高频查询字段
  • 批量操作时用executemany() 
    data = [('Alice', 28), ('Bob', 32)]
cursor.executemany("INSERT INTO users (name, age) VALUES (%s, %s)", data)

Python通过标准DB-API接口提供统一的数据库操作方式,核心步骤包括：

  1. 安装驱动 → 建立连接 → 创建游标
  2. 使用参数化SQL执行查询
  3. 通过fetch*()方法处理结果
  4. 事务提交/回滚 → 关闭连接

掌握这些技能可安全高效地操作SQLite、MySQL、PostgreSQL等主流数据库,实际开发中推荐使用ORM框架或异步驱动提升效率。

引用说明参考Python官方DB-API规范、MySQL Connector/Python文档、SQLite3模块文档及OWASP SQL注入防护指南，技术要点已通过Python 3.10+环境验证。

0

相关文章