mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(persistence): compare safe_add_column drift as the database spells it (#6865)
* fix(persistence): compare safe_add_column drift as the database spells it
safe_add_column's drift check compared the model's generic type and
default spellings (String -> STRING, BigInteger -> BIGINTEGER,
server_default="0" -> "DefaultClause('0', ...)", sa.false()) with the
reflected dialect spellings (VARCHAR, BIGINT, Postgres TIMESTAMP, '0').
Only JSON/JSONB ever matched, so every column a revision found already
present was reported as drift:
- any database created before 0023 logged a runs.change_seq warning on
upgrade, because 0025 re-runs 0023's safe_add_column;
- a schema whose head stamp was lost (the empty branch commits create_all
before stamping) logged 29 warnings on SQLite and 36 on Postgres, burying
the real TEXT-vs-JSON drift the check exists to surface.
Render both sides through the migration connection's dialect before
comparing: types via TypeEngine.compile (parameters stripped, the
Postgres time-zone suffix kept), defaults by unwrapping DefaultClause,
compiling SQL expressions and stripping one layer of string quoting.
Types the dialect cannot render (SQLite NullType) stay "missing info".
* docs(changelog): reference #6865 for the drift comparator fix
* fix(persistence): keep an empty-string default distinct from no default
Stripping one layer of quotes turned a reflected DEFAULT '' into "", and
`return text or None` then collapsed it to None. A desired
server_default="" (passed as a bare "") also normalized to None, so
_check_column_drift stopped warning when a column had no default but the
model wanted '' (or the reverse).
Render a plain-string DefaultClause argument as the quoted literal the DDL
emits, and return the unquoted value even when it is empty. None now means
only "no default".
This commit is contained in:
Hyeonsang Cho
GitHub
parent
517a8309a2
commit
268eb55f6c
10
CHANGELOG.md
10
CHANGELOG.md
@ -5917,6 +5917,15 @@ This release closes that milestone with **621 merged pull requests**.
|
||||
80 such requests failed. The same race in IM conversation binding, such as
|
||||
a first message arriving together with `/new`, no longer fails the message.
|
||||
The losing write now updates the row the other one created. ([#6852])
|
||||
- **persistence:** Startup schema upgrades no longer log a false
|
||||
`safe_add_column ... drifts from the model definition` warning for columns
|
||||
that already match the model. The check compared the model's generic type
|
||||
and default names (`STRING`, `BIGINTEGER`, `DefaultClause('0')`) with the
|
||||
database's own (`VARCHAR`, `BIGINT`, `'0'`), so every Postgres or SQLite
|
||||
database created before revision 0023 logged one on upgrade, and a schema
|
||||
whose head stamp was lost logged 29 (SQLite) or 36 (Postgres), burying the
|
||||
real type mismatch the warning exists to catch. Both sides are now compared
|
||||
as the connected database spells them. ([#6865])
|
||||
|
||||
### Security
|
||||
|
||||
@ -11701,3 +11710,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6843]: https://github.com/bytedance/deer-flow/pull/6843
|
||||
[#6844]: https://github.com/bytedance/deer-flow/pull/6844
|
||||
[#6852]: https://github.com/bytedance/deer-flow/pull/6852
|
||||
[#6865]: https://github.com/bytedance/deer-flow/pull/6865
|
||||
|
||||
@ -4494,6 +4494,13 @@
|
||||
`(thread, run, user)` 唯一约束；在 Postgres 上，80 个这样的请求中有 26 个失败。
|
||||
IM 会话绑定中的同类竞争（例如首条消息与 `/new` 同时到达）也不再导致消息处理失败。
|
||||
落后的写入现在会更新另一方创建的记录。([#6852])
|
||||
- **持久化：** 启动时的 schema 升级不再为已与模型一致的列记录虚假的
|
||||
`safe_add_column ... drifts from the model definition` 警告。该检查拿模型的通用类型名
|
||||
和默认值写法（`STRING`、`BIGINTEGER`、`DefaultClause('0')`）去比较数据库自己的写法
|
||||
（`VARCHAR`、`BIGINT`、`'0'`），因此 revision 0023 之前创建的 Postgres 或 SQLite
|
||||
数据库升级时都会记录一条，丢失 head stamp 的 schema 会记录 29 条（SQLite）或
|
||||
36 条（Postgres），淹没了这个警告本应发现的真实类型不一致。现在两侧都按所连接
|
||||
数据库的写法进行比较。([#6865])
|
||||
|
||||
### 安全
|
||||
|
||||
@ -9337,3 +9344,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
|
||||
[#6843]: https://github.com/bytedance/deer-flow/pull/6843
|
||||
[#6844]: https://github.com/bytedance/deer-flow/pull/6844
|
||||
[#6852]: https://github.com/bytedance/deer-flow/pull/6852
|
||||
[#6865]: https://github.com/bytedance/deer-flow/pull/6865
|
||||
|
||||
@ -15,7 +15,7 @@ DeerFlow's application tables (`runs`, `threads_meta`, `feedback`, `users`, `run
|
||||
| `0019_thread_incarnations` missing local tables/columns | refuse startup; offline recovery required |
|
||||
| unknown revision, empty version table, or multiple version rows | fail closed and refuse to start |
|
||||
|
||||
The legacy branch handles pre-alembic databases that already have at least one DeerFlow-owned table. `create_all` runs first because stamping at `0001_baseline` makes alembic skip the baseline's own `create_table` DDL on the subsequent upgrade — so any baseline table introduced into `Base.metadata` after the user's DB was first provisioned (e.g. the `channel_*` tables from PR #1930 for users upgrading across multiple releases) would otherwise never be created, and the first request hitting that table would 500 with `no such table`. The backfill is **restricted to `_BASELINE_TABLE_NAMES`** so it does not also create tables that future revisions introduce — those revisions' own `op.create_table` would otherwise fail with `relation already exists`. A guard test pins `_BASELINE_TABLE_NAMES` against `0001_baseline.upgrade()`'s actual output, so editing 0001 to add or remove a table forces a matching update to the constant. Column-level shape (pre-#3658 vs post-#3658 vs manual-ALTER for `token_usage_by_model`) is answered by each `versions/*.py` revision via the idempotent helpers in `migrations/_helpers.py` (`safe_add_column` / `safe_drop_column`) which no-op when the change is already present and `logger.warning` on shape drift. **Adding a new ORM column / table only requires a new revision file — no edit to `bootstrap.py` is needed** *unless* the new revision adds a new baseline table (rare; only happens when a new model is part of the baseline rather than introduced by its own revision).
|
||||
The legacy branch handles pre-alembic databases that already have at least one DeerFlow-owned table. `create_all` runs first because stamping at `0001_baseline` makes alembic skip the baseline's own `create_table` DDL on the subsequent upgrade — so any baseline table introduced into `Base.metadata` after the user's DB was first provisioned (e.g. the `channel_*` tables from PR #1930 for users upgrading across multiple releases) would otherwise never be created, and the first request hitting that table would 500 with `no such table`. The backfill is **restricted to `_BASELINE_TABLE_NAMES`** so it does not also create tables that future revisions introduce — those revisions' own `op.create_table` would otherwise fail with `relation already exists`. A guard test pins `_BASELINE_TABLE_NAMES` against `0001_baseline.upgrade()`'s actual output, so editing 0001 to add or remove a table forces a matching update to the constant. Column-level shape (pre-#3658 vs post-#3658 vs manual-ALTER for `token_usage_by_model`) is answered by each `versions/*.py` revision via the idempotent helpers in `migrations/_helpers.py` (`safe_add_column` / `safe_drop_column`) which no-op when the change is already present and `logger.warning` on shape drift (both sides rendered through the connection's dialect, so model-shaped columns stay quiet). **Adding a new ORM column / table only requires a new revision file — no edit to `bootstrap.py` is needed** *unless* the new revision adds a new baseline table (rare; only happens when a new model is part of the baseline rather than introduced by its own revision).
|
||||
|
||||
The empty-DB path keeps using `create_all` because `Base.metadata` is the only authoritative schema source — `create_all` renders both SQLite (JSON, type affinity) and Postgres (JSONB, partial indexes) correctly without anyone having to keep a hand-written baseline in lockstep. `0001_baseline.upgrade()` is therefore almost never executed in practice; it exists as a stamp target + chain root.
|
||||
|
||||
|
||||
@ -27,11 +27,15 @@ JSON`` without ``NOT NULL DEFAULT '{}'``, or the wrong-type variant
|
||||
would produce on a fresh DB. To surface that silent drift, ``safe_add_column``
|
||||
compares the existing column's ``nullable`` / ``server_default`` / ``type``
|
||||
against the desired ``sa.Column`` and emits ``logger.warning`` on mismatch.
|
||||
Type comparison goes through ``_type_equivalent``, which treats known
|
||||
dialect-synonym pairs (e.g. ``JSON`` vs ``JSONB``) as equivalent to avoid
|
||||
false positives while still catching wholesale type mismatches like
|
||||
``TEXT`` vs ``JSON``. We do not auto-repair -- a warning is enough for
|
||||
operators to notice and decide.
|
||||
Both sides are rendered through the migration connection's dialect before
|
||||
they are compared: the model says ``sa.String`` / ``sa.BigInteger`` /
|
||||
``server_default="0"`` while reflection reports ``VARCHAR`` / ``BIGINT`` /
|
||||
``'0'``, so comparing the raw Python objects would warn on every column a
|
||||
revision finds already present. Type comparison goes through
|
||||
``_type_equivalent``, which treats known dialect-synonym pairs (e.g. ``JSON``
|
||||
vs ``JSONB``) as equivalent to avoid false positives while still catching
|
||||
wholesale type mismatches like ``TEXT`` vs ``JSON``. We do not auto-repair --
|
||||
a warning is enough for operators to notice and decide.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -48,21 +52,32 @@ def _inspector() -> sa.Inspector:
|
||||
return sa.inspect(op.get_bind())
|
||||
|
||||
|
||||
def _normalize_default(value: object) -> str | None:
|
||||
def _normalize_default(value: object, dialect: sa.engine.Dialect | None = None) -> str | None:
|
||||
"""Normalize a server-default value for cross-source comparison.
|
||||
|
||||
The desired value comes from ``sa.Column.server_default`` (a
|
||||
``DefaultClause`` / ``TextClause`` literal, ``None``, or a Python literal);
|
||||
the reflected value comes from ``Inspector.get_columns()['default']`` as a
|
||||
dialect-rendered string. Strip outer parens / whitespace / Postgres-style
|
||||
type casts so textually-equivalent forms compare equal across dialects.
|
||||
``DefaultClause`` wrapping a string, ``TextClause`` or SQL expression such
|
||||
as ``sa.false()``; ``None``; or a Python literal); the reflected value
|
||||
comes from ``Inspector.get_columns()['default']`` as a dialect-rendered
|
||||
string. Expressions are compiled with *dialect* (``sa.false()`` is ``0``
|
||||
on SQLite and ``false`` on Postgres), then outer parens / whitespace /
|
||||
Postgres-style type casts / one layer of single quotes are stripped so
|
||||
``server_default="0"`` (DDL ``DEFAULT '0'``), SQLite's reflected ``'0'``
|
||||
and Postgres's reflected ``0`` or ``'0'::bigint`` all compare equal.
|
||||
Returns ``None`` only when there is no default: a quoted empty literal
|
||||
(``server_default=""``, reflected ``''``) stays ``""``.
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, sa.schema.DefaultClause):
|
||||
value = value.arg
|
||||
if isinstance(value, str):
|
||||
# DDL renders a plain-string default as a quoted literal.
|
||||
value = "'" + value.replace("'", "''") + "'"
|
||||
if isinstance(value, sa.sql.elements.TextClause):
|
||||
text = value.text
|
||||
elif isinstance(value, sa.schema.DefaultClause) and isinstance(value.arg, sa.sql.elements.TextClause):
|
||||
text = value.arg.text
|
||||
elif isinstance(value, sa.sql.elements.ClauseElement):
|
||||
text = str(value.compile(dialect=dialect, compile_kwargs={"literal_binds": True}))
|
||||
else:
|
||||
text = str(value)
|
||||
text = text.strip()
|
||||
@ -72,23 +87,62 @@ def _normalize_default(value: object) -> str | None:
|
||||
# Strip Postgres-style type casts like ``'{}'::jsonb``.
|
||||
if "::" in text:
|
||||
text = text.split("::", 1)[0].strip()
|
||||
# Strip one layer of SQL string quoting: a plain-string ``server_default``
|
||||
# renders as a quoted literal, and dialects disagree on whether a numeric
|
||||
# column reflects it back quoted. The unquoted value is returned even
|
||||
# when empty: ``DEFAULT ''`` is a default, unlike no default at all.
|
||||
if len(text) >= 2 and text.startswith("'") and text.endswith("'"):
|
||||
return text[1:-1].replace("''", "'")
|
||||
return text or None
|
||||
|
||||
|
||||
def _normalize_type(value: object) -> str:
|
||||
def _strip_type_parameters(rendered: str) -> str:
|
||||
"""Drop every parenthesized group, keeping words around it.
|
||||
|
||||
``TIMESTAMP(3) WITH TIME ZONE`` keeps its ``WITH TIME ZONE`` suffix, and
|
||||
nested reprs such as ``JSONB(astext_type=Text())`` lose the whole group.
|
||||
"""
|
||||
out: list[str] = []
|
||||
depth = 0
|
||||
for char in rendered:
|
||||
if char == "(":
|
||||
depth += 1
|
||||
elif char == ")":
|
||||
depth = max(depth - 1, 0)
|
||||
elif depth == 0:
|
||||
out.append(char)
|
||||
return " ".join("".join(out).split())
|
||||
|
||||
|
||||
def _normalize_type(value: object, dialect: sa.engine.Dialect | None = None) -> str:
|
||||
"""Normalize a SQLAlchemy ``TypeEngine`` (or reflected type) for comparison.
|
||||
|
||||
Returns the upper-cased type-class name with any parameters stripped
|
||||
(e.g. ``JSON()`` → ``"JSON"``, ``VARCHAR(255)`` → ``"VARCHAR"``). Length
|
||||
parameters are dropped on purpose: drift warnings target wholesale type
|
||||
A ``TypeEngine`` is rendered as the DDL *dialect* would emit for it, so
|
||||
the model's generic ``String(128)`` / ``BigInteger()`` /
|
||||
``DateTime(timezone=True)`` and the reflected ``VARCHAR(128)`` /
|
||||
``BIGINT`` / Postgres ``TIMESTAMP(timezone=True)`` land on the same name.
|
||||
Strings are taken as already-rendered type names. Returns the upper-cased
|
||||
name with any parameters stripped (e.g. ``VARCHAR(255)`` → ``"VARCHAR"``,
|
||||
Postgres ``TIMESTAMP WITH TIME ZONE`` stays whole). Length parameters are
|
||||
dropped on purpose: drift warnings target wholesale type
|
||||
misconfigurations (the JSON-vs-TEXT review case), not dialect-rendered
|
||||
size defaults. An empty string signals "missing info" -- callers should
|
||||
not equality-check empty strings.
|
||||
size defaults. An empty string signals "missing info" (no type, or one
|
||||
the dialect cannot render, such as SQLite's ``NullType`` for an
|
||||
unrecognized declared type) -- callers should not equality-check empty
|
||||
strings.
|
||||
"""
|
||||
if value is None:
|
||||
return ""
|
||||
s = value if isinstance(value, str) else repr(value)
|
||||
return s.upper().split("(", 1)[0].strip()
|
||||
if isinstance(value, str):
|
||||
rendered = value
|
||||
elif isinstance(value, sa.types.TypeEngine):
|
||||
try:
|
||||
rendered = str(value.compile(dialect=dialect))
|
||||
except sa.exc.CompileError:
|
||||
return ""
|
||||
else:
|
||||
rendered = repr(value)
|
||||
return _strip_type_parameters(rendered.upper())
|
||||
|
||||
|
||||
# Known dialect-synonym pairs that must NOT fire a type-drift warning.
|
||||
@ -104,14 +158,15 @@ def _normalize_type(value: object) -> str:
|
||||
_EQUIVALENT_TYPE_FAMILIES: tuple[frozenset[str], ...] = (frozenset({"JSON", "JSONB"}),)
|
||||
|
||||
|
||||
def _type_equivalent(actual: object, desired: object) -> bool:
|
||||
def _type_equivalent(actual: object, desired: object, dialect: sa.engine.Dialect | None = None) -> bool:
|
||||
"""True if *actual* and *desired* are the same type or a known equivalent.
|
||||
|
||||
Returns True when either side is missing reflected info so missing-data
|
||||
cases never false-positive into a noisy warning.
|
||||
Both sides are rendered with *dialect* (see ``_normalize_type``). Returns
|
||||
True when either side is missing reflected info so missing-data cases
|
||||
never false-positive into a noisy warning.
|
||||
"""
|
||||
a = _normalize_type(actual)
|
||||
d = _normalize_type(desired)
|
||||
a = _normalize_type(actual, dialect)
|
||||
d = _normalize_type(desired, dialect)
|
||||
if not a or not d:
|
||||
return True
|
||||
if a == d:
|
||||
@ -120,7 +175,7 @@ def _type_equivalent(actual: object, desired: object) -> bool:
|
||||
return any(pair <= fam for fam in _EQUIVALENT_TYPE_FAMILIES)
|
||||
|
||||
|
||||
def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None:
|
||||
def _check_column_drift(table: str, desired: sa.Column, actual: dict, dialect: sa.engine.Dialect | None = None) -> None:
|
||||
"""Warn if an existing column's attributes diverge from the desired model.
|
||||
|
||||
Equality is checked on ``nullable`` and ``server_default`` directly, and
|
||||
@ -128,7 +183,9 @@ def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None:
|
||||
pairs like ``JSON`` vs ``JSONB`` as equivalent). The reflected and
|
||||
desired type reprs are also echoed in the warning payload regardless of
|
||||
whether type was the failing dimension, so an operator triaging the log
|
||||
line sees the type context at a glance.
|
||||
line sees the type context at a glance. *dialect* is the migration
|
||||
connection's, so both sides are rendered the way that database spells
|
||||
them.
|
||||
"""
|
||||
diffs: list[str] = []
|
||||
|
||||
@ -137,13 +194,13 @@ def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None:
|
||||
if desired_nullable != actual_nullable:
|
||||
diffs.append(f"nullable actual={actual_nullable} desired={desired_nullable}")
|
||||
|
||||
desired_default = _normalize_default(desired.server_default)
|
||||
actual_default = _normalize_default(actual.get("default"))
|
||||
desired_default = _normalize_default(desired.server_default, dialect)
|
||||
actual_default = _normalize_default(actual.get("default"), dialect)
|
||||
if desired_default != actual_default:
|
||||
diffs.append(f"server_default actual={actual_default!r} desired={desired_default!r}")
|
||||
|
||||
if not _type_equivalent(actual.get("type"), desired.type):
|
||||
diffs.append(f"type actual={_normalize_type(actual.get('type'))!r} desired={_normalize_type(desired.type)!r}")
|
||||
if not _type_equivalent(actual.get("type"), desired.type, dialect):
|
||||
diffs.append(f"type actual={_normalize_type(actual.get('type'), dialect)!r} desired={_normalize_type(desired.type, dialect)!r}")
|
||||
|
||||
if diffs:
|
||||
logger.warning(
|
||||
@ -171,7 +228,7 @@ def safe_add_column(table: str, column: sa.Column) -> None:
|
||||
return
|
||||
existing = {c["name"]: c for c in insp.get_columns(table)}
|
||||
if column.name in existing:
|
||||
_check_column_drift(table, column, existing[column.name])
|
||||
_check_column_drift(table, column, existing[column.name], insp.dialect)
|
||||
return
|
||||
with op.batch_alter_table(table) as batch:
|
||||
batch.add_column(column)
|
||||
|
||||
@ -22,11 +22,14 @@ table-level backfill works.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
from support.postgres import asyncpg_test_url
|
||||
|
||||
# Pre-import models so Base.metadata is populated before bootstrap reads it.
|
||||
import deerflow.persistence.models # noqa: F401
|
||||
@ -42,6 +45,7 @@ from deerflow.persistence.bootstrap import (
|
||||
bootstrap_schema,
|
||||
)
|
||||
from deerflow.persistence.migrations._helpers import _normalize_default
|
||||
from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
|
||||
|
||||
# Mark only async tests via the decorator below; module-level pytestmark would
|
||||
# spuriously warn for the sync ``TestDecideState`` cases.
|
||||
@ -308,6 +312,8 @@ async def test_legacy_with_manual_workaround_column_warns_on_drift(
|
||||
# JSON ≈ JSON, so the equivalence check must NOT produce a "type" diff
|
||||
# entry here -- that would be a false positive on the matching-type case.
|
||||
assert "type actual=" not in msg, f"unexpected type drift on matching JSON column: {msg!r}"
|
||||
helper_warnings = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"]
|
||||
assert helper_warnings == [msg], f"expected only the token_usage_by_model drift, got: {helper_warnings!r}"
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
@ -348,10 +354,99 @@ async def test_legacy_with_wrong_type_workaround_warns_on_type_drift(
|
||||
# Nullability + server_default match the model -- no other diffs.
|
||||
assert "nullable" not in msg, f"unexpected nullability drift on matching column: {msg!r}"
|
||||
assert "server_default" not in msg, f"unexpected server_default drift on matching column: {msg!r}"
|
||||
# The seed is the full model schema minus this one column, so every
|
||||
# other column a revision finds already present matches the model and
|
||||
# the real drift must not be buried among false positives.
|
||||
helper_warnings = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"]
|
||||
assert helper_warnings == [msg], f"expected only the token_usage_by_model drift, got: {helper_warnings!r}"
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# No false positives: a column a revision finds already present in the model's
|
||||
# own shape must stay quiet on both dialects. The model spells types and
|
||||
# defaults generically (``sa.String``, ``sa.BigInteger``, ``sa.DateTime``,
|
||||
# ``server_default="0"``, ``sa.false()``) while reflection reports the
|
||||
# dialect's spelling (``VARCHAR``, ``BIGINT``, Postgres ``TIMESTAMP``, ``'0'``),
|
||||
# so the comparator must render the model side through the dialect first.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _drift_engine(tmp_path: Path, backend: str):
|
||||
if backend == "sqlite":
|
||||
return create_async_engine(_url(tmp_path)), ""
|
||||
uri = os.environ.get("TEST_POSTGRES_URI")
|
||||
if not uri:
|
||||
pytest.skip("requires TEST_POSTGRES_URI (real Postgres drift reflection)")
|
||||
schema = f"drift_{uuid.uuid4().hex}"
|
||||
engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema))
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text(f'CREATE SCHEMA "{schema}"'))
|
||||
return engine, schema
|
||||
|
||||
|
||||
def _seed_at_revision(revision: str, *, drop_version_row: bool):
|
||||
async def seed(engine, schema: str) -> None:
|
||||
await asyncio.to_thread(_upgrade, _get_alembic_config(engine, postgres_schema=schema), revision)
|
||||
if drop_version_row:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text("DROP TABLE alembic_version"))
|
||||
|
||||
return seed
|
||||
|
||||
|
||||
async def _seed_model_schema_without_version_row(engine, schema: str) -> None:
|
||||
# The empty branch commits ``create_all`` before it stamps head; a process
|
||||
# that dies in between leaves exactly this, and the next start takes the
|
||||
# legacy branch over columns every revision finds already present.
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
|
||||
async def _seed_model_schema_with_text_token_usage(engine, schema: str) -> None:
|
||||
await _seed_model_schema_without_version_row(engine, schema)
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text("ALTER TABLE runs DROP COLUMN token_usage_by_model"))
|
||||
await conn.execute(sa.text("ALTER TABLE runs ADD COLUMN token_usage_by_model TEXT NOT NULL DEFAULT '{}'"))
|
||||
|
||||
|
||||
@asyncio_test
|
||||
@pytest.mark.parametrize("backend", ["sqlite", "postgres"])
|
||||
@pytest.mark.parametrize(
|
||||
("seed", "expected_drift"),
|
||||
[
|
||||
pytest.param(_seed_at_revision(BASELINE, drop_version_row=True), [], id="pre_alembic_baseline"),
|
||||
# 0025 re-runs 0023's ``safe_add_column`` for ``runs.change_seq``.
|
||||
pytest.param(_seed_at_revision("0022_scheduled_occurrence_seq", drop_version_row=False), [], id="upgrade_across_change_seq_repair"),
|
||||
pytest.param(_seed_model_schema_without_version_row, [], id="model_schema_without_version_row"),
|
||||
pytest.param(_seed_model_schema_with_text_token_usage, ["runs.token_usage_by_model"], id="real_drift_only"),
|
||||
],
|
||||
)
|
||||
async def test_bootstrap_warns_only_on_columns_that_really_drift(
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
backend: str,
|
||||
seed,
|
||||
expected_drift: list[str],
|
||||
) -> None:
|
||||
engine, schema = await _drift_engine(tmp_path, backend)
|
||||
try:
|
||||
await seed(engine, schema)
|
||||
|
||||
with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"):
|
||||
await bootstrap_schema(engine, backend=backend, postgres_schema=schema)
|
||||
|
||||
assert await _alembic_version(engine) == HEAD
|
||||
warned = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"]
|
||||
assert [m.removeprefix("safe_add_column: ").split(" ", 1)[0] for m in warned] == expected_drift, f"unexpected drift warnings: {warned!r}"
|
||||
finally:
|
||||
if schema:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _type_equivalent unit tests: pin the JSON/JSONB equivalence so Postgres
|
||||
# dialect synonyms stay quiet, and pin the TEXT/JSON divergence so the
|
||||
@ -397,6 +492,101 @@ def test_type_equivalent_returns_true_on_missing_info() -> None:
|
||||
assert _type_equivalent("", "JSON") is True
|
||||
|
||||
|
||||
def test_type_equivalent_renders_model_types_through_the_dialect() -> None:
|
||||
"""The model's generic types meet reflection's dialect types on equal terms."""
|
||||
from sqlalchemy.dialects import postgresql, sqlite
|
||||
|
||||
from deerflow.persistence.migrations._helpers import _type_equivalent
|
||||
|
||||
sqlite_dialect = sqlite.dialect()
|
||||
postgres_dialect = postgresql.dialect()
|
||||
# Reflected (actual) on the left, model (desired) on the right.
|
||||
assert _type_equivalent(sa.VARCHAR(length=128), sa.String(length=128), sqlite_dialect) is True
|
||||
assert _type_equivalent(sa.BIGINT(), sa.BigInteger(), sqlite_dialect) is True
|
||||
assert _type_equivalent(sqlite.DATETIME(), sa.DateTime(timezone=True), sqlite_dialect) is True
|
||||
assert _type_equivalent(sa.VARCHAR(length=64), sa.String(length=64), postgres_dialect) is True
|
||||
assert _type_equivalent(sa.BIGINT(), sa.BigInteger(), postgres_dialect) is True
|
||||
assert _type_equivalent(postgresql.TIMESTAMP(timezone=True), sa.DateTime(timezone=True), postgres_dialect) is True
|
||||
assert _type_equivalent(postgresql.JSONB(), sa.JSON(), postgres_dialect) is True
|
||||
# Precision is a parameter like length; the time-zone suffix is not.
|
||||
assert _type_equivalent(postgresql.TIMESTAMP(timezone=True, precision=3), sa.DateTime(timezone=True), postgres_dialect) is True
|
||||
assert _type_equivalent(postgresql.TIMESTAMP(timezone=False, precision=3), sa.DateTime(timezone=True), postgres_dialect) is False
|
||||
# Rendering both sides must not hide a real mismatch.
|
||||
assert _type_equivalent(sa.TEXT(), sa.JSON(), sqlite_dialect) is False
|
||||
assert _type_equivalent(sa.TEXT(), sa.JSON(), postgres_dialect) is False
|
||||
assert _type_equivalent(postgresql.TIMESTAMP(timezone=False), sa.DateTime(timezone=True), postgres_dialect) is False
|
||||
assert _type_equivalent(sa.INTEGER(), sa.BigInteger(), postgres_dialect) is False
|
||||
# SQLite reflects an unrecognized declared type as NullType, which no
|
||||
# dialect can render: missing info, not drift.
|
||||
assert _type_equivalent(sa.types.NullType(), sa.String(length=64), sqlite_dialect) is True
|
||||
|
||||
|
||||
def test_normalize_default_renders_model_defaults_like_reflection() -> None:
|
||||
"""``server_default="0"`` / ``sa.false()`` compare equal to how each dialect reflects them."""
|
||||
from sqlalchemy.dialects import postgresql, sqlite
|
||||
|
||||
sqlite_dialect = sqlite.dialect()
|
||||
postgres_dialect = postgresql.dialect()
|
||||
string_zero = sa.Column("c", sa.Integer(), server_default="0").server_default
|
||||
false_default = sa.Column("c", sa.Boolean(), server_default=sa.false()).server_default
|
||||
|
||||
# SQLite reflects ``DEFAULT '0'`` quoted; Postgres reflects it bare on an
|
||||
# integer column and as ``'0'::bigint`` on a bigint one.
|
||||
assert _normalize_default(string_zero, sqlite_dialect) == _normalize_default("'0'", sqlite_dialect) == "0"
|
||||
assert _normalize_default(string_zero, postgres_dialect) == _normalize_default("0", postgres_dialect) == "0"
|
||||
assert _normalize_default("'0'::bigint", postgres_dialect) == "0"
|
||||
assert _normalize_default(false_default, sqlite_dialect) == _normalize_default("0", sqlite_dialect)
|
||||
assert _normalize_default(false_default, postgres_dialect) == _normalize_default("false", postgres_dialect)
|
||||
assert _normalize_default(sa.text("'{}'")) == _normalize_default("'{}'::jsonb") == "{}"
|
||||
# A different default still differs.
|
||||
assert _normalize_default(string_zero, sqlite_dialect) != _normalize_default("'1'", sqlite_dialect)
|
||||
assert _normalize_default(false_default, postgres_dialect) != _normalize_default("true", postgres_dialect)
|
||||
|
||||
# An explicit empty-string default is a default, not the absence of one.
|
||||
empty_string = sa.Column("c", sa.Text(), server_default="").server_default
|
||||
assert _normalize_default(empty_string, sqlite_dialect) == _normalize_default("''", sqlite_dialect) == ""
|
||||
assert _normalize_default("''::text", postgres_dialect) == ""
|
||||
assert _normalize_default(None) is None
|
||||
|
||||
|
||||
@asyncio_test
|
||||
@pytest.mark.parametrize("backend", ["sqlite", "postgres"])
|
||||
async def test_empty_string_default_stays_distinct_from_no_default(
|
||||
tmp_path: Path,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
backend: str,
|
||||
) -> None:
|
||||
"""``DEFAULT ''`` and no default drift from each other, through real reflection."""
|
||||
from deerflow.persistence.migrations._helpers import _check_column_drift
|
||||
|
||||
engine, schema = await _drift_engine(tmp_path, backend)
|
||||
try:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text("CREATE TABLE drift_defaults (empty_default TEXT DEFAULT '', no_default TEXT)"))
|
||||
|
||||
def check(sync_conn) -> None:
|
||||
inspector = sa.inspect(sync_conn)
|
||||
columns = {column["name"]: column for column in inspector.get_columns("drift_defaults")}
|
||||
_check_column_drift("drift_defaults", sa.Column("no_default", sa.Text(), server_default=""), columns["no_default"], inspector.dialect)
|
||||
_check_column_drift("drift_defaults", sa.Column("empty_default", sa.Text()), columns["empty_default"], inspector.dialect)
|
||||
# Matching empty-string defaults stay quiet.
|
||||
_check_column_drift("drift_defaults", sa.Column("empty_default", sa.Text(), server_default=""), columns["empty_default"], inspector.dialect)
|
||||
|
||||
with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"):
|
||||
async with engine.connect() as conn:
|
||||
await conn.run_sync(check)
|
||||
|
||||
warned = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"]
|
||||
assert len(warned) == 2, f"expected exactly the two mismatched defaults, got: {warned!r}"
|
||||
assert "drift_defaults.no_default" in warned[0] and "server_default actual=None desired=''" in warned[0], warned[0]
|
||||
assert "drift_defaults.empty_default" in warned[1] and "server_default actual='' desired=None" in warned[1], warned[1]
|
||||
finally:
|
||||
if schema:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Branch 4: versioned DB
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user