i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: 迁移文件内联 helpers，删除共享 alembic/helpers.py

Browse Source 
将 _SchemaCache、replace_fk_if_needed、batch_alter_type 等辅助函数
内联到各迁移文件中，使每个迁移自包含、不依赖外部模块。
同时修正 a3f1b7c9d2e4 的 down_revision 为 d7649c1f8e21。
This commit is contained in:
fawney19
2026-03-10 10:53:41 +08:00
parent 7c580e843f
commit 86449cae52
7 changed files with 415 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
alembic/env.py
View File

@@ -14,8 +14,6 @@ from alembic import context
# 添加项目根目录到 Python 路径
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
# 添加 alembic 目录，让迁移文件可以 `from helpers import ...`
sys.path.insert(0, os.path.dirname(__file__))
# 加载 .env 文件（本地开发时需要）
try:

220
alembic/helpers.py
View File

@@ -1,220 +0,0 @@
"""Shared idempotent helpers for Alembic migrations.
All metadata lookups are batched: one query loads an entire table's column info
or all FK delete rules, then results are cached for the migration's lifetime.
"""
from __future__ import annotations
from collections import defaultdict
from collections.abc import Callable
import sqlalchemy as sa
from alembic import op
# ---------------------------------------------------------------------------
# Batch metadata cache
# ---------------------------------------------------------------------------
class _SchemaCache:
"""Lazy, per-migration cache for information_schema lookups.
Call ``load_columns(tables)`` / ``load_fk_rules(tables)`` once at the top
of ``upgrade()`` or ``downgrade()`` to prime the cache. Subsequent
``column_exists`` / ``column_type`` / ``fk_ondelete`` calls are pure
dict lookups -- zero extra DB round-trips.
"""
def __init__(self) -> None:
# {table_name: {column_name: data_type}}
self._columns: dict[str, dict[str, str]] = {}
# {(table_name, constraint_name): delete_rule}
self._fk_rules: dict[tuple[str, str], str] = {}
self._fk_loaded_tables: set[str] = set()
# -- loaders (one query per call) --------------------------------------
def load_columns(self, tables: list[str]) -> None:
"""Fetch column names + data types for *tables* in one query."""
need = [t for t in tables if t not in self._columns]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT table_name, column_name, data_type "
"FROM information_schema.columns "
"WHERE table_name = ANY(:tables) "
" AND table_schema = current_schema()"
),
{"tables": need},
).fetchall()
# Initialise even empty tables so we don't re-query
for t in need:
self._columns.setdefault(t, {})
for table, col, dtype in rows:
self._columns[table][col] = dtype
def load_fk_rules(self, tables: list[str]) -> None:
"""Fetch FK delete rules for *tables* in one query."""
need = [t for t in tables if t not in self._fk_loaded_tables]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT tc.table_name, tc.constraint_name, rc.delete_rule "
"FROM information_schema.referential_constraints rc "
"JOIN information_schema.table_constraints tc "
" ON rc.constraint_name = tc.constraint_name "
" AND rc.constraint_schema = tc.constraint_schema "
"WHERE tc.table_name = ANY(:tables) "
" AND tc.table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for table, name, rule in rows:
self._fk_rules[(table, name)] = rule
self._fk_loaded_tables.update(need)
# -- lookups (pure dict, zero DB) --------------------------------------
def column_exists(self, table: str, column: str) -> bool:
return column in self._columns.get(table, {})
def column_type(self, table: str, column: str) -> str | None:
return self._columns.get(table, {}).get(column)
def is_numeric(self, table: str, column: str) -> bool:
return self.column_type(table, column) == "numeric"
def fk_ondelete(self, table: str, constraint: str) -> str | None:
return self._fk_rules.get((table, constraint))
def invalidate_columns(self, table: str) -> None:
"""Force re-load on next load_columns() for *table* (after ADD/DROP COLUMN)."""
self._columns.pop(table, None)
def new_cache() -> _SchemaCache:
"""Create a fresh schema cache for a single migration run."""
return _SchemaCache()
# ---------------------------------------------------------------------------
# Idempotent DDL helpers
# ---------------------------------------------------------------------------
def _fk_exists(constraint_name: str, table_name: str) -> bool:
"""Check if a FK constraint exists via pg_constraint (definitive)."""
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_constraint c "
"JOIN pg_class r ON c.conrelid = r.oid "
"JOIN pg_namespace n ON r.relnamespace = n.oid "
"WHERE c.conname = :name AND r.relname = :table "
" AND n.nspname = current_schema() AND c.contype = 'f'"
),
{"name": constraint_name, "table": table_name},
)
return result.scalar() is not None
def replace_fk_if_needed(
cache: _SchemaCache,
constraint_name: str,
table_name: str,
ref_table: str,
local_cols: list[str],
remote_cols: list[str],
desired_ondelete: str,
) -> None:
"""Drop and recreate a FK only if the current ON DELETE rule differs."""
current = cache.fk_ondelete(table_name, constraint_name)
if current and current.upper() == desired_ondelete.upper():
return
# Cache may miss existing constraints; fall back to pg_constraint lookup
if current or _fk_exists(constraint_name, table_name):
op.drop_constraint(constraint_name, table_name, type_="foreignkey")
op.create_foreign_key(
constraint_name,
table_name,
ref_table,
local_cols,
remote_cols,
ondelete=desired_ondelete,
)
def index_exists(index_name: str) -> bool:
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_indexes "
"WHERE indexname = :name AND schemaname = current_schema()::text"
),
{"name": index_name},
)
return result.scalar() is not None
# ---------------------------------------------------------------------------
# Batch ALTER TYPE helper
# ---------------------------------------------------------------------------
def _numeric_max(type_spec: str) -> float | None:
"""Parse NUMERIC(p,s) and return the maximum absolute value, or None."""
# e.g. "NUMERIC(20,8)" -> precision=20, scale=8 -> max = 10^(20-8) - 10^(-8)
import re
m = re.match(r"NUMERIC\((\d+),(\d+)\)", type_spec, re.IGNORECASE)
if not m:
return None
precision, scale = int(m.group(1)), int(m.group(2))
return 10 ** (precision - scale) - 10 ** (-scale)
def batch_alter_type(
cache: _SchemaCache,
columns: list[tuple[str, str, bool, str | None]],
cast_suffix: str,
type_fn: Callable[[str], str],
) -> None:
"""Group columns by table and issue ONE ``ALTER TABLE`` per table.
Skips columns that don't exist. Each tuple is
``(table_name, column_name, nullable, server_default)``.
When converting to NUMERIC(p,s), values exceeding the target precision
are clamped before the type cast to prevent overflow errors.
"""
by_table: dict[str, list[tuple[str, str]]] = defaultdict(list)
for table, col, _nullable, _default in columns:
if not cache.column_exists(table, col):
continue
by_table[table].append((col, type_fn(col)))
bind = op.get_bind()
for table, col_types in by_table.items():
# Clamp out-of-range values before ALTER TYPE
for col, target in col_types:
cap = _numeric_max(target)
if cap is not None:
bind.execute(
sa.text(
f"UPDATE {table} SET {col} = :cap "
f"WHERE {col} IS NOT NULL AND abs({col}) > :cap"
),
{"cap": cap},
)
parts = [
f"ALTER COLUMN {col} TYPE {target} USING {col}::{cast_suffix}"
for col, target in col_types
]
if parts:
bind.execute(sa.text(f"ALTER TABLE {table} " + ", ".join(parts)))

115
alembic/versions/20260308_0348_45b118150a78_usage_stats_retention_set_null_on_.py
View File

@@ -6,8 +6,9 @@ Create Date: 2026-03-08 03:48:49.622091+00:00
"""
from __future__ import annotations
import sqlalchemy as sa
from helpers import new_cache, replace_fk_if_needed
from alembic import op
@@ -20,8 +21,108 @@ depends_on = None
_TABLES = ["usage", "stats_user_daily", "stats_daily_api_key"]
# ---------------------------------------------------------------------------
# Inline helpers
# ---------------------------------------------------------------------------
class _SchemaCache:
def __init__(self) -> None:
self._columns: dict[str, dict[str, str]] = {}
self._fk_rules: dict[tuple[str, str], str] = {}
self._fk_loaded_tables: set[str] = set()
def load_columns(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._columns]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT table_name, column_name, data_type "
"FROM information_schema.columns "
"WHERE table_name = ANY(:tables) "
" AND table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for t in need:
self._columns.setdefault(t, {})
for table, col, dtype in rows:
self._columns[table][col] = dtype
def load_fk_rules(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._fk_loaded_tables]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT tc.table_name, tc.constraint_name, rc.delete_rule "
"FROM information_schema.referential_constraints rc "
"JOIN information_schema.table_constraints tc "
" ON rc.constraint_name = tc.constraint_name "
" AND rc.constraint_schema = tc.constraint_schema "
"WHERE tc.table_name = ANY(:tables) "
" AND tc.table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for table, name, rule in rows:
self._fk_rules[(table, name)] = rule
self._fk_loaded_tables.update(need)
def column_exists(self, table: str, column: str) -> bool:
return column in self._columns.get(table, {})
def fk_ondelete(self, table: str, constraint: str) -> str | None:
return self._fk_rules.get((table, constraint))
def _fk_exists(constraint_name: str, table_name: str) -> bool:
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_constraint c "
"JOIN pg_class r ON c.conrelid = r.oid "
"JOIN pg_namespace n ON r.relnamespace = n.oid "
"WHERE c.conname = :name AND r.relname = :table "
" AND n.nspname = current_schema() AND c.contype = 'f'"
),
{"name": constraint_name, "table": table_name},
)
return result.scalar() is not None
def _replace_fk_if_needed(
cache: _SchemaCache,
constraint_name: str,
table_name: str,
ref_table: str,
local_cols: list[str],
remote_cols: list[str],
desired_ondelete: str,
) -> None:
current = cache.fk_ondelete(table_name, constraint_name)
if current and current.upper() == desired_ondelete.upper():
return
if current or _fk_exists(constraint_name, table_name):
op.drop_constraint(constraint_name, table_name, type_="foreignkey")
op.create_foreign_key(
constraint_name,
table_name,
ref_table,
local_cols,
remote_cols,
ondelete=desired_ondelete,
)
# ---------------------------------------------------------------------------
def upgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_columns(_TABLES)
c.load_fk_rules(["stats_user_daily", "stats_daily_api_key"])
@@ -37,7 +138,7 @@ def upgrade() -> None:
)
# --- StatsUserDaily: CASCADE -> SET NULL, add username snapshot ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"stats_user_daily_user_id_fkey",
"stats_user_daily",
@@ -59,7 +160,7 @@ def upgrade() -> None:
)
# --- StatsDailyApiKey: CASCADE -> SET NULL, add api_key_name snapshot ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"stats_daily_api_key_api_key_id_fkey",
"stats_daily_api_key",
@@ -82,7 +183,7 @@ def upgrade() -> None:
def downgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_columns(["stats_daily_api_key", "stats_user_daily", "usage"])
c.load_fk_rules(["stats_daily_api_key", "stats_user_daily"])
@@ -97,7 +198,7 @@ def downgrade() -> None:
op.drop_column("usage", "username")
# --- StatsDailyApiKey: SET NULL -> CASCADE ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"stats_daily_api_key_api_key_id_fkey",
"stats_daily_api_key",
@@ -111,7 +212,7 @@ def downgrade() -> None:
)
# --- StatsUserDaily: SET NULL -> CASCADE ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"stats_user_daily_user_id_fkey",
"stats_user_daily",

123
alembic/versions/20260308_1215_13a4c8f6d9e0_request_candidate_video_task_retention.py
View File

@@ -6,8 +6,9 @@ Create Date: 2026-03-08 12:15:00.000000+00:00
"""
from __future__ import annotations
import sqlalchemy as sa
from helpers import new_cache, replace_fk_if_needed
from alembic import op
@@ -20,8 +21,108 @@ depends_on = None
_TABLES = ["request_candidates", "video_tasks"]
# ---------------------------------------------------------------------------
# Inline helpers
# ---------------------------------------------------------------------------
class _SchemaCache:
def __init__(self) -> None:
self._columns: dict[str, dict[str, str]] = {}
self._fk_rules: dict[tuple[str, str], str] = {}
self._fk_loaded_tables: set[str] = set()
def load_columns(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._columns]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT table_name, column_name, data_type "
"FROM information_schema.columns "
"WHERE table_name = ANY(:tables) "
" AND table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for t in need:
self._columns.setdefault(t, {})
for table, col, dtype in rows:
self._columns[table][col] = dtype
def load_fk_rules(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._fk_loaded_tables]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT tc.table_name, tc.constraint_name, rc.delete_rule "
"FROM information_schema.referential_constraints rc "
"JOIN information_schema.table_constraints tc "
" ON rc.constraint_name = tc.constraint_name "
" AND rc.constraint_schema = tc.constraint_schema "
"WHERE tc.table_name = ANY(:tables) "
" AND tc.table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for table, name, rule in rows:
self._fk_rules[(table, name)] = rule
self._fk_loaded_tables.update(need)
def column_exists(self, table: str, column: str) -> bool:
return column in self._columns.get(table, {})
def fk_ondelete(self, table: str, constraint: str) -> str | None:
return self._fk_rules.get((table, constraint))
def _fk_exists(constraint_name: str, table_name: str) -> bool:
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_constraint c "
"JOIN pg_class r ON c.conrelid = r.oid "
"JOIN pg_namespace n ON r.relnamespace = n.oid "
"WHERE c.conname = :name AND r.relname = :table "
" AND n.nspname = current_schema() AND c.contype = 'f'"
),
{"name": constraint_name, "table": table_name},
)
return result.scalar() is not None
def _replace_fk_if_needed(
cache: _SchemaCache,
constraint_name: str,
table_name: str,
ref_table: str,
local_cols: list[str],
remote_cols: list[str],
desired_ondelete: str,
) -> None:
current = cache.fk_ondelete(table_name, constraint_name)
if current and current.upper() == desired_ondelete.upper():
return
if current or _fk_exists(constraint_name, table_name):
op.drop_constraint(constraint_name, table_name, type_="foreignkey")
op.create_foreign_key(
constraint_name,
table_name,
ref_table,
local_cols,
remote_cols,
ondelete=desired_ondelete,
)
# ---------------------------------------------------------------------------
def upgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_columns(_TABLES)
c.load_fk_rules(_TABLES)
@@ -43,7 +144,7 @@ def upgrade() -> None:
)
# --- request_candidates: CASCADE -> SET NULL ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"request_candidates_user_id_fkey",
"request_candidates",
@@ -52,7 +153,7 @@ def upgrade() -> None:
["id"],
"SET NULL",
)
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"request_candidates_api_key_id_fkey",
"request_candidates",
@@ -81,7 +182,7 @@ def upgrade() -> None:
# --- video_tasks: CASCADE -> SET NULL, user_id nullable ---
op.alter_column("video_tasks", "user_id", existing_type=sa.String(length=36), nullable=True)
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"video_tasks_user_id_fkey",
"video_tasks",
@@ -90,7 +191,7 @@ def upgrade() -> None:
["id"],
"SET NULL",
)
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"video_tasks_api_key_id_fkey",
"video_tasks",
@@ -102,12 +203,12 @@ def upgrade() -> None:
def downgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_columns(_TABLES)
c.load_fk_rules(_TABLES)
# --- video_tasks: SET NULL -> default (no action), restore NOT NULL ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"video_tasks_api_key_id_fkey",
"video_tasks",
@@ -116,7 +217,7 @@ def downgrade() -> None:
["id"],
"NO ACTION",
)
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"video_tasks_user_id_fkey",
"video_tasks",
@@ -132,7 +233,7 @@ def downgrade() -> None:
op.drop_column("video_tasks", "username")
# --- request_candidates: SET NULL -> CASCADE ---
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"request_candidates_api_key_id_fkey",
"request_candidates",
@@ -141,7 +242,7 @@ def downgrade() -> None:
["id"],
"CASCADE",
)
replace_fk_if_needed(
_replace_fk_if_needed(
c,
"request_candidates_user_id_fkey",
"request_candidates",

114
alembic/versions/20260308_1530_2053ab8ed764_cost_fields_float_to_numeric.py
View File

@@ -6,8 +6,15 @@ Create Date: 2026-03-08 15:30:00.000000+00:00
"""
from __future__ import annotations
import re
from collections import defaultdict
from collections.abc import Callable
import sqlalchemy as sa
from alembic import op
from helpers import batch_alter_type, index_exists, new_cache
# revision identifiers, used by Alembic.
revision = "2053ab8ed764"
@@ -79,13 +86,106 @@ _COST_COLUMNS: list[tuple[str, str, bool, str | None]] = [
_ALL_TABLES = list({t for t, *_ in _COST_COLUMNS})
# ---------------------------------------------------------------------------
# Inline helpers
# ---------------------------------------------------------------------------
class _SchemaCache:
def __init__(self) -> None:
self._columns: dict[str, dict[str, str]] = {}
def load_columns(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._columns]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT table_name, column_name, data_type "
"FROM information_schema.columns "
"WHERE table_name = ANY(:tables) "
" AND table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for t in need:
self._columns.setdefault(t, {})
for table, col, dtype in rows:
self._columns[table][col] = dtype
def column_exists(self, table: str, column: str) -> bool:
return column in self._columns.get(table, {})
def column_type(self, table: str, column: str) -> str | None:
return self._columns.get(table, {}).get(column)
def is_numeric(self, table: str, column: str) -> bool:
return self.column_type(table, column) == "numeric"
def _index_exists(index_name: str) -> bool:
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_indexes "
"WHERE indexname = :name AND schemaname = current_schema()::text"
),
{"name": index_name},
)
return result.scalar() is not None
def _numeric_max(type_spec: str) -> float | None:
m = re.match(r"NUMERIC\((\d+),(\d+)\)", type_spec, re.IGNORECASE)
if not m:
return None
precision, scale = int(m.group(1)), int(m.group(2))
return 10 ** (precision - scale) - 10 ** (-scale)
def _batch_alter_type(
cache: _SchemaCache,
columns: list[tuple[str, str, bool, str | None]],
cast_suffix: str,
type_fn: Callable[[str], str],
) -> None:
by_table: dict[str, list[tuple[str, str]]] = defaultdict(list)
for table, col, _nullable, _default in columns:
if not cache.column_exists(table, col):
continue
by_table[table].append((col, type_fn(col)))
bind = op.get_bind()
for table, col_types in by_table.items():
for col, target in col_types:
cap = _numeric_max(target)
if cap is not None:
bind.execute(
sa.text(
f"UPDATE {table} SET {col} = :cap "
f"WHERE {col} IS NOT NULL AND abs({col}) > :cap"
),
{"cap": cap},
)
parts = [
f"ALTER COLUMN {col} TYPE {target} USING {col}::{cast_suffix}"
for col, target in col_types
]
if parts:
bind.execute(sa.text(f"ALTER TABLE {table} " + ", ".join(parts)))
# ---------------------------------------------------------------------------
def _type_spec(col: str) -> str:
"""Return the SQL type literal for a given column name."""
return "NUMERIC(10,6)" if col == "rate_multiplier" else "NUMERIC(20,8)"
def upgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_columns(_ALL_TABLES)
# -- 1. cost fields: Float -> Numeric (batched per table)
@@ -94,10 +194,10 @@ def upgrade() -> None:
for t, col, n, d in _COST_COLUMNS
if c.column_exists(t, col) and not c.is_numeric(t, col)
]
batch_alter_type(c, cols_to_convert, cast_suffix="numeric", type_fn=_type_spec)
_batch_alter_type(c, cols_to_convert, cast_suffix="numeric", type_fn=_type_spec)
# -- 2. provider_api_keys composite index
if not index_exists("idx_provider_api_keys_provider_active"):
if not _index_exists("idx_provider_api_keys_provider_active"):
op.create_index(
"idx_provider_api_keys_provider_active",
"provider_api_keys",
@@ -107,14 +207,14 @@ def upgrade() -> None:
def downgrade() -> None:
# -- 2. drop composite index
if index_exists("idx_provider_api_keys_provider_active"):
if _index_exists("idx_provider_api_keys_provider_active"):
op.drop_index(
"idx_provider_api_keys_provider_active",
table_name="provider_api_keys",
)
# -- 1. Numeric -> Float (batched per table)
c = new_cache()
c = _SchemaCache()
c.load_columns(_ALL_TABLES)
cols_to_revert = [
@@ -122,7 +222,7 @@ def downgrade() -> None:
for t, col, n, d in _COST_COLUMNS
if c.column_exists(t, col) and c.is_numeric(t, col)
]
batch_alter_type(
_batch_alter_type(
c,
cols_to_revert,
cast_suffix="double precision",

91
alembic/versions/20260309_0100_d7649c1f8e21_video_task_key_id_set_null_on_delete.py
View File

@@ -6,7 +6,11 @@ Create Date: 2026-03-09 01:00:00.000000+00:00
"""
from helpers import new_cache, replace_fk_if_needed
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "d7649c1f8e21"
@@ -18,10 +22,87 @@ _TABLE = "video_tasks"
_FK_NAME = "video_tasks_key_id_fkey"
# ---------------------------------------------------------------------------
# Inline helpers
# ---------------------------------------------------------------------------
class _SchemaCache:
def __init__(self) -> None:
self._fk_rules: dict[tuple[str, str], str] = {}
self._fk_loaded_tables: set[str] = set()
def load_fk_rules(self, tables: list[str]) -> None:
need = [t for t in tables if t not in self._fk_loaded_tables]
if not need:
return
bind = op.get_bind()
rows = bind.execute(
sa.text(
"SELECT tc.table_name, tc.constraint_name, rc.delete_rule "
"FROM information_schema.referential_constraints rc "
"JOIN information_schema.table_constraints tc "
" ON rc.constraint_name = tc.constraint_name "
" AND rc.constraint_schema = tc.constraint_schema "
"WHERE tc.table_name = ANY(:tables) "
" AND tc.table_schema = current_schema()"
),
{"tables": need},
).fetchall()
for table, name, rule in rows:
self._fk_rules[(table, name)] = rule
self._fk_loaded_tables.update(need)
def fk_ondelete(self, table: str, constraint: str) -> str | None:
return self._fk_rules.get((table, constraint))
def _fk_exists(constraint_name: str, table_name: str) -> bool:
bind = op.get_bind()
result = bind.execute(
sa.text(
"SELECT 1 FROM pg_constraint c "
"JOIN pg_class r ON c.conrelid = r.oid "
"JOIN pg_namespace n ON r.relnamespace = n.oid "
"WHERE c.conname = :name AND r.relname = :table "
" AND n.nspname = current_schema() AND c.contype = 'f'"
),
{"name": constraint_name, "table": table_name},
)
return result.scalar() is not None
def _replace_fk_if_needed(
cache: _SchemaCache,
constraint_name: str,
table_name: str,
ref_table: str,
local_cols: list[str],
remote_cols: list[str],
desired_ondelete: str,
) -> None:
current = cache.fk_ondelete(table_name, constraint_name)
if current and current.upper() == desired_ondelete.upper():
return
if current or _fk_exists(constraint_name, table_name):
op.drop_constraint(constraint_name, table_name, type_="foreignkey")
op.create_foreign_key(
constraint_name,
table_name,
ref_table,
local_cols,
remote_cols,
ondelete=desired_ondelete,
)
# ---------------------------------------------------------------------------
def upgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_fk_rules([_TABLE])
replace_fk_if_needed(
_replace_fk_if_needed(
c,
_FK_NAME,
_TABLE,
@@ -33,9 +114,9 @@ def upgrade() -> None:
def downgrade() -> None:
c = new_cache()
c = _SchemaCache()
c.load_fk_rules([_TABLE])
replace_fk_if_needed(
_replace_fk_if_needed(
c,
_FK_NAME,
_TABLE,

4
alembic/versions/20260310_1200_a3f1b7c9d2e4_strip_window_from_health_json.py
View File

@@ -3,7 +3,7 @@
This data is now maintained in process memory only, no longer persisted to DB.
Revision ID: a3f1b7c9d2e4
Revises: 2053ab8ed764
Revises: d7649c1f8e21
Create Date: 2026-03-10 12:00:00.000000+00:00
"""
@@ -12,7 +12,7 @@ from alembic import op
# revision identifiers, used by Alembic.
revision = "a3f1b7c9d2e4"
down_revision = "2053ab8ed764"
down_revision = "d7649c1f8e21"
branch_labels = None
depends_on = None