一、Python 与数据库的连接基础

1. 标准接口：DB-API 2.0

Python 数据库连接遵循 DB-API 2.0 标准：

import sqlite3  # 内置模块
import mysql.connector  # MySQL 驱动
import psycopg2  # PostgreSQL 驱动

2. 基本操作流程

# 1. 连接数据库
conn = sqlite3.connect('example.db')

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

# 3. 执行 SQL
cursor.execute("CREATE TABLE users (id INTEGER, name TEXT)")

# 4. 提交事务
conn.commit()

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

3. 推荐使用上下文管理器

# 更安全的方式
with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
# 自动关闭连接和游标

---

二、SQLite：轻量级数据库

1. 特点

• 零配置：无需安装服务器
• 单文件：整个数据库就是一个文件
• 适合场景：小型应用、测试、移动端、嵌入式
• 限制：不支持高并发写入

2. 安装与连接

import sqlite3

# 内存数据库（不保存）
conn = sqlite3.connect(':memory:')

# 文件数据库
conn = sqlite3.connect('myapp.db')

# SQLite 自带驱动，无需额外安装

3. 实战示例

import sqlite3

# 创建数据库表
with sqlite3.connect('users.db') as conn:
    cursor = conn.cursor()
    
    # 创建表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL,
            email TEXT UNIQUE,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    # 插入数据（防 SQL 注入）
    cursor.execute(
        "INSERT INTO users (username, email) VALUES (?, ?)",
        ('zhangsan', 'zhangsan@example.com')
    )
    
    # 批量插入
    users = [
        ('lisi', 'lisi@example.com'),
        ('wangwu', 'wangwu@example.com')
    ]
    cursor.executemany(
        "INSERT INTO users (username, email) VALUES (?, ?)",
        users
    )
    
    conn.commit()
    
    # 查询数据
    cursor.execute("SELECT * FROM users WHERE email LIKE '%@example.com'")
    results = cursor.fetchall()
    
    for row in results:
        print(f"ID: {row[0]}, 用户名：{row[1]}, 邮箱：{row[2]}")

4. 性能优化

# 1. 使用事务批量操作
conn = sqlite3.connect('db.sqlite3')
conn.execute("BEGIN")

try:
    for i in range(1000):
        conn.execute(
            "INSERT INTO logs (action) VALUES (?)",
            (f"action_{i}",)
        )
    conn.commit()
except:
    conn.rollback()
    raise

# 2. 创建索引加速查询
cursor.execute("CREATE INDEX idx_email ON users(email)")

# 3. 使用 WAL 模式（提升并发）
conn.execute("PRAGMA journal_mode=WAL")

---

三、MySQL：最流行的开源数据库

1. 特点

• 成熟稳定：应用最广泛的数据库
• 支持高并发：适合 Web 应用
• 功能丰富：存储过程、触发器、视图
• 生态完善：支持多种编程语言

2. 安装与配置

# 安装 MySQL 驱动
pip install mysql-connector-python
# 或使用 PyMySQL
pip install pymysql

3. 连接示例

import mysql.connector

# 连接配置
config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': 'your_password',
    'database': 'myapp',
    'charset': 'utf8mb4',
    'use_unicode': True
}

# 建立连接
conn = mysql.connector.connect(**config)
cursor = conn.cursor()

# 执行查询
cursor.execute("SELECT * FROM users")
for row in cursor:
    print(row)

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

4. 使用连接池

from mysql.connector import pooling

# 创建连接池
pool = pooling.MySQLConnectionPool(
    pool_name="mypool",
    pool_size=5,
    pool_reset_session=True,
    host='localhost',
    user='root',
    password='password',
    database='myapp'
)

# 获取连接（自动复用）
conn = pool.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")

5. 实战：商品管理系统

import mysql.connector
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    conn = mysql.connector.connect(**config)
    try:
        yield conn
        conn.commit()
    except:
        conn.rollback()
        raise
    finally:
        conn.close()

# 创建商品表
with get_db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(100) NOT NULL,
            price DECIMAL(10, 2) NOT NULL,
            stock INT DEFAULT 0,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')

# 添加商品
with get_db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)",
        ('iPhone 15', 6999.00, 100)
    )

# 查询商品
with get_db_connection() as conn:
    cursor = conn.cursor(dictionary=True)  # 返回字典格式
    cursor.execute("SELECT * FROM products WHERE stock > 50")
    products = cursor.fetchall()
    for p in products:
        print(f"{p['name']}: ¥{p['price']}, 库存：{p['stock']}")

---

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

1. 特点

• 功能强大：支持 JSON、数组、全文搜索
• ACID compliant：事务保证强
• 扩展性强：支持自定义类型、函数
• 适合场景：复杂查询、数据分析、企业级应用

2. 安装与配置

# 安装驱动
pip install psycopg2-binary
# 或 psycopg2（需要系统安装 PostgreSQL 客户端库）

3. 连接示例

import psycopg2
from psycopg2 import sql

# 连接配置
conn = psycopg2.connect(
    host="localhost",
    port=5432,
    database="myapp",
    user="postgres",
    password="your_password"
)

cursor = conn.cursor()

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

# 使用参数化查询
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))

4. PostgreSQL 特色功能

JSON 数据处理

# 创建带 JSON 字段的表
cursor.execute('''
    CREATE TABLE products (
        id SERIAL PRIMARY KEY,
        name VARCHAR(100),
        spec JSONB  -- JSONB 是二进制 JSON，查询更快
    )
''')

# 插入 JSON 数据
cursor.execute('''
    INSERT INTO products (name, spec) VALUES (%s, %s)
''', ('Laptop', '{"cpu": "Intel i7", "ram": "16GB", "storage": "512GB SSD"}'))

# 查询 JSON 字段
cursor.execute('''
    SELECT name, spec->>'cpu' as cpu
    FROM products
''')

全文搜索

# 创建全文索引
cursor.execute('''
    CREATE INDEX idx_products_search ON products
    USING GIN(to_tsvector('english', name))
''')

# 全文搜索查询
cursor.execute('''
    SELECT name FROM products
    WHERE to_tsvector('english', name) @@ to_tsquery('english', 'laptop & fast')
''')

数组类型

# 创建带数组的表
cursor.execute('''
    CREATE TABLE courses (
        id SERIAL PRIMARY KEY,
        name VARCHAR(100),
        tags TEXT[]  -- 文本数组
    )
''')

# 插入数组
cursor.execute(
    "INSERT INTO courses (name, tags) VALUES (%s, %s)",
    ('Python 入门', ['编程', '入门', '数据科学'])
)

# 查询包含特定标签的课程
cursor.execute('''
    SELECT name FROM courses
    WHERE '编程' = ANY(tags)
''')

---

五、三种数据库对比

特性	SQLite	MySQL	PostgreSQL
安装配置	零配置	需安装服务	需安装服务
文件大小	单文件	多个文件	多个文件
并发性能	低（写冲突）	高	很高
SQL 标准	部分支持	完整支持	完整支持
事务支持	支持	支持	完善支持
JSON 支持	弱	5.7+ 支持	强（JSONB）
扩展能力	无	中等	极强
适用场景	小型应用	Web 应用	企业/复杂查询
学习曲线	简单	中等	较陡

选型建议：

• SQLite：个人项目、测试、移动端、原型开发
• MySQL：Web 应用、博客、电商平台
• PostgreSQL：数据分析、复杂查询、金融系统

---

六、最佳实践

1. 连接管理

# 使用连接池
from sqlalchemy import create_engine

# SQLAlchemy  ORM（推荐）
engine = create_engine(
    'mysql+pymysql://user:pass@localhost/myapp',
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600
)

# 使用会话
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

2. SQL 注入防护

# ❌ 错误方式：字符串拼接
query = "SELECT * FROM users WHERE id = " + user_id

# ✅ 正确方式：参数化查询
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

3. 错误处理

import sqlite3

try:
    conn = sqlite3.connect('db.sqlite3')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM nonexistent_table")
except sqlite3.OperationalError as e:
    print(f"数据库错误：{e}")
except Exception as e:
    print(f"其他错误：{e}")
finally:
    if conn:
        conn.close()

4. 性能优化

# 1. 使用索引
cursor.execute("CREATE INDEX idx_name ON users(name)")

# 2. 批量操作
cursor.executemany(
    "INSERT INTO logs (action) VALUES (?)",
    [(f'action_{i}',) for i in range(1000)]
)

# 3. 避免全表扫描
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
# 而不是
cursor.execute("SELECT * FROM users WHERE name = '张三'")

---

七、从实践中学

小项目的数据库选择

几年前帮邻居老李做社区团购小程序，初期用 SQLite，用户多了就迁移到 MySQL。过程很简单：

# 迁移脚本示例
import sqlite3
import mysql.connector

# 1. 从 SQLite 导出
sqlite_conn = sqlite3.connect('old.db')
sqlite_cursor = sqlite_conn.cursor()
sqlite_cursor.execute("SELECT * FROM orders")
data = sqlite_cursor.fetchall()

# 2. 写入 MySQL
mysql_conn = mysql.connector.connect(**mysql_config)
mysql_cursor = mysql_conn.cursor()

for row in data:
    mysql_cursor.execute(
        "INSERT INTO orders VALUES (%s, %s, %s, %s)",
        row
    )

mysql_conn.commit()

遇到过的坑

1. 字符编码问题：统一用 UTF-8
2. 时区问题：数据库和服务器时区要一致
3. 连接泄漏：一定要关闭连接
4. 事务未提交：记得 conn.commit()