Python 数据库访问是指通过 Python 代码连接并操作数据库（如 MySQL、PostgreSQL、SQLite、Oracle 等），核心步骤为：连接数据库 → 执行 SQL 语句 → 处理结果 → 关闭连接。Python 提供了统一的数据库操作接口标准（DB-API），各数据库厂商或社区提供了对应的实现库。以下是详细介绍：

一、核心概念（DB-API 标准）

Python 数据库操作遵循 PEP 249 定义的 DB-API 标准，确保不同数据库的操作接口一致，主要核心对象和方法如下：

对象/方法	说明
connect()	建立数据库连接，返回连接对象（Connection）
Connection.cursor()	创建游标对象（Cursor），用于执行 SQL 语句
Cursor.execute(sql)	执行单条 SQL 语句（返回受影响的行数，或 None）
Cursor.executemany(sql, params)	批量执行 SQL 语句（如批量插入）
Cursor.fetchone()	获取查询结果的第一条记录（返回元组或 None）
Cursor.fetchmany(size)	获取查询结果的前 size 条记录（返回列表，元素为元组）
Cursor.fetchall()	获取查询结果的所有记录（返回列表，元素为元组）
Connection.commit()	提交事务（用于增删改操作，确保数据持久化）
Connection.rollback()	回滚事务（发生错误时撤销已执行的操作）
Connection.close()	关闭数据库连接
Cursor.close()	关闭游标

注意：所有数据库操作都应通过 游标对象（Cursor） 执行，连接对象（Connection）仅用于管理连接和事务。

二、常用数据库连接示例

以下是 Python 连接主流数据库的具体实现（需先安装对应驱动库）：

1. SQLite（内置数据库，无需额外安装）

SQLite 是轻量级嵌入式数据库，数据存储在单个文件中，Python 内置 sqlite3 库支持直接访问，无需额外安装驱动。

示例：SQLite 操作

import sqlite3

# 1. 连接数据库（若文件不存在则自动创建）
conn = sqlite3.connect("test.db")  # 数据库文件：test.db（当前目录）

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

# 3. 执行 SQL 语句（创建表）
create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    age INTEGER DEFAULT 0,
    email TEXT UNIQUE
)
"""
cursor.execute(create_table_sql)

# 4. 增删改操作（需 commit 提交）
# 插入单条数据
cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", ("Alice", 25, "alice@example.com"))
# 批量插入数据
users = [("Bob", 30, "bob@example.com"), ("Charlie", 35, "charlie@example.com")]
cursor.executemany("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", users)
# 提交事务（关键！否则数据不生效）
conn.commit()

# 5. 查询操作（无需 commit）
cursor.execute("SELECT * FROM users WHERE age > ?", (28,))
# 获取所有结果
results = cursor.fetchall()
print("查询结果：")
for row in results:
    print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[2]}, 邮箱: {row[3]}")

# 6. 关闭游标和连接（推荐用 try...finally 确保关闭）
cursor.close()
conn.close()

特点：无需安装数据库服务，适合小型应用、测试或嵌入式场景。

2. MySQL（主流关系型数据库）

MySQL 是常用开源关系型数据库，需安装第三方驱动库 mysql-connector-python 或 pymysql。

步骤 1：安装驱动

# 方式 1：安装 mysql-connector-python（官方推荐）
pip install mysql-connector-python

# 方式 2：安装 pymysql（社区维护，兼容性好）
pip install pymysql

示例：MySQL 操作（使用 mysql-connector-python）

import mysql.connector
from mysql.connector import errorcode

try:
    # 1. 连接数据库（需替换为实际配置）
    conn = mysql.connector.connect(
        host="localhost",       # 数据库主机地址
        port=3306,              # 端口（默认 3306）
        user="root",            # 用户名
        password="your_password",  # 密码
        database="test_db"      # 要连接的数据库（需提前创建）
    )

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

    # 3. 执行 SQL（创建表、插入数据等）
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(50) NOT NULL,
        age INT DEFAULT 0,
        email VARCHAR(100) UNIQUE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """
    cursor.execute(create_table_sql)

    # 插入数据（参数化查询，避免 SQL 注入）
    cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ("Alice", 25, "alice@example.com"))
    conn.commit()  # 提交事务

    # 查询数据
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
    print("MySQL 查询结果：")
    for row in results:
        print(row)

except mysql.connector.Error as err:
    if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        print("用户名或密码错误")
    elif err.errno == errorcode.ER_BAD_DB_ERROR:
        print("数据库不存在")
    else:
        print(f"错误：{err}")
finally:
    # 确保关闭连接
    if 'conn' in locals() and conn.is_connected():
        cursor.close()
        conn.close()
        print("数据库连接已关闭")

注意：使用 %s 作为占位符（而非 SQLite 的 ?），不同数据库占位符可能不同。

3. PostgreSQL（开源企业级数据库）

PostgreSQL 是功能强大的开源数据库，支持复杂查询、JSON 等，需安装驱动库 psycopg2-binary。

步骤 1：安装驱动

pip install psycopg2-binary

示例：PostgreSQL 操作

import psycopg2
from psycopg2 import OperationalError

try:
    # 1. 连接数据库（替换为实际配置）
    conn = psycopg2.connect(
        host="localhost",
        port=5432,          # 默认端口 5432
        user="postgres",    # 默认用户名
        password="your_password",
        database="test_db"
    )

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

    # 3. 执行 SQL
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id SERIAL PRIMARY KEY,
        name VARCHAR(50) NOT NULL,
        age INTEGER DEFAULT 0,
        email VARCHAR(100) UNIQUE
    )
    """
    cursor.execute(create_table_sql)

    # 插入数据（占位符用 %s）
    cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ("Bob", 30, "bob@example.com"))
    conn.commit()

    # 查询数据
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
    print("PostgreSQL 查询结果：")
    for row in results:
        print(row)

except OperationalError as e:
    print(f"数据库连接失败：{e}")
finally:
    # 关闭连接
    if conn:
        cursor.close()
        conn.close()
        print("连接已关闭")

4. Oracle（商业级数据库）

Oracle 是大型商业数据库，需安装官方驱动 cx_Oracle。

步骤 1：安装驱动

pip install cx_Oracle

示例：Oracle 操作

import cx_Oracle

try:
    # 1. 连接数据库（替换为实际配置）
    dsn = cx_Oracle.makedsn("localhost", 1521, service_name="ORCLCDB")  # 服务名需根据实际配置
    conn = cx_Oracle.connect(user="sys", password="your_password", dsn=dsn, mode=cx_Oracle.SYSDBA)

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

    # 3. 执行 SQL
    create_table_sql = """
    CREATE TABLE users (
        id NUMBER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
        name VARCHAR2(50) NOT NULL,
        age NUMBER DEFAULT 0,
        email VARCHAR2(100) UNIQUE
    )
    """
    cursor.execute(create_table_sql)
    conn.commit()

    # 插入数据（占位符用 :1, :2...）
    cursor.execute("INSERT INTO users (name, age, email) VALUES (:1, :2, :3)", ("Charlie", 35, "charlie@example.com"))
    conn.commit()

    # 查询数据
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
    print("Oracle 查询结果：")
    for row in results:
        print(row)

except cx_Oracle.Error as e:
    print(f"Oracle 错误：{e}")
finally:
    if conn:
        cursor.close()
        conn.close()
        print("连接已关闭")

注意：Oracle 占位符使用 :n（如 :1、:2），与其他数据库不同。

5. SQL Server（企业级关系型数据库）

SQL Server 是 Microsoft 开发的企业级关系型数据库，在Windows环境和企业应用中应用广泛，Python连接需使用pyodbc库并配合ODBC驱动。

步骤 1：安装驱动和库

需同时安装Python库和SQL Server ODBC驱动，ODBC驱动需从微软官网下载对应操作系统版本。

# 安装Python库
pip install pyodbc

示例：SQL Server 操作

import pyodbc

try:
    # 1. 定义连接字符串（两种认证方式二选一）
    # 方式1：Windows身份验证（Trusted_Connection=yes）
    conn_str = (
        r'DRIVER={ODBC Driver 17 for SQL Server};'  # 驱动版本需与安装一致
        r'SERVER=localhost\SQLEXPRESS;'            # 服务器地址+实例名
        r'DATABASE=test_db;'                       # 目标数据库
        r'Trusted_Connection=yes;'
    )
    # 方式2：SQL Server身份验证（用户名+密码）
    # conn_str = (
    #     r'DRIVER={ODBC Driver 17 for SQL Server};'
    #     r'SERVER=localhost\SQLEXPRESS;'
    #     r'DATABASE=test_db;'
    #     r'UID=sa;'
    #     r'PWD=your_password;'
    # )

    # 2. 建立连接
    conn = pyodbc.connect(conn_str)
    # 3. 创建游标
    cursor = conn.cursor()

    # 4. 执行SQL（创建表）
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id INT IDENTITY(1,1) PRIMARY KEY,
        name NVARCHAR(50) NOT NULL,
        age INT DEFAULT 0,
        email NVARCHAR(100) UNIQUE
    )
    """
    cursor.execute(create_table_sql)

    # 5. 插入数据（占位符用?）
    # 单条插入
    cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", ("David", 40, "david@example.com"))
    # 批量插入
    users = [("Ella", 32, "ella@example.com"), ("Frank", 28, "frank@example.com")]
    cursor.executemany("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", users)
    conn.commit()  # 提交事务

    # 6. 查询数据
    cursor.execute("SELECT * FROM users WHERE age < ?", (35,))
    results = cursor.fetchall()
    print("SQL Server 查询结果：")
    for row in results:
        print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[2]}, 邮箱: {row[3]}")

except pyodbc.Error as ex:
    sqlstate = ex.args[0]
    print(f"SQL Server 操作错误：{ex}")
    conn.rollback()  # 出错回滚
finally:
    # 关闭资源
    if 'conn' in locals() and conn:
        cursor.close()
        conn.close()
        print("SQL Server 连接已关闭")

注意：SQL Server 占位符使用?，服务器地址格式为“主机名\实例名”（如localhost\SQLEXPRESS），创建自增字段需用IDENTITY(1,1)语法。

三、关键注意事项

1. 防止 SQL 注入

绝对禁止直接拼接 SQL 语句（如 f"INSERT INTO users (name) VALUES ('{name}')"），恶意用户可能通过注入攻击窃取或篡改数据。

正确做法：使用 参数化查询（通过占位符传递参数），所有数据库驱动都会自动处理参数转义：

# 错误示例（易受 SQL 注入）
name = "'); DROP TABLE users; --"  # 恶意输入
cursor.execute(f"INSERT INTO users (name) VALUES ('{name}')")  # 危险！

# 正确示例（参数化查询）
cursor.execute("INSERT INTO users (name) VALUES (%s)", (name,))  # MySQL/PostgreSQL
# cursor.execute("INSERT INTO users (name) VALUES (?)", (name,))  # SQLite
# cursor.execute("INSERT INTO users (name) VALUES (:1)", (name,))  # Oracle

2. 事务管理

• 增删改操作（INSERT/UPDATE/DELETE）必须通过 conn.commit() 提交，否则数据仅在当前连接有效，关闭连接后丢失。

• 发生错误时，使用 conn.rollback() 回滚事务，撤销已执行的操作：

  try:
      cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", ("Dave", "dave@example.com"))
      # 模拟错误
      1 / 0
      conn.commit()  # 无错误则提交
  except Exception as e:
      conn.rollback()  # 出错则回滚
      print(f"操作失败，已回滚：{e}")

3. 连接池（高并发场景）

频繁创建和关闭数据库连接会消耗大量资源，高并发场景下需使用 连接池 管理连接（如 DBUtils.PooledDB）。

示例：使用 DBUtils.PooledDB 实现 MySQL 连接池

# 安装依赖
pip install DBUtils

from DBUtils.PooledDB import PooledDB
import mysql.connector

# 创建连接池
pool = PooledDB(
    creator=mysql.connector,  # 数据库驱动
    maxconnections=10,        # 最大连接数
    mincached=2,             # 最小空闲连接数
    maxcached=5,             # 最大空闲连接数
    host="localhost",
    user="root",
    password="your_password",
    database="test_db"
)

# 从连接池获取连接
conn = pool.connection()
cursor = conn.cursor()

# 执行操作（与普通连接一致）
cursor.execute("SELECT * FROM users")
print(cursor.fetchall())

# 关闭连接（实际是放回连接池，而非真正关闭）
cursor.close()
conn.close()

4. 数据类型转换

数据库返回的字段类型需与 Python 类型对应（如 MySQL 的 DATETIME 对应 Python 的 datetime.datetime），复杂类型（如 JSON、BLOB）需手动处理：

# 读取 JSON 类型字段（MySQL 5.7+ 支持）
cursor.execute("SELECT json_field FROM table WHERE id = %s", (1,))
row = cursor.fetchone()
import json
json_data = json.loads(row[0])  # 字符串转 JSON

四、ORM 框架（简化数据库操作）

直接写 SQL 语句繁琐且易出错，实际开发中常用 ORM（对象关系映射）框架 将数据库表映射为 Python 类，通过操作类和对象实现数据库操作。

主流 ORM 框架：

1. SQLAlchemy：功能强大，支持多种数据库，适合中大型项目。

2. Django ORM：Django 框架内置，简洁易用，适合 Django 项目。

3. Peewee：轻量级 ORM，语法直观，适合小型项目。

示例：SQLAlchemy 操作 MySQL

# 安装依赖
pip install sqlalchemy

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 1. 创建数据库引擎（连接字符串格式：数据库+驱动://用户名:密码@主机:端口/数据库名）
engine = create_engine("mysql+mysqlconnector://root:your_password@localhost:3306/test_db")

# 2. 定义基类
Base = declarative_base()

# 3. 定义模型（映射数据库表）
class User(Base):
    __tablename__ = "users"  # 表名
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer, default=0)
    email = Column(String(100), unique=True)

# 4. 创建表（若不存在）
Base.metadata.create_all(engine)

# 5. 创建会话（类似游标）
Session = sessionmaker(bind=engine)
session = Session()

# 6. 操作数据
# 插入数据
user1 = User(name="Alice", age=25, email="alice@example.com")
session.add(user1)
session.commit()

# 查询数据
users = session.query(User).filter(User.age > 20).all()
for user in users:
    print(f"ID: {user.id}, 姓名: {user.name}")

# 更新数据
user1.age = 26
session.commit()

# 删除数据
session.delete(user1)
session.commit()

# 关闭会话
session.close()

ORM 优势：无需编写 SQL 语句，语法更符合 Python 习惯，自动处理数据类型转换和 SQL 注入防护。

五、总结

Python 数据库访问的核心是遵循 DB-API 标准，通过驱动库连接数据库，使用游标执行 SQL 语句。实际开发中需注意：

• 用参数化查询防止 SQL 注入；

• 增删改操作需提交事务；

• 高并发场景使用连接池；

• 复杂项目优先使用 ORM 框架简化开发。

根据项目需求选择合适的数据库（如小型项目用 SQLite，企业级项目用 MySQL/PostgreSQL/Oracle），并搭配对应的驱动和工具。