数据库操作是绝大多数 Python 应用的核心环节。无论是 Web 后端、数据分析管道，还是自动化脚本，都离不开与数据库的交互。

Python 通过 DB-API 2.0（PEP 249） 规范，为开发者提供了统一的数据库操作接口。本文将系统讲解 Python 连接数据库的全部知识点，涵盖 SQLite、MySQL、PostgreSQL 三大主流关系型数据库，以及 SQLAlchemy ORM、连接池、异步编程、NoSQL 数据库等进阶内容，并提供大量可直接运行的实战案例。

一、核心概念：DB-API 2.0 规范

1.1 什么是 DB-API？

DB-API 是 Python 官方定义的数据库接口规范（PEP 249），它统一了不同数据库驱动的使用方法。无论你使用哪种数据库，操作流程都是一致的：

  1. 建立连接（Connection）—— 连接到数据库服务器
  2. 创建游标（Cursor）—— 用于执行 SQL 语句
  3. 执行 SQL —— 通过游标执行查询或更新
  4. 获取结果 —— 从游标中读取查询结果
  5. 提交/回滚事务 —— 确认或撤销更改
  6. 关闭连接 —— 释放资源

1.2 核心对象

对象 说明
Connection 代表与数据库的连接，负责事务管理和创建游标
Cursor 执行 SQL 语句并获取结果
参数化查询 使用占位符（%s 或 ?）防止 SQL 注入

二、SQLite——开箱即用的嵌入式数据库

SQLite 是 Python 标准库自带的轻量级数据库，无需安装任何额外依赖即可使用。数据存储在单个文件中，适合小型应用、原型开发和测试环境。

2.1 连接与基本操作

import sqlite3

# 连接数据库（文件不存在会自动创建）
conn = sqlite3.connect('example.db')

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

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

# 插入数据
cursor.execute(
    "INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
    ('张三', 25, 'zhangsan@example.com')
)

# 提交事务
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# 关闭连接
conn.close()

2.2 高级连接配置

import sqlite3

conn = sqlite3.connect(
    'shop.db',
    timeout=10,                      # 锁等待时间（秒）
    isolation_level='IMMEDIATE',     # 事务隔离级别
    detect_types=sqlite3.PARSE_DECLTYPES  # 自动类型转换
)
  • timeout：多线程并发时，等待数据库锁释放的超时时间
  • isolation_level：控制事务行为（DEFERRED/IMMEDIATE/EXCLUSIVE）
  • detect_types：自动将 SQLite 的 TIMESTAMP 等类型转换为 Python 类型

2.3 使用上下文管理器

import sqlite3

# 使用 with 语句自动提交和关闭
with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    rows = cursor.fetchall()
    # 退出 with 块时自动 commit

三、MySQL——最流行的开源关系型数据库

连接 MySQL 需要使用第三方驱动。常用的有 PyMySQL（纯 Python 实现）和 mysql-connector-python（官方驱动）。

3.1 安装与连接

pip install pymysql

import pymysql

# 建立连接
conn = pymysql.connect(
    host='localhost',      # 主机地址
    port=3306,             # 端口，默认 3306
    user='root',           # 用户名
    password='123456',     # 密码
    database='test_db',    # 数据库名
    charset='utf8mb4'      # 字符编码
)

cursor = conn.cursor()

# 执行查询
cursor.execute("SELECT VERSION()")
version = cursor.fetchone()
print(f"MySQL 版本: {version[0]}")

conn.close()

3.2 连接参数详解

参数 说明 默认值
host MySQL 服务器 IP 或主机名 localhost
port 端口号 3306
user 用户名 —
password 密码 —
database 要连接的数据库名 —
charset 连接字符集 utf8

3.3 完整 CRUD 操作

import pymysql

class MySQLManager:
    def __init__(self, host, user, password, database):
        self.conn = pymysql.connect(
            host=host, user=user, password=password,
            database=database, charset='utf8mb4'
        )
        self.cursor = self.conn.cursor()
    
    def create_table(self):
        sql = '''
            CREATE TABLE IF NOT EXISTS products (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                price DECIMAL(10, 2),
                stock INT DEFAULT 0
            )
        '''
        self.cursor.execute(sql)
        self.conn.commit()
        print("表创建成功")
    
    def insert(self, name, price, stock):
        sql = "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"
        self.cursor.execute(sql, (name, price, stock))
        self.conn.commit()
        return self.cursor.lastrowid
    
    def select_all(self):
        self.cursor.execute("SELECT * FROM products")
        return self.cursor.fetchall()  # 获取所有行
    
    def select_one(self, product_id):
        self.cursor.execute("SELECT * FROM products WHERE id = %s", (product_id,))
        return self.cursor.fetchone()  # 获取单行
    
    def update(self, product_id, stock):
        sql = "UPDATE products SET stock = %s WHERE id = %s"
        self.cursor.execute(sql, (stock, product_id))
        self.conn.commit()
        return self.cursor.rowcount  # 受影响的行数
    
    def delete(self, product_id):
        sql = "DELETE FROM products WHERE id = %s"
        self.cursor.execute(sql, (product_id,))
        self.conn.commit()
        return self.cursor.rowcount
    
    def close(self):
        self.cursor.close()
        self.conn.close()

# 使用示例
db = MySQLManager('localhost', 'root', '123456', 'test_db')
db.create_table()
db.insert('苹果', 5.50, 100)
db.insert('香蕉', 3.80, 200)
print(db.select_all())
db.close()

3.4 批量插入（executemany）

# 批量插入多条数据
data = [
    ('西瓜', 2.50, 50),
    ('葡萄', 8.90, 30),
    ('橙子', 4.20, 80)
]
cursor.executemany(
    "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)",
    data
)
conn.commit()
print(f"插入了 {cursor.rowcount} 条记录")

四、PostgreSQL——功能最强大的开源数据库

PostgreSQL 的官方 Python 驱动是 psycopg2。它用 C 语言编写，性能优越，严格遵循 DB-API 2.0 规范。

4.1 安装

# 推荐安装二进制版本（无需编译）
pip install psycopg2-binary

# 或标准版（需要编译环境）
pip install psycopg2

4.2 基本连接与操作

import psycopg2

# 建立连接
conn = psycopg2.connect(
    host='localhost',
    port='5432',
    database='postgres',
    user='postgres',
    password='postgres'
)

cursor = conn.cursor()

# 执行查询
cursor.execute("SELECT version()")
print(cursor.fetchone()[0])

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

# 插入数据
cursor.execute(
    "INSERT INTO employees (name, department, salary) VALUES (%s, %s, %s)",
    ('李四', '技术部', 15000.00)
)

# 提交事务
conn.commit()

conn.close()

4.3 事务管理

psycopg2 默认关闭自动提交，需要显式调用 commit() 或 rollback()。

import psycopg2

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

try:
    # 开始事务
    cursor.execute("INSERT INTO accounts VALUES (%s, %s)", (1, 100))
    cursor.execute("UPDATE balances SET amount = amount - %s WHERE id = %s", (100, 2))
    conn.commit()  # 全部成功则提交
    print("事务提交成功")
except Exception as e:
    conn.rollback()  # 出现错误则回滚
    print(f"事务回滚: {e}")
finally:
    conn.close()

4.4 使用上下文管理器

from contextlib import closing

# closing 确保连接被正确关闭
with closing(psycopg2.connect(
    dbname='testdb', user='app_user', password='secure_pwd',
    host='localhost', connect_timeout=10  # 连接超时
)) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM employees")
        for row in cursor.fetchall():
            print(row)

五、SQLAlchemy——Python 最强大的 ORM 框架

SQLAlchemy 是 Python 生态中最流行的 ORM（对象关系映射） 工具。它提供两层架构：

  • Core：SQL 表达式语言，提供底层 SQL 构建能力
  • ORM：将数据库表映射为 Python 类，实现对象化操作

5.1 安装

pip install sqlalchemy
# 根据数据库安装对应驱动
pip install pymysql        # MySQL
pip install psycopg2-binary # PostgreSQL
# SQLite 使用标准库 sqlite3，无需额外安装

5.2 创建引擎（Engine）

Engine 是 SQLAlchemy 的入口，管理连接池和数据库方言。

from sqlalchemy import create_engine

# SQLite（内存数据库）
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)

# SQLite（文件数据库）
engine = create_engine("sqlite:///example.db")

# MySQL
engine = create_engine(
    "mysql+pymysql://root:123456@localhost:3306/test_db",
    echo=True
)

# PostgreSQL
engine = create_engine(
    "postgresql+psycopg2://postgres:postgres@localhost:5432/postgres"
)

URL 格式：dialect+driver://username:password@host:port/database

5.3 Core 方式——使用 SQL 表达式

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData

engine = create_engine("sqlite:///example.db")

# 定义表结构
metadata = MetaData()
users = Table(
    'users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('email', String(100))
)
metadata.create_all(engine)

# 插入数据
with engine.connect() as conn:
    # 使用 SQL 表达式
    conn.execute(
        users.insert().values(name='张三', email='zhangsan@test.com')
    )
    conn.commit()

    # 查询数据
    result = conn.execute(users.select())
    for row in result:
        print(row)

5.4 ORM 方式——定义模型类

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

# 创建基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

# 创建引擎和表
engine = create_engine("sqlite:///example.db")
Base.metadata.create_all(engine)

# 创建 Session
Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
user = User(name='李四', email='lisi@test.com')
session.add(user)
session.commit()

# 查询数据
users = session.query(User).all()
for u in users:
    print(u)

# 条件查询
user = session.query(User).filter_by(name='李四').first()
print(user.email)

# 更新
user.email = 'newemail@test.com'
session.commit()

# 删除
session.delete(user)
session.commit()

session.close()

六、连接池——高并发场景的必备组件

在高并发应用中，频繁创建和销毁数据库连接开销巨大。连接池通过复用连接来提升性能。

6.1 使用 DBUtils

DBUtils 是 Python 常用的连接池库。

pip install DBUtils==3.1.0
pip install pymysql
PooledDB —— 适用于多线程环境
from dbutils.pooled_db import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    pymysql,                # 数据库驱动
    maxcached=20,           # 池中闲置连接的最大数量
    maxconnections=100,     # 允许的最大连接数量
    maxusage=20,            # 单个连接的最大重用次数
    blocking=True,          # 超过最大连接数时是否阻塞等待
    host='localhost',
    port=3306,
    user='root',
    password='123456',
    database='test_db',
    charset='utf8mb4'
)

# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM products")
print(cursor.fetchall())

# 归还连接到池（调用 close 实际是归还而非真正关闭）
cursor.close()
conn.close()

6.2 连接池参数详解

参数 说明
mincached 启动时预先创建的闲置连接数
maxcached 池中最大闲置连接数
maxconnections 允许的最大连接总数
maxusage 单个连接最大复用次数
blocking 连接数满时的行为（True=等待，False=抛异常）

6.3 SQLAlchemy 内置连接池

SQLAlchemy 的 Engine 默认带有连接池功能，无需额外配置：

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://root:123456@localhost/test_db",
    pool_size=10,          # 连接池大小
    max_overflow=20,       # 超出 pool_size 后允许的额外连接数
    pool_recycle=3600      # 连接回收时间（秒）
)

七、NoSQL 数据库

7.1 MongoDB——文档数据库

使用 PyMongo 驱动连接 MongoDB。

pip install pymongo

from pymongo import MongoClient

# 连接 MongoDB
client = MongoClient('localhost', 27017)  # 默认端口 27017

# 选择数据库和集合
db = client['test_db']
collection = db['users']

# 插入文档
user = {'name': '张三', 'age': 25, 'email': 'zhangsan@test.com'}
result = collection.insert_one(user)
print(f"插入 ID: {result.inserted_id}")

# 批量插入
users = [
    {'name': '李四', 'age': 30},
    {'name': '王五', 'age': 28}
]
collection.insert_many(users)

# 查询
for doc in collection.find({'age': {'$gt': 25}}):
    print(doc)

# 更新
collection.update_one(
    {'name': '张三'},
    {'$set': {'age': 26}}
)

# 删除
collection.delete_one({'name': '王五'})

client.close()

7.2 Redis——键值缓存数据库

使用 redis-py 驱动连接 Redis。

pip install redis

import redis

# 连接 Redis
r = redis.Redis(
    host='localhost',
    port=6379,        # 默认端口 6379
    db=0,
    decode_responses=True  # 自动解码为字符串
)

# 字符串操作
r.set('username', '张三')
print(r.get('username'))  # 张三

# 哈希操作
r.hset('user:1', 'name', '李四')
r.hset('user:1', 'age', 30)
print(r.hgetall('user:1'))  # {'name': '李四', 'age': '30'}

# 列表操作
r.lpush('logs', 'log1', 'log2')
print(r.lrange('logs', 0, -1))  # ['log2', 'log1']

# 设置过期时间（秒）
r.setex('session_token', 3600, 'abc123')

# 连接池
pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
r = redis.Redis(connection_pool=pool)

八、异步数据库操作

对于高并发 I/O 密集型应用，异步数据库驱动能显著提升性能。

8.1 aiomysql——异步 MySQL

pip install aiomysql

import asyncio
import aiomysql

async def main():
    # 建立异步连接
    conn = await aiomysql.connect(
        host='localhost', port=3306,
        user='root', password='123456',
        db='test_db'
    )
    
    cursor = await conn.cursor()
    
    # 执行查询（使用 await）
    await cursor.execute("SELECT * FROM products")
    rows = await cursor.fetchall()
    for row in rows:
        print(row)
    
    await cursor.close()
    conn.close()

asyncio.run(main())

8.2 asyncpg——异步 PostgreSQL

pip install asyncpg

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(
        host='localhost', port=5432,
        user='postgres', password='postgres',
        database='test'
    )
    
    rows = await conn.fetch("SELECT * FROM employees")
    for row in rows:
        print(row)
    
    await conn.close()

asyncio.run(main())

九、安全最佳实践

9.1 使用环境变量存储敏感信息

绝不将数据库密码硬编码在代码中。

import os
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件

DB_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'user': os.getenv('DB_USER'),
    'password': os.getenv('DB_PASSWORD'),
    'database': os.getenv('DB_NAME')
}

.env 文件示例：

DB_HOST=localhost
DB_USER=app_user
DB_PASSWORD=secure_password_123
DB_NAME=production_db

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

# ❌ 危险：字符串拼接
cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# ✅ 安全：参数化查询
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

9.3 最小权限原则

为应用程序创建专用的数据库用户，只授予必要的最小权限（如只读账号、只写账号），避免使用 root 账号。

9.4 使用 SSL/TLS 加密连接

import pymysql

conn = pymysql.connect(
    host='your-database.amazonaws.com',
    user='user',
    password='password',
    database='db',
    ssl={'ca': '/path/to/ca-cert.pem'}  # SSL 证书
)

十、综合实战案例

10.1 案例一：用户管理系统（SQLite + 完整 CRUD）

import sqlite3
from contextlib import contextmanager

class UserManager:
    def __init__(self, db_path='users.db'):
        self.db_path = db_path
        self._init_db()
    
    @contextmanager
    def _get_connection(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row  # 返回字典风格的行
        try:
            yield conn
        finally:
            conn.close()
    
    def _init_db(self):
        with self._get_connection() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT UNIQUE NOT NULL,
                    email TEXT UNIQUE NOT NULL,
                    age INTEGER CHECK(age >= 0 AND age <= 150),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')
            conn.commit()
    
    def create_user(self, username, email, age):
        with self._get_connection() as conn:
            cursor = conn.execute(
                "INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
                (username, email, age)
            )
            conn.commit()
            return cursor.lastrowid
    
    def get_user(self, user_id):
        with self._get_connection() as conn:
            cursor = conn.execute(
                "SELECT * FROM users WHERE id = ?", (user_id,)
            )
            return cursor.fetchone()
    
    def get_all_users(self):
        with self._get_connection() as conn:
            cursor = conn.execute("SELECT * FROM users ORDER BY id")
            return cursor.fetchall()
    
    def update_user(self, user_id, **kwargs):
        allowed_fields = {'username', 'email', 'age'}
        updates = {k: v for k, v in kwargs.items() if k in allowed_fields}
        if not updates:
            return 0
        
        set_clause = ', '.join(f"{k} = ?" for k in updates.keys())
        values = list(updates.values()) + [user_id]
        
        with self._get_connection() as conn:
            cursor = conn.execute(
                f"UPDATE users SET {set_clause} WHERE id = ?",
                values
            )
            conn.commit()
            return cursor.rowcount
    
    def delete_user(self, user_id):
        with self._get_connection() as conn:
            cursor = conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
            conn.commit()
            return cursor.rowcount

# 使用示例
um = UserManager()
um.create_user('alice', 'alice@test.com', 28)
um.create_user('bob', 'bob@test.com', 35)

users = um.get_all_users()
for user in users:
    print(f"{user['id']}: {user['username']} ({user['email']})")

um.update_user(1, age=29)
um.delete_user(2)

10.2 案例二：数据批量导入（MySQL + pandas）

使用 pandas 的 read_sql 和 to_sql 方法可以高效地进行数据导入导出。

import pandas as pd
from sqlalchemy import create_engine

# 创建引擎
engine = create_engine("mysql+pymysql://root:123456@localhost:3306/test_db")

# 从 CSV 读取数据
df = pd.read_csv('products.csv')

# 批量写入数据库
df.to_sql('products', engine, if_exists='replace', index=False)
print(f"成功导入 {len(df)} 条记录")

# 从数据库读取到 DataFrame
df_from_db = pd.read_sql("SELECT * FROM products", engine)
print(df_from_db.head())

10.3 案例三：Web 应用数据库集成（Flask + SQLAlchemy）

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    stock = db.Column(db.Integer, default=0)

@app.route('/products', methods=['GET'])
def get_products():
    products = Product.query.all()
    return jsonify([{
        'id': p.id, 'name': p.name,
        'price': p.price, 'stock': p.stock
    } for p in products])

@app.route('/products', methods=['POST'])
def create_product():
    data = request.json
    product = Product(
        name=data['name'],
        price=data['price'],
        stock=data.get('stock', 0)
    )
    db.session.add(product)
    db.session.commit()
    return jsonify({'id': product.id}), 201

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

十一、总结与速查表

11.1 核心要点

要点 说明
统一接口 所有 DB-API 兼容驱动使用相同的 connect → cursor → execute 模式
资源管理 使用 with 语句或上下文管理器自动关闭连接
防止注入 始终使用参数化查询，绝不拼接 SQL 字符串
连接池 高并发场景必须使用连接池（DBUtils 或 SQLAlchemy 内置）
配置安全 敏感信息使用环境变量，不硬编码
ORM 选型 复杂项目推荐 SQLAlchemy，简单场景直接用原生驱动

11.2 数据库连接速查表

数据库 驱动 连接示例
SQLite sqlite3（内置） sqlite3.connect('db.db')
MySQL pymysql pymysql.connect(host='localhost', user='root', ...)
PostgreSQL psycopg2 psycopg2.connect(host='localhost', database='db', ...)
MongoDB pymongo MongoClient('localhost', 27017)
Redis redis redis.Redis(host='localhost', port=6379)
ORM SQLAlchemy create_engine('mysql+pymysql://user:pass@host/db')

11.3 游标方法速查

方法 说明
execute(sql, params) 执行单条 SQL
executemany(sql, params_list) 批量执行
fetchone() 获取一行结果
fetchall() 获取所有行
fetchmany(size) 获取指定行数
rowcount 最近操作影响的行数
lastrowid 最后插入行的 ID

通过本文的系统学习，你应该已经掌握了 Python 连接和操作数据库的全部核心技能——从 SQLite 到 MySQL/PostgreSQL，从原生驱动到 SQLAlchemy ORM，从连接池到异步编程，从 NoSQL 到安全实践。这些知识将帮助你在实际项目中构建高效、安全、可维护的数据访问层。

# python # 数据库
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

KMP全栈开发：从Android到AI Agent

avatar AI Agent技术社区

AI Agent可观测性：破解多步推理黑盒的技术实践

初创团队：最小可行可观测性（MVP）清单成熟团队：构建企业级可观测性平台的关键考量开源贡献者：可扩展性与社区生态建设。

avatar AI Agent技术社区
cover

Temu批量视频翻译Python实现方案

avatar AI Agent技术社区