chenshuobo
b76e5bd515
按 docs/数据库迁移.md 第 3 节扫雷清单落地代码层改动: - pyproject.toml: 新增 psycopg[binary]>=3.1 依赖 (psycopg3, SQLAlchemy 2.0 时代默认, 不要再装 psycopg2) - app/db/session.py: 非 SQLite 自动加连接池参数 - pool_size=10, max_overflow=20, pool_recycle=3600 - SQLite 单文件不池化, _is_sqlite 判断分支保留以便本地 dev 临时回退 - app/models/savings.py: dishes 列从 JSON 改为 JSONB - PG 上能用 GIN 索引和 jsonb 操作符 - SQLite 上 JSON 实际是 TEXT, 切到 PG 后用 JSON 类型存的是 json 不是 jsonb - alembic/versions/ef96beb47b1e_*.py: 新增迁移把 savings_record.dishes 从 json 转 jsonb - 用 op.get_bind().dialect.name 判断 PG 才执行 ALTER COLUMN - SQLite 上是 no-op (SQLite 没有 jsonb 类型) - 旧的 savings_shop_dishes.py 迁移保持 sa.JSON() 不动 (按文档"不改老迁移"原则) 本地切 PG 验证: 10 张表全部建出, dishes 字段为 jsonb 类型, 登录/me/美团 feed 三个真接口全 200, user 数据正确写入 PG. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""SQLAlchemy engine + Session 工厂。
|
|
|
|
FastAPI 路由通过 `Depends(get_db)` 拿到一个 request-scope 的 Session,
|
|
请求结束自动 close。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Iterator
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def _ensure_sqlite_dir(url: str) -> None:
|
|
"""SQLite 文件路径如果在子目录,首启时自动 mkdir,避免 sqlalchemy 报 unable to open。"""
|
|
prefix = "sqlite:///"
|
|
if not url.startswith(prefix):
|
|
return
|
|
path = url[len(prefix):]
|
|
dirname = os.path.dirname(path)
|
|
if dirname:
|
|
os.makedirs(dirname, exist_ok=True)
|
|
|
|
|
|
_ensure_sqlite_dir(settings.DATABASE_URL)
|
|
|
|
_is_sqlite = settings.DATABASE_URL.startswith("sqlite")
|
|
|
|
# SQLite 跨线程访问要 check_same_thread=False;PG/MySQL 不需要这个参数
|
|
_connect_args: dict = {}
|
|
if _is_sqlite:
|
|
_connect_args["check_same_thread"] = False
|
|
|
|
_engine_kwargs: dict = {
|
|
"connect_args": _connect_args,
|
|
# echo 在 dev 下打 SQL,生产关掉
|
|
"echo": settings.APP_DEBUG and not settings.is_prod,
|
|
"future": True,
|
|
"pool_pre_ping": True,
|
|
}
|
|
# SQLite 用单文件不需要池;PG/MySQL 必须显式池化 + recycle 防 idle 断连
|
|
if not _is_sqlite:
|
|
_engine_kwargs.update(pool_size=10, max_overflow=20, pool_recycle=3600)
|
|
|
|
engine = create_engine(settings.DATABASE_URL, **_engine_kwargs)
|
|
|
|
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False)
|
|
|
|
|
|
def get_db() -> Iterator[Session]:
|
|
"""FastAPI 依赖。请求开始建 session,请求结束自动 close。
|
|
|
|
出异常时 SQLAlchemy 默认会标记 transaction 为 invalid,close 时自动 rollback。
|
|
"""
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|