docs: 统一业务资源ID为服务端生成的稳定UUID
更新所有业务API文档,明确持久化资源的正式ID必须为服务端生成的稳定UUID,本地调试或接口失败时可使用语义ID作为fallback。新增数据库迁移脚本0022_opaque_ids,用于将历史语义ID转换为稳定UUID,并同步外键关联、详情记录的key字段以及审计日志的实体ID引用。新增该迁移的单元测试用例,验证ID替换与关联数据同步的逻辑正确性。调整MiniAPP前端代码,优化导航工具函数的格式,移除废弃函数并修改首页跳转逻辑,使用接口返回的UUID作为详情跳转参数。
This commit is contained in:
1 parent
e082bd2d98
commit
d3a246a873
14 files changed
+314
-24
No files matched your search
@@ -0,0 +1,210 @@
|
||||
"""Replace historical semantic IDs with stable opaque UUID strings.
|
||||
|
||||
Revision ID: 0022_opaque_ids
|
||||
Revises: 0021_detail_records
|
||||
|
||||
New records already use ``uuid4`` in the ORM. This migration only repairs
|
||||
historical rows created by seed migrations and keeps all IDs stable after the
|
||||
conversion. Generating a new ID during serialization would break detail URLs,
|
||||
reorder requests and foreign-key relationships, so this migration is
|
||||
intentionally data-oriented and cannot be reversed automatically.
|
||||
"""
|
||||
|
||||
from collections.abc import Iterable
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from alembic import context, op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import inspect
|
||||
|
||||
|
||||
revision = "0022_opaque_ids"
|
||||
down_revision = "0021_detail_records"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
AUDIT_ENTITY_TABLES = {
|
||||
"hero_slide": "HeroSlide",
|
||||
"destination_hero": "DestinationHero",
|
||||
"demand_hero": "DemandHero",
|
||||
"demand_feature_card": "DemandFeatureCard",
|
||||
"vehicle_option": "VehicleOption",
|
||||
"home_experience": "HomeExperience",
|
||||
"home_team_building": "HomeTeamBuilding",
|
||||
"home_wild_archive": "HomeWildArchive",
|
||||
"wanfa_category": "WanfaCategory",
|
||||
"wanfa_route": "WanfaRoute",
|
||||
"detail": "DetailRecord",
|
||||
"concierge_advisor": "ConciergeAdvisor",
|
||||
"lead": "Lead",
|
||||
}
|
||||
|
||||
|
||||
def is_opaque_id(value: object) -> bool:
|
||||
"""Return whether a persisted ID is already a UUID-like opaque value."""
|
||||
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return False
|
||||
try:
|
||||
UUID(value.strip())
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def build_id_mapping(ids: Iterable[object]) -> dict[str, str]:
|
||||
"""Build a one-time semantic-to-UUID mapping without touching UUID IDs."""
|
||||
|
||||
mapping: dict[str, str] = {}
|
||||
for raw_id in ids:
|
||||
if raw_id is None:
|
||||
continue
|
||||
old_id = str(raw_id).strip()
|
||||
if old_id and not is_opaque_id(old_id) and old_id not in mapping:
|
||||
mapping[old_id] = str(uuid4())
|
||||
return mapping
|
||||
|
||||
|
||||
def _quote_identifier(identifier: str) -> str:
|
||||
return '"' + identifier.replace('"', '""') + '"'
|
||||
|
||||
|
||||
def _replace_values(conn, table_name: str, column_name: str, mapping: dict[str, str]) -> None:
|
||||
if not mapping:
|
||||
return
|
||||
table = _quote_identifier(table_name)
|
||||
column = _quote_identifier(column_name)
|
||||
statement = sa.text(f"UPDATE {table} SET {column} = :new_id WHERE {column} = :old_id")
|
||||
for old_id, new_id in mapping.items():
|
||||
conn.execute(statement, {"old_id": old_id, "new_id": new_id})
|
||||
|
||||
|
||||
def _id_tables(conn) -> list[str]:
|
||||
inspector = inspect(conn)
|
||||
return [
|
||||
table_name
|
||||
for table_name in inspector.get_table_names()
|
||||
if table_name != "alembic_version"
|
||||
and any(column["name"] == "id" for column in inspector.get_columns(table_name))
|
||||
]
|
||||
|
||||
|
||||
def _id_mappings(conn, table_names: list[str]) -> dict[str, dict[str, str]]:
|
||||
mappings: dict[str, dict[str, str]] = {}
|
||||
for table_name in table_names:
|
||||
rows = conn.execute(
|
||||
sa.text(f"SELECT {_quote_identifier('id')} FROM {_quote_identifier(table_name)}")
|
||||
).scalars()
|
||||
mapping = build_id_mapping(rows)
|
||||
if mapping:
|
||||
mappings[table_name] = mapping
|
||||
return mappings
|
||||
|
||||
|
||||
def collect_fk_specs(foreign_keys, mappings: dict[str, dict[str, str]]) -> list[dict]:
|
||||
"""Keep enough FK metadata to drop, rewrite and recreate each constraint."""
|
||||
|
||||
specs: list[dict] = []
|
||||
for source_table, foreign_key in foreign_keys:
|
||||
target_table = foreign_key.get("referred_table")
|
||||
target_columns = foreign_key.get("referred_columns") or []
|
||||
source_columns = foreign_key.get("constrained_columns") or []
|
||||
if not target_table or target_table not in mappings or "id" not in target_columns:
|
||||
continue
|
||||
options = foreign_key.get("options") or {}
|
||||
specs.append(
|
||||
{
|
||||
"name": foreign_key.get("name"),
|
||||
"source_table": source_table,
|
||||
"source_columns": source_columns,
|
||||
"target_table": target_table,
|
||||
"target_columns": target_columns,
|
||||
"ondelete": options.get("ondelete"),
|
||||
"onupdate": options.get("onupdate"),
|
||||
}
|
||||
)
|
||||
return specs
|
||||
|
||||
|
||||
def _foreign_key_specs(conn, table_names: list[str], mappings: dict[str, dict[str, str]]) -> list[dict]:
|
||||
inspector = inspect(conn)
|
||||
foreign_keys = [
|
||||
(table_name, foreign_key)
|
||||
for table_name in table_names
|
||||
for foreign_key in inspector.get_foreign_keys(table_name)
|
||||
]
|
||||
return collect_fk_specs(foreign_keys, mappings)
|
||||
|
||||
|
||||
def _drop_foreign_keys(specs: list[dict]) -> None:
|
||||
for spec in specs:
|
||||
if spec["name"]:
|
||||
op.drop_constraint(spec["name"], spec["source_table"], type_="foreignkey")
|
||||
|
||||
|
||||
def _replace_foreign_keys(conn, specs: list[dict], mappings: dict[str, dict[str, str]]) -> None:
|
||||
for spec in specs:
|
||||
target_mapping = mappings[spec["target_table"]]
|
||||
for source_column, target_column in zip(spec["source_columns"], spec["target_columns"]):
|
||||
if target_column == "id":
|
||||
_replace_values(conn, spec["source_table"], source_column, target_mapping)
|
||||
|
||||
|
||||
def _restore_foreign_keys(specs: list[dict]) -> None:
|
||||
for spec in specs:
|
||||
if not spec["name"]:
|
||||
continue
|
||||
op.create_foreign_key(
|
||||
spec["name"],
|
||||
spec["source_table"],
|
||||
spec["target_table"],
|
||||
spec["source_columns"],
|
||||
spec["target_columns"],
|
||||
ondelete=spec["ondelete"],
|
||||
onupdate=spec["onupdate"],
|
||||
)
|
||||
|
||||
|
||||
def _replace_detail_route_keys(conn, table_names: list[str], mappings: dict[str, dict[str, str]]) -> None:
|
||||
if "DetailRecord" not in table_names:
|
||||
return
|
||||
_replace_values(conn, "DetailRecord", "key", mappings.get("WanfaRoute", {}))
|
||||
|
||||
|
||||
def _replace_audit_entity_ids(conn, mappings: dict[str, dict[str, str]]) -> None:
|
||||
if not mappings.get("AuditLog") and "AuditLog" not in _id_tables(conn):
|
||||
return
|
||||
for entity, table_name in AUDIT_ENTITY_TABLES.items():
|
||||
_replace_values(conn, "AuditLog", "entityId", mappings.get(table_name, {}))
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
if context.is_offline_mode():
|
||||
return
|
||||
|
||||
conn = op.get_bind()
|
||||
table_names = _id_tables(conn)
|
||||
mappings = _id_mappings(conn, table_names)
|
||||
if not mappings:
|
||||
return
|
||||
|
||||
# PostgreSQL checks existing foreign keys immediately. Temporarily remove
|
||||
# affected constraints, rewrite all IDs and references in one transaction,
|
||||
# then recreate the same constraints with their original actions.
|
||||
foreign_key_specs = _foreign_key_specs(conn, table_names, mappings)
|
||||
_drop_foreign_keys(foreign_key_specs)
|
||||
_replace_foreign_keys(conn, foreign_key_specs, mappings)
|
||||
_replace_detail_route_keys(conn, table_names, mappings)
|
||||
_replace_audit_entity_ids(conn, mappings)
|
||||
|
||||
for table_name, mapping in mappings.items():
|
||||
_replace_values(conn, table_name, "id", mapping)
|
||||
|
||||
_restore_foreign_keys(foreign_key_specs)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# UUID replacement is intentionally one-way: the original semantic IDs
|
||||
# are not retained in the database and cannot be reconstructed safely.
|
||||
pass
|
||||
@@ -0,0 +1,60 @@
|
||||
from importlib.util import module_from_spec, spec_from_file_location
|
||||
from pathlib import Path
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
|
||||
MIGRATION_PATH = Path(__file__).parents[1] / "alembic" / "versions" / "0022_opaque_ids.py"
|
||||
SPEC = spec_from_file_location("opaque_ids_migration", MIGRATION_PATH)
|
||||
assert SPEC and SPEC.loader
|
||||
MIGRATION = module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(MIGRATION)
|
||||
|
||||
|
||||
def test_existing_semantic_ids_are_replaced_with_stable_uuid_values():
|
||||
existing_uuid = str(uuid4())
|
||||
|
||||
mapping = MIGRATION.build_id_mapping(["family-route", existing_uuid])
|
||||
|
||||
assert set(mapping) == {"family-route"}
|
||||
UUID(mapping["family-route"])
|
||||
assert mapping["family-route"] != "family-route"
|
||||
|
||||
|
||||
def test_uuid_values_are_not_regenerated_during_migration():
|
||||
existing_uuid = str(uuid4())
|
||||
|
||||
assert MIGRATION.is_opaque_id(existing_uuid)
|
||||
assert not MIGRATION.is_opaque_id("family-route")
|
||||
assert MIGRATION.build_id_mapping([existing_uuid]) == {}
|
||||
|
||||
|
||||
def test_foreign_key_specs_capture_constraints_for_recreation():
|
||||
foreign_keys = [
|
||||
(
|
||||
"WanfaRoute",
|
||||
{
|
||||
"name": "WanfaRoute_categoryId_fkey",
|
||||
"constrained_columns": ["categoryId"],
|
||||
"referred_table": "WanfaCategory",
|
||||
"referred_columns": ["id"],
|
||||
"options": {"ondelete": "RESTRICT"},
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
specs = MIGRATION.collect_fk_specs(
|
||||
foreign_keys,
|
||||
{"WanfaCategory": {"family-route": "category-uuid"}},
|
||||
)
|
||||
|
||||
assert specs == [
|
||||
{
|
||||
"name": "WanfaRoute_categoryId_fkey",
|
||||
"source_table": "WanfaRoute",
|
||||
"source_columns": ["categoryId"],
|
||||
"target_table": "WanfaCategory",
|
||||
"target_columns": ["id"],
|
||||
"ondelete": "RESTRICT",
|
||||
"onupdate": None,
|
||||
}
|
||||
]
|
||||
Reference in new issue
Block a user