feat(db): C3 langgraph 检查点建表迁移 + alembic 漂移豁免
多章工作流链(docs/design/chain-workflow.md §4/§7)唯一新增 DDL:langgraph 检查点四表(checkpoints/checkpoint_blobs/checkpoint_writes/checkpoint_migrations)。 - 新增迁移 d3e4f5a6b7c8:在迁移内调 langgraph 同步 PostgresSaver.setup() 建表 (DDL 只在 migrations/CI,绝不 app-runtime)。从 database_url 派生 psycopg conn string(去 +asyncpg),与 alembic 的 asyncpg 连接解耦。 - §11 风险落地:setup() 跑 CREATE INDEX CONCURRENTLY,不能在事务块内且会与 alembic 开着的事务死锁 → 用 op.get_context().autocommit_block() 先提交退出 事务块再建表。downgrade 幂等 DROP ... CASCADE。 - env.py 加 include_object:langgraph 四表不在 Base.metadata,过滤掉以免 alembic check 误报漂移。 - jobs 表零迁移(C2 判定:result.awaiting_chapter 表达待裁决,复用现有状态枚举)。 验证:alembic upgrade head 应用、downgrade/re-upgrade 往返、alembic check 无漂移;ruff/format/mypy(packages/db) 全绿。
This commit is contained in:
@@ -20,9 +20,27 @@ if config.config_file_name is not None:
|
||||
config.set_main_option("sqlalchemy.url", get_settings().database_url)
|
||||
target_metadata = Base.metadata
|
||||
|
||||
# langgraph 检查点表由 langgraph 自管 schema(迁移 d3e4f5a6b7c8 调其 setup() 建表),
|
||||
# 不在 Base.metadata。autogenerate/check 须忽略它们,否则会误报"待 drop"漂移。
|
||||
_LANGGRAPH_TABLES = frozenset(
|
||||
{"checkpoints", "checkpoint_blobs", "checkpoint_writes", "checkpoint_migrations"}
|
||||
)
|
||||
|
||||
|
||||
def include_object(
|
||||
obj: object, name: str | None, type_: str, reflected: bool, compare_to: object
|
||||
) -> bool:
|
||||
if type_ == "table" and name in _LANGGRAPH_TABLES:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def do_run_migrations(connection: Connection) -> None:
|
||||
context.configure(connection=connection, target_metadata=target_metadata)
|
||||
context.configure(
|
||||
connection=connection,
|
||||
target_metadata=target_metadata,
|
||||
include_object=include_object,
|
||||
)
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
"""langgraph checkpoint setup (multi-chapter chain workflow)
|
||||
|
||||
Revision ID: d3e4f5a6b7c8
|
||||
Revises: c2b3d4e5f6a7
|
||||
Create Date: 2026-06-23 10:00:00.000000
|
||||
|
||||
多章工作流链(docs/design/chain-workflow.md §4/§7)唯一新增 DDL:langgraph
|
||||
检查点表(`checkpoints` / `checkpoint_blobs` / `checkpoint_writes` /
|
||||
`checkpoint_migrations`)。这些表**不在** ``Base.metadata``(由 langgraph 自己
|
||||
管理 schema),故迁移直接调用 langgraph 的一次性 ``setup()`` 建表,而非 ORM
|
||||
``op.create_table``。env.py 的 ``include_object`` 过滤这些表,避免 ``alembic
|
||||
check`` 误报漂移(要 drop 这些"非 metadata"表)。
|
||||
|
||||
实现说明(CLAUDE.md「LangGraph」+ 设计 §11 回退):
|
||||
- DDL 只在 **migrations/CI** 跑(绝不在 app-runtime)——本迁移即该入口。
|
||||
- ``upgrade()`` 跑在 alembic 的**同步**迁移上下文里(``connection.run_sync``
|
||||
下已有 event loop 在跑 async 迁移),故**不能**调 async ``setup_checkpointer``
|
||||
/``asyncio.run``(会嵌套 loop 报错)。改用 langgraph 同步 ``PostgresSaver``
|
||||
(psycopg),用一条从 ``database_url`` 派生的 psycopg conn string 自建连接,
|
||||
与 alembic 的 asyncpg 连接解耦。``setup()`` 幂等(langgraph 内部 migration 版本表)。
|
||||
- **关键**:langgraph ``setup()`` 跑 ``CREATE INDEX CONCURRENTLY``,它不能在
|
||||
事务块内执行,且会等 alembic 那条仍开着事务(持 ``alembic_version`` 锁)的连接
|
||||
→ 死锁(§11 风险落地)。故用 ``op.get_context().autocommit_block()`` 把 alembic
|
||||
的事务先提交、退出事务块,再让 langgraph 在独立 autocommit 连接里建表/建索引。
|
||||
|
||||
jobs 表:C2 判定零迁移复用(用 ``result.awaiting_chapter`` 表达待裁决,
|
||||
``status`` 复用现有枚举),故本迁移不动 jobs。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
revision: str = "d3e4f5a6b7c8"
|
||||
down_revision: str | None = "c2b3d4e5f6a7"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
# langgraph 检查点表(langgraph.checkpoint.postgres.base 的 MIGRATIONS 建的全部表)
|
||||
_CHECKPOINT_TABLES: tuple[str, ...] = (
|
||||
"checkpoint_writes",
|
||||
"checkpoint_blobs",
|
||||
"checkpoints",
|
||||
"checkpoint_migrations",
|
||||
)
|
||||
|
||||
|
||||
def _psycopg_conn_string() -> str:
|
||||
"""从 settings.database_url 派生 psycopg(同步)conn string。
|
||||
|
||||
settings 用 asyncpg URL(``postgresql+asyncpg://…``);langgraph 的同步
|
||||
``PostgresSaver`` 走 psycopg,需去掉 SQLAlchemy 的 ``+asyncpg`` 驱动后缀。
|
||||
"""
|
||||
from ww_config import get_settings
|
||||
|
||||
return get_settings().database_url.replace("+asyncpg", "")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
from alembic import op
|
||||
from langgraph.checkpoint.postgres import PostgresSaver
|
||||
|
||||
# 退出 alembic 事务块(提交 alembic_version 更新,释放其锁),langgraph 的
|
||||
# CREATE INDEX CONCURRENTLY 才能在独立连接里跑而不死锁。
|
||||
with op.get_context().autocommit_block():
|
||||
with PostgresSaver.from_conn_string(_psycopg_conn_string()) as saver:
|
||||
saver.setup()
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
import psycopg
|
||||
|
||||
with psycopg.connect(_psycopg_conn_string()) as conn:
|
||||
conn.execute("DROP TABLE IF EXISTS " + ", ".join(_CHECKPOINT_TABLES) + " CASCADE")
|
||||
conn.commit()
|
||||
Reference in New Issue
Block a user