SQLite – Python

下面是一份 完整的 SQLite + Python 中文教程，涵盖从环境搭建到 CRUD 操作、事务、ORM（SQLAlchemy）、连接池等实用技巧，适合初学者和进阶开发者。

---

一、SQLite 简介

SQLite 是一个轻量级、嵌入式、零配置的数据库引擎，适合：

• 桌面应用
• Web 后端（Flask/Django）
• 数据分析
• 本地缓存

它将整个数据库保存在 单个文件 中，无需服务器。Python 从 2.5 起内置 sqlite3 模块，无需额外安装！

---

二、Python 环境要求

• Python 版本：3.6+（推荐 3.11+）
• 内置模块：sqlite3（默认自带）
• 验证安装：

  import sqlite3
  print("SQLite 版本:", sqlite3.sqlite_version)  # 例如: 3.46.1

当前最新 SQLite 版本：3.46.1（2024-08-13）

---

三、Python 连接 SQLite 数据库（基础示例）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import sqlite3
from pathlib import Path

DB_PATH = "users.db"

def main():
    # 连接数据库（自动创建文件）
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    print("成功连接 SQLite 数据库！")

    # 创建表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            age INTEGER NOT NULL
        )
    ''')
    print("表 'users' 已创建或已存在。")

    # 插入数据
    users = [("张三", 28), ("李四", 35), ("王五", 22)]
    cursor.executemany("INSERT INTO users (name, age) VALUES (?, ?)", users)
    conn.commit()
    print(f"已插入 {len(users)} 条数据。")

    # 查询数据
    cursor.execute("SELECT id, name, age FROM users ORDER BY age DESC")
    print("\n--- 查询结果 ---")
    for row in cursor.fetchall():
        print(f"ID: {row[0]} | 姓名: {row[1]} | 年龄: {row[2]}")

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

if __name__ == "__main__":
    main()

运行：python demo.py

---

四、推荐：使用 with 上下文管理器 + Path

from pathlib import Path
import sqlite3

DB_PATH = Path("data/users.db")
DB_PATH.parent.mkdir(exist_ok=True)  # 确保目录存在

with sqlite3.connect(DB_PATH) as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS logs (msg TEXT, ts DATETIME DEFAULT CURRENT_TIMESTAMP)")
    conn.execute("INSERT INTO logs (msg) VALUES (?)", ("系统启动",))
    conn.commit()

    for row in conn.execute("SELECT * FROM logs"):
        print(row)

---

五、封装 CRUD 操作（UserDAO 类）

# user_dao.py
import sqlite3
from typing import List, Optional, Dict
from pathlib import Path

class UserDAO:
    def __init__(self, db_path: str = "users.db"):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(exist_ok=True)
        self._create_table()

    def _connect(self):
        return sqlite3.connect(self.db_path)

    def _create_table(self):
        with self._connect() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL,
                    age INTEGER NOT NULL
                )
            ''')

    # 增
    def insert(self, name: str, age: int) -> int:
        with self._connect() as conn:
            cursor = conn.execute(
                "INSERT INTO users (name, age) VALUES (?, ?)",
                (name, age)
            )
            conn.commit()
            return cursor.lastrowid

    # 删
    def delete(self, user_id: int) -> int:
        with self._connect() as conn:
            cursor = conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
            conn.commit()
            return cursor.rowcount

    # 改
    def update(self, user_id: int, name: str = None, age: int = None) -> int:
        updates = []
        params = []
        if name is not None:
            updates.append("name = ?")
            params.append(name)
        if age is not None:
            updates.append("age = ?")
            params.append(age)
        if not updates:
            return 0

        sql = f"UPDATE users SET {', '.join(updates)} WHERE id = ?"
        params.append(user_id)

        with self._connect() as conn:
            cursor = conn.execute(sql, params)
            conn.commit()
            return cursor.rowcount

    # 查单个
    def find(self, user_id: int) -> Optional[Dict]:
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row  # 返回字典-like 对象
            cursor = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,))
            row = cursor.fetchone()
            return dict(row) if row else None

    # 查全部
    def find_all(self) -> List[Dict]:
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute("SELECT * FROM users ORDER BY id DESC")
            return [dict(row) for row in cursor.fetchall()]

    # 搜索（示例）
    def search_by_age(self, min_age: int) -> List[Dict]:
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute("SELECT * FROM users WHERE age >= ?", (min_age,))
            return [dict(row) for row in cursor.fetchall()]

使用示例：

# demo_dao.py
from user_dao import UserDAO

dao = UserDAO()

# 增
id1 = dao.insert("赵六", 40)
print(f"插入 ID: {id1}")

# 查全部
print("\n所有用户:")
for user in dao.find_all():
    print(f"{user['id']}: {user['name']} ({user['age']}岁)")

# 改
dao.update(id1, age=41)

# 删
dao.delete(id1)

---

六、事务支持（Transaction）

def transfer(db, from_id, to_id, amount):
    with db:
        db.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, from_id))
        db.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, to_id))
    print("转账成功！")

# 使用
with sqlite3.connect("bank.db") as conn:
    conn.execute("CREATE TABLE IF NOT EXISTS accounts (id INTEGER PRIMARY KEY, balance INTEGER)")
    conn.execute("INSERT OR IGNORE INTO accounts VALUES (1, 1000), (2, 0)")
    transfer(conn, 1, 2, 100)

with conn: 自动提交或回滚（异常时回滚）

---

七、使用 SQLAlchemy（ORM 推荐）

pip install sqlalchemy

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

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

    def __repr__(self):
        return f"<User {self.name}, {self.age}岁>"

# 初始化
engine = create_engine('sqlite:///users_orm.db', echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# 使用
session = Session()
session.add(User(name="陈七", age=30))
session.commit()

for user in session.query(User).filter(User.age > 25):
    print(user)

session.close()

---

八、常见问题与最佳实践

问题	解决方案
中文乱码	Python 3 默认 UTF-8，sqlite3 支持良好
数据库路径	使用 Path("data/app.db")，自动处理跨平台
多线程安全	每个线程使用独立 connect()；或使用 check_same_thread=False（谨慎）
高并发优化	启用 WAL 模式：PRAGMA journal_mode=WAL;

conn = sqlite3.connect("app.db", check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA synchronous=NORMAL;")

---

九、推荐工具

• DB Browser for SQLite：图形化管理
• 下载：https://sqlitebrowser.org/
• SQLiteStudio：跨平台 GUI
• VS Code：安装 SQLite 插件（查看 .db 文件）
• Jupyter Notebook：数据分析神器

---

十、完整项目结构示例

project/
├── main.py             ← 主入口
├── user_dao.py         ← DAO 类
├── models.py           ← SQLAlchemy 模型
├── data/
│   └── users.db        ← 自动生成
└── requirements.txt

requirements.txt：

sqlalchemy

---

总结：SQLite + Python 是最简单高效的本地数据库方案，内置、无依赖、支持事务和 ORM。

立即复制代码运行，你将拥有一个完整的 SQLite 数据库系统！

如需 Flask + SQLite、FastAPI + SQLAlchemy、数据备份/加密 或 性能测试，欢迎继续提问！