本文档详细说明了将项目从 PostgreSQL 迁移到 SQLite 的完整改动流程和关键注意点。
迁移缘由：
客户暂无网站服务器，首选本地化部署
项目需打包为单一 exe，嵌入 SQLite 便于随包分发
降低额外运维成本（无需独立 DB 服务）
一、改动概述
1.1 主要变化
数据库驱动：从
asyncpg/psycopg2改为aiosqlite连接字符串格式：从
postgresql+asyncpg://...改为sqlite+aiosqlite:///...数据库位置：从远程 PostgreSQL 服务器改为本地 SQLite 文件
时间戳处理：从数据库默认值改为 Python 默认值
并发处理：需要特别注意 SQLite 的锁定问题
1.2 文件结构变化
新增/修改的文件：
├── src/utils/sqlite_util.py # SQLite 工具函数（数据库路径管理）
├── src/conf.py # 配置改为 SQLite
├── src/database.py # 数据库配置改为 SQLite
├── alembic/versions/2025-12-24_*.py # SQLite 时间戳修复迁移
└── src/business_report/router.py # DDL 操作改为使用当前连接
二、详细改动步骤
2.1 依赖项更改 (pyproject.toml)
dependencies = [
"aiosqlite>=0.20.0", # ✅ 新增：SQLite 异步驱动
"asyncpg>=0.29.0", # ⚠️ 保留：测试环境仍可能使用 PostgreSQL
"psycopg2-binary>=2.9.11", # ⚠️ 保留：同步操作可能需要
]
注意点：
aiosqlite是 SQLite 的异步驱动，必须添加建议保留 PostgreSQL 相关依赖，方便后续切换回 PostgreSQL 或用于测试
2.2 配置更改 (src/conf.py)
from src.utils.sqlite_util import get_app_db_path
class Settings(BaseSettings):
# ❌ 旧配置（PostgreSQL）
# DATABASE_URL: str = "postgresql+asyncpg://user:password@host:5432/dbname"
# SYNC_DATABASE_URL: str = "postgresql+psycopg2://user:password@host:5432/dbname"
# ✅ 新配置（SQLite）
_db_path = get_app_db_path("tianlu_spc", "tianlu_database.db")
DATABASE_URL: str = f"sqlite+aiosqlite:///{_db_path.as_posix()}"
SYNC_DATABASE_URL: str = f"sqlite:///{_db_path.as_posix()}"
注意点：
路径格式：SQLite 使用文件路径，Windows 路径需要使用
.as_posix()转换为 POSIX 格式数据库位置：使用
get_app_db_path()将数据库存储在用户 APPDATA 目录，确保：Windows:
%APPDATA%\tianlu_spc\tianlu_database.db便于用户数据管理和备份
避免权限问题
2.3 数据库引擎配置 (src/database.py)
2.3.1 异步引擎配置
connect_args = {}
if settings.DATABASE_URL.startswith("sqlite"):
connect_args = {"check_same_thread": False} # ✅ SQLite 必需
async_engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
future=True,
connect_args=connect_args,
pool_pre_ping=not settings.DATABASE_URL.startswith("sqlite"), # ✅ SQLite 不需要
)
关键注意点：
check_same_thread=False：SQLite 默认不允许跨线程访问，必须设置此参数pool_pre_ping=False：SQLite 不需要连接池预检，应该禁用
2.3.2 同步引擎配置
sync_connect_args = {}
if settings.SYNC_DATABASE_URL.startswith("sqlite"):
sync_connect_args = {"check_same_thread": False} # ✅ SQLite 必需
engine = create_engine(
settings.SYNC_DATABASE_URL,
echo=settings.DEBUG,
connect_args=sync_connect_args,
pool_pre_ping=not settings.SYNC_DATABASE_URL.startswith("sqlite"), # ✅ SQLite 不需要
)
2.4 时间戳字段处理 (src/database.py)
SQLite 不支持 now() 函数，必须使用 Python 默认值：
class TimestampTableMixin:
_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
if _is_sqlite:
# ✅ SQLite: 使用 Python 默认值
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=text("(CURRENT_TIMESTAMP)"),
nullable=False,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=text("(CURRENT_TIMESTAMP)"),
onupdate=text("(CURRENT_TIMESTAMP)"),
nullable=False,
)
else:
# PostgreSQL: 使用数据库默认值
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(), # 数据库默认值
nullable=False,
)
# ...注意点：
动态判断：使用
_is_sqlite标志在运行时判断数据库类型Python 默认值：SQLite 必须在 Python 层面设置时间戳，而不是数据库层面
迁移影响：如果从 PostgreSQL 迁移已有数据，需要创建迁移脚本修复时间戳字段
mapped_column传参：server_default参数是数据库自动添加的，而default是python内置自动添加的
2.5 DDL 操作处理（关键改动）
2.5.1 问题背景
SQLite 对并发写入有严格限制。当会话已有活跃事务时，使用不同连接执行 DDL 会导致 “database is locked” 错误。
2.5.2 错误示例
# ❌ 错误：使用全局引擎会创建新连接
table.create(engine, checkfirst=True) # 会导致锁定
# ❌ 错误：使用 get_bind() 可能仍会使用连接池
table.create(db.get_bind(), checkfirst=True) # 仍可能锁定
2.5.3 正确做法
# ✅ 正确：使用当前会话的连接
table.create(db.connection(), checkfirst=True)
table.drop(db.connection(), checkfirst=True)
注意点：
使用
db.connection()：确保 DDL 操作使用与当前事务相同的连接同一事务中：所有操作必须在同一个事务和连接中完成
应用范围：所有动态创建/删除表的操作都需要修改
2.5.4 实际代码示例
# src/business_report/router.py
# 创建动态表
table = business_report.to_sqlalchemy_table(include_system_column=True)
table.create(db.connection(), checkfirst=True) # ✅ 使用当前连接
# 删除动态表
table = report.to_sqlalchemy_table(include_system_column=True)
table.drop(db.connection(), checkfirst=True) # ✅ 使用当前连接
2.6 数据库迁移（Alembic）
2.6.1 迁移脚本调整
alembic/env.py 无需特殊修改，会自动使用 settings.DATABASE_URL。
2.6.2 时间戳字段迁移
如果需要从 PostgreSQL 迁移已有数据，需要创建迁移脚本修复时间戳字段的默认值：
# alembic/versions/2025-12-24_19e269aa0a4e_fix_sqlite_timestamp_defaults.py
def upgrade() -> None:
# SQLite 不支持 ALTER TABLE ... SET DEFAULT
# 如果之前使用了 server_default，需要手动处理
pass # 新项目不需要，旧项目需要数据迁移逻辑
def downgrade() -> None:
pass
2.7 SQLite 工具函数 (src/utils/sqlite_util.py)
新增的工具函数用于管理 SQLite 数据库路径：
def get_app_db_path(app_name: str, db_name: str) -> Path:
"""获取应用数据目录下的数据库路径"""
# Windows: %APPDATA%\app_name\db_name
# Linux/Mac: $HOME/.config/app_name/db_name
...
def get_sqlite_connection_string(app_name: str, db_name: str) -> str:
"""生成 SQLite 连接字符串"""
...
三、关键注意点总结
3.1 连接配置
3.2 DDL 操作
3.3 数据类型差异
3.4 并发限制
SQLite 的并发限制比较严格：
同一时间只能有一个写操作
读操作可以并发
DDL 操作必须在同一连接中执行
长时间事务会阻塞其他操作
建议：
尽量缩短事务时间
DDL 操作使用
db.connection()确保同一连接考虑启用 WAL 模式（如果需要更高并发）
3.5 性能考虑
连接池：SQLite 不需要连接池，但
sessionmaker仍可正常使用索引：SQLite 的索引性能良好，建议保持合理的索引设计
文件大小：SQLite 数据库是单个文件，注意备份和管理
并发写入：如果有多进程写入需求，考虑使用文件锁或消息队列
3.6 测试环境
测试代码（tests/conftest.py）仍使用 PostgreSQL，因为：
PostgreSQL 测试容器更方便
测试环境需要隔离
生产环境使用 SQLite 是独立的决策
四、常见问题排查
4.1 “database is locked” 错误
原因：
DDL 操作使用了不同的连接
有长时间未提交的事务
有其他进程/线程正在写入
解决：
确保 DDL 使用
db.connection()检查是否有未提交的事务
确保没有其他程序打开数据库文件
4.2 时间戳字段为 NULL
原因：
迁移前使用了
server_default=func.now()SQLite 不支持数据库层面的默认值函数
解决：
确保模型使用 Python
default如果是迁移的数据，需要手动更新 NULL 值
4.3 路径问题（Windows）
原因：
Windows 路径格式
C:\path\to\db.db需要特殊处理
解决：
使用
Path.as_posix()转换为 POSIX 格式连接字符串格式：
sqlite+aiosqlite:///C:/path/to/db.db
五、回滚到 PostgreSQL
如果需要回滚到 PostgreSQL：
修改
src/conf.py中的DATABASE_URL和SYNC_DATABASE_URL确保已安装
asyncpg和psycopg2-binary时间戳字段会自动使用数据库默认值（
TimestampTableMixin已处理）DDL 操作可以改回使用
db.get_bind()（PostgreSQL 支持并发 DDL）
六、最佳实践建议
数据库位置：使用 APPDATA 目录，便于用户管理
连接管理：DDL 操作始终使用
db.connection()事务管理：尽量缩短事务时间，及时提交
错误处理：捕获并处理 “database is locked” 错误
备份策略：定期备份 SQLite 数据库文件
性能监控：关注长时间运行的查询和事务
附录：迁移涉及的源码汇总
配置：src/conf.py
from src.utils.sqlite_util import get_app_db_path
class Settings(BaseSettings):
_db_path = get_app_db_path("tianlu_spc", "tianlu_database.db")
DATABASE_URL: str = f"sqlite+aiosqlite:///{_db_path.as_posix()}"
SYNC_DATABASE_URL: str = f"sqlite:///{_db_path.as_posix()}"
数据库引擎：src/database.py
# 异步引擎
connect_args = {}
if settings.DATABASE_URL.startswith("sqlite"):
connect_args = {"check_same_thread": False}
async_engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
future=True,
connect_args=connect_args,
pool_pre_ping=not settings.DATABASE_URL.startswith("sqlite"),
)
# 同步引擎
sync_connect_args = {}
if settings.SYNC_DATABASE_URL.startswith("sqlite"):
sync_connect_args = {"check_same_thread": False}
engine = create_engine(
settings.SYNC_DATABASE_URL,
echo=settings.DEBUG,
connect_args=sync_connect_args,
pool_pre_ping=not settings.SYNC_DATABASE_URL.startswith("sqlite"),
)
时间戳字段：src/database.py
class TimestampTableMixin:
_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
if _is_sqlite:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=text("(CURRENT_TIMESTAMP)"),
nullable=False,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=text("(CURRENT_TIMESTAMP)"),
onupdate=text("(CURRENT_TIMESTAMP)"),
nullable=False,
)
else:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)DDL 操作示例：src/business_report/router.py
# 创建动态表
table = business_report.to_sqlalchemy_table(include_system_column=True)
table.create(db.connection(), checkfirst=True)
# 删除动态表
table = report.to_sqlalchemy_table(include_system_column=True)
table.drop(db.connection(), checkfirst=True)
