feat(db): 支持接团队共享库 fnsvr-01(PG 17.6),测试库与开发库分离
CI / backend (push) Canceled after 0s

fnsvr-01 的 PostgreSQL 在 8432 端口(不是 5432),实测 PostgreSQL 17.6 / Alpine。
已在该实例建好独立库与角色:
- dadaotask(开发)/ dadaotask_test(测试),owner 都是 dadao_owner
- dadao_owner 迁移用;dadao_app 应用用(NOSUPERUSER NOBYPASSRLS)
- 两个库都已 ALTER SCHEMA public OWNER TO dadao_owner + REVOKE CREATE FROM PUBLIC + GRANT 给 dadao_app

代码改动:
- config 增加 test_database_url / test_migration_database_url,并给出
  resolved_database_url() / resolved_migration_database_url():ENV=test 且有测试库时优先用测试库,
  否则退回开发库(CI 里直接给 DATABASE_URL 即可,行为不变)。
- db.py / alembic/env.py / tests/conftest.py 统一走这两个解析方法。
  原先 conftest 清表用的是 settings.migration_database_url,而应用连的是测试库,
  一旦测试库与开发库不同,TRUNCATE 就清到另一个库上——表现为"测试库脏数据越攒越多、
  列表类断言随机失败"。这是本次真踩到的坑,不是假想。
- tests/conftest.py 保留 E402 豁免(环境变量必须在 import app 模块前写好)。

验证:
- 远程 dadaotask_test 上 pytest 33 passed,跑完表内 0 行(确认清表打在同一个库)
- 远程 dadaotask 上 alembic upgrade head 成功,/health 与 /version 正常,db_role 为 dadao_app 且三项全 false
- 本机便携 PG 16.4 上同样 33 passed(两个 PG 大版本都过)
- ruff check / format 干净

已知限制:该实例没有 pgvector(只有 pg_trgm / pgcrypto),M3 向量检索前必须先定扩展来源,
已写进 2.Design/DEVELOPMENT.md。
This commit is contained in:
Steven
2026-10-28 03:37:41 +08:00
parent b40ea77eec
commit d390b7b58a
7 changed files with 66 additions and 3 deletions
+20
View File
@@ -90,6 +90,26 @@ ruff check . && ruff format --check .
| Docker 起不来 | Docker Desktop 未启动 / 需要管理员 | 先启动 Desktop,`docker info` 有输出再继续 |
| 迁移报 `cannot insert multiple commands into a prepared statement` | asyncpg 走预备语句,一次只能一条 SQL | 多语句 DDL 拆成每语句一次 `op.execute()` |
### 团队共享开发库(fnsvr-01:8432)
`fnsvr-01`(`192.168.8.160`)上有一个 **PostgreSQL 17.6**(Alpine),端口是 **8432**(不是 5432),
已为本项目建好独立库与角色:
| 库 | 用途 | owner |
|---|---|---|
| `dadaotask` | 开发库 | `dadao_owner` |
| `dadaotask_test` | 测试库(**测试会 TRUNCATE 所有表,绝不能用开发库**) | `dadao_owner` |
角色:`dadao_owner`(迁移用,库 owner)与 `dadao_app`(应用用,非 superuser / 无 BYPASSRLS)。
连接串见 `backend/.env.example` 里的注释行,直接拷进 `backend/.env` 即可。
已验证:在该库上 `alembic upgrade head` 成功,`pytest` **33 passed**(服务端 PG 17.6 vs 本机 16.4,两版都过)。
> ⚠️ **这台服务器没有 pgvector**(`pg_available_extensions` 里只有 `pg_trgm`、`pgcrypto`)。
> M3 知识库要做向量检索的话,必须先解决扩展来源:装 pgvector(需要服务器 root 装 OS 包)/
> 给知识库单独跑一个带 pgvector 的容器 / 或退回「纯全文检索 + trigram」先不上语义检索。
> 这件事必须在写 `chunk` 表和检索接口之前定下来,否则要返工。
### 本机特例(2026-10 实测,换机器可跳过)
这台开发机的 Docker Desktop 起不来,且外网 TLS 被拦,实际用的是:
+15
View File
@@ -2,17 +2,32 @@
ENV=dev
APP_NAME=达道Task
# ── 数据库 ───────────────────────────────────────────────────────────
# 应用角色(受 RLS 约束)—— 应用进程只能用它
DATABASE_URL=postgresql+asyncpg://dadao_app:dadao_dev_pw@127.0.0.1:5433/dadaotask
# 迁移角色(owner,绕过 RLS)—— 只有 alembic 用
MIGRATION_DATABASE_URL=postgresql+asyncpg://dadao_owner:dadao_dev_pw@127.0.0.1:5433/dadaotask
# 团队共享库(fnsvr-01,PostgreSQL 17.6)。断网时用上面的本机便携 PG。
# DATABASE_URL=postgresql+asyncpg://dadao_app:dadao_dev_pw@192.168.8.160:8432/dadaotask
# MIGRATION_DATABASE_URL=postgresql+asyncpg://dadao_owner:dadao_dev_pw@192.168.8.160:8432/dadaotask
# 测试库必须独立:测试会 TRUNCATE 所有表,打到开发库上数据就没了
# TEST_DATABASE_URL=postgresql+asyncpg://dadao_app:dadao_dev_pw@192.168.8.160:8432/dadaotask_test
# TEST_MIGRATION_DATABASE_URL=postgresql+asyncpg://dadao_owner:dadao_dev_pw@192.168.8.160:8432/dadaotask_test
REDIS_URL=redis://127.0.0.1:6380/0
# ── 认证 ─────────────────────────────────────────────────────────────
# 生产环境启动时会校验:JWT_SECRET 不得是默认值且长度 ≥ 32
JWT_SECRET=dev-only-change-me
ACCESS_TOKEN_MINUTES=30
REFRESH_TOKEN_DAYS=14
# 生产环境禁止 SMS_PROVIDER=dev(否则任意验证码可登录),启动即失败
SMS_PROVIDER=dev
SMS_DEV_CODE=123456
# 同一手机号两次发码的最小间隔(秒),测试里置 0 可关闭
SMS_THROTTLE_SECONDS=60
CORS_ORIGINS=http://localhost:5180,http://127.0.0.1:5180
+3 -1
View File
@@ -17,7 +17,9 @@ config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
config.set_main_option("sqlalchemy.url", get_settings().migration_database_url)
_settings = get_settings()
# 测试环境用独立测试库(conftest 会在 ENV=test 下跑迁移)
config.set_main_option("sqlalchemy.url", _settings.resolved_migration_database_url())
target_metadata = Base.metadata
+16
View File
@@ -32,6 +32,10 @@ class Settings(BaseSettings):
migration_database_url: str = (
"postgresql+asyncpg://dadao_owner:dadao_dev_pw@127.0.0.1:5433/dadaotask"
)
# 测试专用库:测试会 TRUNCATE 所有表,绝不能落到开发库上。
# 留空则退回 database_url(CI 里直接给 DATABASE_URL 即可)。
test_database_url: str | None = None
test_migration_database_url: str | None = None
db_echo: bool = False
db_pool_size: int = 10
@@ -79,6 +83,18 @@ class Settings(BaseSettings):
if problems:
raise RuntimeError("配置校验失败:" + ";".join(problems))
def resolved_database_url(self) -> str:
"""应用实际连的库:测试环境优先用独立测试库(测试会 TRUNCATE)。"""
if self.env == "test" and self.test_database_url:
return self.test_database_url
return self.database_url
def resolved_migration_database_url(self) -> str:
"""迁移/清表实际连的库:必须和应用同一个库,否则清表清到别处去了。"""
if self.env == "test" and self.test_migration_database_url:
return self.test_migration_database_url
return self.migration_database_url
@lru_cache
def get_settings() -> Settings:
+4 -1
View File
@@ -25,6 +25,9 @@ from .errors import AppError, ErrorCode
_settings = get_settings()
# 测试环境优先用独立测试库:测试会 TRUNCATE 所有表,落到开发库上数据就没了。
_db_url = _settings.resolved_database_url()
_engine_kwargs: dict[str, Any] = {"echo": _settings.db_echo, "pool_pre_ping": True}
if _settings.env == "test":
# pytest-asyncio 每个用例一个事件循环,连接池跨循环复用会报
@@ -33,7 +36,7 @@ if _settings.env == "test":
else:
_engine_kwargs["pool_size"] = _settings.db_pool_size
engine = create_async_engine(_settings.database_url, **_engine_kwargs)
engine = create_async_engine(_db_url, **_engine_kwargs)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False, autoflush=False)
+3
View File
@@ -10,6 +10,9 @@ ignore = ["S101"] # 测试里允许 assert
[tool.ruff.lint.per-file-ignores]
"tests/*" = ["S105", "S106"] # 测试里的假令牌/假口令
"alembic/env.py" = ["E402"]
# conftest 必须在 import 任何 app 模块之前写好环境变量(否则配置已被缓存),
# 因此 import 语句不能提到文件顶部。
"tests/conftest.py" = ["E402"]
[tool.ruff.lint.isort]
known-first-party = ["app"]
+5 -1
View File
@@ -13,6 +13,8 @@ os.environ["ENV"] = "test"
os.environ.setdefault("SMS_DEV_CODE", "123456")
# 测试里关掉发码节流,否则"同一手机号再登录一次"要等 60 秒
os.environ.setdefault("SMS_THROTTLE_SECONDS", "0")
# 测试库由 config 的 test_database_url / test_migration_database_url 决定(见 backend/.env),
# 这里不再改 os.environ —— 配置对象自己会读 .env。
import uuid
from collections.abc import AsyncIterator
@@ -49,7 +51,9 @@ def migrated_database() -> None:
@pytest.fixture(autouse=True)
async def clean_database(migrated_database: None) -> AsyncIterator[None]:
owner_engine = create_async_engine(settings.migration_database_url, poolclass=NullPool)
owner_engine = create_async_engine(
settings.resolved_migration_database_url(), poolclass=NullPool
)
async with owner_engine.begin() as conn:
await conn.execute(text(f"TRUNCATE {_TABLES} CASCADE"))
await owner_engine.dispose()