1019 lines
38 KiB
Python
1019 lines
38 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass, replace
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sysconfig
|
|
from typing import Any
|
|
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from alembic.runtime.migration import MigrationContext
|
|
from alembic.script import ScriptDirectory
|
|
from sqlalchemy import create_engine, inspect, text
|
|
|
|
from govoplan_core.core.migrations import MigrationMetadataPlan, migration_metadata_plan
|
|
from govoplan_core.core import change_sequence as core_change_sequence_models # noqa: F401 - populate core metadata
|
|
from govoplan_core.security import credential_envelopes as core_credential_models # noqa: F401 - populate core metadata
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry, ChangeSequenceRetentionFloor
|
|
from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids
|
|
from govoplan_core.core.modules import ModuleMigrationTaskContext, ModuleMigrationTaskResult
|
|
from govoplan_core.db.session import configure_database, get_database
|
|
from govoplan_core.server.default_config import get_server_config
|
|
from govoplan_core.server.config import ManifestFactory
|
|
from govoplan_core.server.registry import available_module_manifests, build_platform_registry
|
|
from govoplan_core.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Historic development databases could be created partly through Alembic and
|
|
# partly through Base.metadata.create_all(). In that state Alembic still says
|
|
# "2c..." while the 3d/4e file-storage tables already exist, so a normal
|
|
# upgrade attempts to create file_blobs again. This reconciliation is kept
|
|
# deliberately narrow and only advances the marker when the complete expected
|
|
# schema for the skipped revisions is already present.
|
|
REVISION_AUTH_RBAC = "2c3d4e5f6a7b"
|
|
REVISION_FILE_STORAGE = "3d4e5f6a7b8c"
|
|
REVISION_FILE_FOLDERS = "4e5f6a7b8c9d"
|
|
REVISION_NAMESPACE_PLATFORM_TABLES = "2e3f4a5b6c7d"
|
|
REVISION_CORE_CHANGE_SEQUENCE = "3f4a5b6c7d8e"
|
|
REVISION_HIERARCHICAL_SETTINGS = "f5a6b7c8d9e0"
|
|
LEGACY_SCOPE_TABLE = "tenancy_tenants"
|
|
CORE_SCOPE_TABLE = "core_scopes"
|
|
PRE_MIGRATION_TASK_PHASES = ("pre_migration_check", "pre_migration_prepare")
|
|
POST_MIGRATION_TASK_PHASES = ("post_migration_backfill", "post_migration_verify")
|
|
MIGRATION_TASK_PHASES = (*PRE_MIGRATION_TASK_PHASES, *POST_MIGRATION_TASK_PHASES)
|
|
MIGRATION_TRACK_RELEASE = "release"
|
|
MIGRATION_TRACK_DEV = "dev"
|
|
MIGRATION_TRACKS = (MIGRATION_TRACK_RELEASE, MIGRATION_TRACK_DEV)
|
|
_SQL_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
|
|
_NAMESPACE_TABLE_RENAMES = (
|
|
("tenants", "tenancy_tenants"),
|
|
("accounts", "access_accounts"),
|
|
("users", "access_users"),
|
|
("groups", "access_groups"),
|
|
("roles", "access_roles"),
|
|
("system_role_assignments", "access_system_role_assignments"),
|
|
("user_group_memberships", "access_user_group_memberships"),
|
|
("user_role_assignments", "access_user_role_assignments"),
|
|
("group_role_assignments", "access_group_role_assignments"),
|
|
("api_keys", "access_api_keys"),
|
|
("auth_sessions", "access_auth_sessions"),
|
|
("system_settings", "core_system_settings"),
|
|
("governance_templates", "admin_governance_templates"),
|
|
("governance_template_assignments", "admin_governance_template_assignments"),
|
|
)
|
|
|
|
_FILE_STORAGE_TABLES = {
|
|
"file_blobs",
|
|
"file_assets",
|
|
"file_versions",
|
|
"file_shares",
|
|
"campaign_attachment_uses",
|
|
}
|
|
_FILE_FOLDER_TABLES = {"file_folders"}
|
|
|
|
_FILE_STORAGE_COLUMNS = {
|
|
"file_blobs": {
|
|
"id",
|
|
"tenant_id",
|
|
"storage_backend",
|
|
"storage_key",
|
|
"checksum_sha256",
|
|
"size_bytes",
|
|
},
|
|
"file_assets": {
|
|
"id",
|
|
"tenant_id",
|
|
"owner_type",
|
|
"display_path",
|
|
"filename",
|
|
"current_version_id",
|
|
},
|
|
"file_versions": {
|
|
"id",
|
|
"file_asset_id",
|
|
"blob_id",
|
|
"version_number",
|
|
"checksum_sha256",
|
|
},
|
|
"file_shares": {"id", "file_asset_id", "target_type", "target_id", "permission"},
|
|
"campaign_attachment_uses": {
|
|
"id",
|
|
"campaign_id",
|
|
"campaign_version_id",
|
|
"file_asset_id",
|
|
"file_version_id",
|
|
"file_blob_id",
|
|
},
|
|
"file_folders": {"id", "tenant_id", "owner_type", "path"},
|
|
}
|
|
|
|
_CREATE_ALL_THROUGH_HIERARCHICAL_TABLES = {
|
|
"access_accounts",
|
|
"access_auth_sessions",
|
|
"audit_log",
|
|
"campaign_versions",
|
|
"campaign_jobs",
|
|
"send_attempts",
|
|
"access_system_role_assignments",
|
|
"core_system_settings",
|
|
"admin_governance_templates",
|
|
"admin_governance_template_assignments",
|
|
"mail_server_profiles",
|
|
}
|
|
|
|
_CREATE_ALL_THROUGH_HIERARCHICAL_COLUMNS = {
|
|
"access_auth_sessions": {"account_id", "csrf_token_hash"},
|
|
"audit_log": {"scope", "tenant_id", "user_id", "api_key_id"},
|
|
"campaign_versions": {
|
|
"workflow_state",
|
|
"current_flow",
|
|
"user_lock_state",
|
|
"user_locked_at",
|
|
"user_locked_by_user_id",
|
|
"execution_snapshot",
|
|
"execution_snapshot_hash",
|
|
"execution_snapshot_at",
|
|
},
|
|
"campaign_jobs": {"claimed_at", "claim_token", "smtp_started_at", "outcome_unknown_at", "eml_sha256"},
|
|
"send_attempts": {"status", "claim_token"},
|
|
"access_users": {"account_id", "settings", "mail_profile_policy"},
|
|
"access_groups": {"system_template_id", "system_required", "settings", "mail_profile_policy"},
|
|
"access_roles": {"system_template_id", "system_required"},
|
|
"campaigns": {"settings", "mail_profile_policy"},
|
|
"mail_server_profiles": {"scope_type", "scope_id"},
|
|
"core_system_settings": {"settings", "allow_tenant_custom_groups", "allow_tenant_custom_roles", "allow_tenant_api_keys"},
|
|
}
|
|
|
|
_SCOPE_TABLE_COLUMNS = {"settings", "allow_custom_groups", "allow_custom_roles", "allow_api_keys"}
|
|
|
|
_CREATE_ALL_THROUGH_HIERARCHICAL_INDEXES = {
|
|
"file_folders": {"uq_file_folders_active_user_path", "uq_file_folders_active_group_path"},
|
|
"campaign_versions": {
|
|
"ix_campaign_versions_user_lock_state",
|
|
"ix_campaign_versions_user_locked_by_user_id",
|
|
"ix_campaign_versions_execution_snapshot_hash",
|
|
},
|
|
"campaign_jobs": {"ix_campaign_jobs_claim_token", "ix_campaign_jobs_eml_sha256"},
|
|
"send_attempts": {"ix_send_attempts_status", "ix_send_attempts_claim_token"},
|
|
"audit_log": {"ix_audit_log_scope_created_at", "ix_audit_log_tenant_scope_created_at"},
|
|
"access_auth_sessions": {"ix_access_auth_sessions_account_id"},
|
|
"access_users": {"ix_access_users_account_id"},
|
|
"access_groups": {"ix_access_groups_system_template_id"},
|
|
"access_roles": {"ix_access_roles_system_template_id"},
|
|
"mail_server_profiles": {"ix_mail_server_profiles_scope_type", "ix_mail_server_profiles_scope_id", "ix_mail_server_profiles_scope"},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _LegacyCreateAllSchemaState:
|
|
current: str | None
|
|
has_no_revision: bool
|
|
has_file_storage: bool
|
|
has_file_folders: bool
|
|
has_create_all_hierarchical_schema: bool
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class MigrationResult:
|
|
previous_revision: str | None
|
|
reconciled_revision: str | None
|
|
current_revision: str | None
|
|
|
|
|
|
class ModuleMigrationTaskExecutionError(RuntimeError):
|
|
def __init__(self, message: str, *, records: tuple[dict[str, object], ...]) -> None:
|
|
super().__init__(message)
|
|
self.records = records
|
|
|
|
|
|
def registered_module_migration_plan(
|
|
database_url: str | None = None,
|
|
*,
|
|
enabled_modules: tuple[str, ...] | list[str] | None = None,
|
|
manifest_factories: tuple[ManifestFactory, ...] = (),
|
|
migration_track: str | None = None,
|
|
) -> MigrationMetadataPlan:
|
|
plan = migration_metadata_plan(_registered_module_registry(
|
|
database_url=database_url,
|
|
enabled_modules=enabled_modules,
|
|
manifest_factories=manifest_factories,
|
|
))
|
|
return _migration_plan_for_track(plan, _normalize_migration_track(migration_track))
|
|
|
|
|
|
def _registered_module_registry(
|
|
*,
|
|
database_url: str | None = None,
|
|
enabled_modules: tuple[str, ...] | list[str] | None = None,
|
|
manifest_factories: tuple[ManifestFactory, ...] = (),
|
|
):
|
|
server_config = get_server_config()
|
|
active_database_url = database_url or settings.database_url
|
|
if active_database_url:
|
|
# Registry planning may target a different database than the currently
|
|
# configured handle. The global handle is replaced either way, so
|
|
# dispose its superseded pool instead of leaking open DBAPI connections.
|
|
configure_database(active_database_url, dispose_previous=True)
|
|
active_manifest_factories = manifest_factories or tuple(server_config.manifest_factories)
|
|
raw_enabled_modules = tuple(enabled_modules) if enabled_modules is not None else load_startup_enabled_modules(server_config.enabled_modules)
|
|
candidate_modules = startup_candidate_module_ids(server_config.enabled_modules, raw_enabled_modules)
|
|
available_modules = available_module_manifests(
|
|
active_manifest_factories,
|
|
enabled_modules=candidate_modules,
|
|
ignore_load_errors=True,
|
|
)
|
|
active_enabled_modules = (
|
|
tuple(enabled_modules)
|
|
if enabled_modules is not None
|
|
else load_startup_enabled_modules(server_config.enabled_modules, available=available_modules)
|
|
)
|
|
registry = build_platform_registry(
|
|
active_enabled_modules,
|
|
manifest_factories=active_manifest_factories,
|
|
)
|
|
return registry
|
|
|
|
|
|
def run_registered_module_migration_tasks(
|
|
*,
|
|
database_url: str | None = None,
|
|
enabled_modules: tuple[str, ...] | list[str] | None = None,
|
|
migration_module_order: tuple[str, ...] | list[str] | None = None,
|
|
phases: tuple[str, ...] | list[str] = MIGRATION_TASK_PHASES,
|
|
dry_run: bool = False,
|
|
manifest_factories: tuple[ManifestFactory, ...] = (),
|
|
) -> tuple[dict[str, object], ...]:
|
|
active_phases = _normalized_migration_task_phases(phases)
|
|
url = database_url or settings.database_url
|
|
registry = _registered_module_registry(
|
|
database_url=url,
|
|
enabled_modules=enabled_modules,
|
|
manifest_factories=manifest_factories,
|
|
)
|
|
manifests = {manifest.id: manifest for manifest in registry.manifests()}
|
|
ordered_ids = _ordered_migration_task_module_ids(migration_module_order, manifests)
|
|
records: list[dict[str, object]] = []
|
|
database = get_database()
|
|
with database.SessionLocal() as session:
|
|
for phase in active_phases:
|
|
for manifest, task in _iter_phase_migration_tasks(phase, ordered_ids=ordered_ids, manifests=manifests):
|
|
_run_registered_module_migration_task(
|
|
session,
|
|
manifest=manifest,
|
|
task=task,
|
|
database_url=url,
|
|
dry_run=dry_run,
|
|
records=records,
|
|
)
|
|
return tuple(records)
|
|
|
|
|
|
def _normalized_migration_task_phases(phases: tuple[str, ...] | list[str]) -> tuple[str, ...]:
|
|
active_phases = tuple(dict.fromkeys(str(item).strip() for item in phases if str(item).strip()))
|
|
invalid_phases = tuple(phase for phase in active_phases if phase not in MIGRATION_TASK_PHASES)
|
|
if invalid_phases:
|
|
raise ValueError("Unsupported module migration task phase(s): " + ", ".join(invalid_phases))
|
|
return active_phases
|
|
|
|
|
|
def _ordered_migration_task_module_ids(
|
|
migration_module_order: tuple[str, ...] | list[str] | None,
|
|
manifests: Mapping[str, object],
|
|
) -> tuple[str, ...]:
|
|
return tuple(dict.fromkeys([
|
|
*(str(item).strip() for item in (migration_module_order or ()) if str(item).strip()),
|
|
*manifests.keys(),
|
|
]))
|
|
|
|
|
|
def _iter_phase_migration_tasks(
|
|
phase: str,
|
|
*,
|
|
ordered_ids: tuple[str, ...],
|
|
manifests: Mapping[str, object],
|
|
) -> Iterable[tuple[object, object]]:
|
|
for module_id in ordered_ids:
|
|
manifest = manifests.get(module_id)
|
|
migration_spec = getattr(manifest, "migration_spec", None)
|
|
if manifest is None or migration_spec is None:
|
|
continue
|
|
for task in migration_spec.migration_tasks:
|
|
if task.phase == phase:
|
|
yield manifest, task
|
|
|
|
|
|
def _run_registered_module_migration_task(
|
|
session: object,
|
|
*,
|
|
manifest: object,
|
|
task: object,
|
|
database_url: str,
|
|
dry_run: bool,
|
|
records: list[dict[str, object]],
|
|
) -> None:
|
|
record = _migration_task_record(manifest, task, dry_run=dry_run)
|
|
_validate_migration_task_can_run(manifest, task, record, records, dry_run=dry_run)
|
|
if dry_run:
|
|
record.update({"status": "skipped", "message": "Dry run; executor was not called."})
|
|
records.append(record)
|
|
return
|
|
normalized = _execute_module_migration_task(
|
|
session,
|
|
manifest=manifest,
|
|
task=task,
|
|
record=record,
|
|
records=records,
|
|
database_url=database_url,
|
|
dry_run=dry_run,
|
|
)
|
|
record.update({
|
|
"status": normalized.status,
|
|
"message": normalized.message,
|
|
"details": _jsonable_migration_task_details(normalized.details),
|
|
})
|
|
records.append(record)
|
|
if normalized.status == "blocked":
|
|
session.rollback()
|
|
raise ModuleMigrationTaskExecutionError(
|
|
normalized.message or f"Module migration task {manifest.id}/{task.task_id} blocked migration.",
|
|
records=tuple(records),
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _migration_task_record(manifest: object, task: object, *, dry_run: bool) -> dict[str, object]:
|
|
record: dict[str, object] = {
|
|
"module_id": manifest.id,
|
|
"task_id": task.task_id,
|
|
"phase": task.phase,
|
|
"summary": task.summary,
|
|
"task_version": task.task_version,
|
|
"safety": task.safety,
|
|
"idempotent": task.idempotent,
|
|
"dry_run": dry_run,
|
|
}
|
|
if task.timeout_seconds is not None:
|
|
record["timeout_seconds"] = task.timeout_seconds
|
|
return record
|
|
|
|
|
|
def _validate_migration_task_can_run(
|
|
manifest: object,
|
|
task: object,
|
|
record: dict[str, object],
|
|
records: list[dict[str, object]],
|
|
*,
|
|
dry_run: bool,
|
|
) -> None:
|
|
if not task.idempotent:
|
|
_block_migration_task(
|
|
manifest,
|
|
task,
|
|
record,
|
|
records,
|
|
message="Task is not idempotent.",
|
|
error=f"Module migration task {manifest.id}/{task.task_id} is not idempotent.",
|
|
)
|
|
if not dry_run and task.executor is None:
|
|
_block_migration_task(
|
|
manifest,
|
|
task,
|
|
record,
|
|
records,
|
|
message="Task has no executor.",
|
|
error=f"Module migration task {manifest.id}/{task.task_id} has no executor.",
|
|
)
|
|
|
|
|
|
def _block_migration_task(
|
|
manifest: object,
|
|
task: object,
|
|
record: dict[str, object],
|
|
records: list[dict[str, object]],
|
|
*,
|
|
message: str,
|
|
error: str,
|
|
) -> None:
|
|
record.update({"status": "blocked", "message": message})
|
|
records.append(record)
|
|
raise ModuleMigrationTaskExecutionError(
|
|
error,
|
|
records=tuple(records),
|
|
)
|
|
|
|
|
|
def _execute_module_migration_task(
|
|
session: object,
|
|
*,
|
|
manifest: object,
|
|
task: object,
|
|
record: dict[str, object],
|
|
records: list[dict[str, object]],
|
|
database_url: str,
|
|
dry_run: bool,
|
|
) -> ModuleMigrationTaskResult:
|
|
context = ModuleMigrationTaskContext(
|
|
module_id=manifest.id,
|
|
task_id=task.task_id,
|
|
phase=task.phase,
|
|
database_url=database_url,
|
|
target_version=manifest.version,
|
|
session=session,
|
|
dry_run=dry_run,
|
|
metadata=task.metadata,
|
|
)
|
|
try:
|
|
return _normalize_migration_task_result(task.executor(context))
|
|
except Exception as exc:
|
|
session.rollback()
|
|
record.update({
|
|
"status": "blocked",
|
|
"message": f"{type(exc).__name__}: {exc}",
|
|
})
|
|
records.append(record)
|
|
raise ModuleMigrationTaskExecutionError(
|
|
f"Module migration task {manifest.id}/{task.task_id} failed.",
|
|
records=tuple(records),
|
|
) from exc
|
|
|
|
|
|
def _normalize_migration_task_result(result: ModuleMigrationTaskResult | None) -> ModuleMigrationTaskResult:
|
|
if result is None:
|
|
return ModuleMigrationTaskResult()
|
|
if not isinstance(result, ModuleMigrationTaskResult):
|
|
raise TypeError("Module migration task executors must return ModuleMigrationTaskResult or None.")
|
|
if result.status not in {"ok", "warning", "blocked", "skipped"}:
|
|
raise ValueError(f"Unsupported module migration task status: {result.status!r}")
|
|
return result
|
|
|
|
|
|
def _jsonable_migration_task_details(value: Mapping[str, Any]) -> Mapping[str, Any] | str:
|
|
try:
|
|
json.dumps(value)
|
|
except (TypeError, ValueError):
|
|
return str(dict(value))
|
|
return value
|
|
|
|
|
|
def _repo_root() -> Path:
|
|
packaged_root = Path(__file__).resolve().parents[3]
|
|
installed_runtime_root = Path(sysconfig.get_path("data")) / "govoplan_core_runtime"
|
|
configured = os.environ.get("GOVOPLAN_CORE_SOURCE_ROOT")
|
|
candidates = [
|
|
Path(configured).expanduser() if configured else None,
|
|
Path.cwd(),
|
|
packaged_root,
|
|
installed_runtime_root,
|
|
]
|
|
for candidate in candidates:
|
|
if candidate is None:
|
|
continue
|
|
resolved = candidate.resolve()
|
|
if (resolved / "alembic.ini").exists() and (resolved / "alembic").is_dir():
|
|
return resolved
|
|
return packaged_root
|
|
|
|
|
|
def _normalize_migration_track(value: str | None = None) -> str:
|
|
raw_value = value if value is not None else settings.migration_track
|
|
track = str(raw_value or MIGRATION_TRACK_RELEASE).strip().lower()
|
|
aliases = {
|
|
"baseline": MIGRATION_TRACK_RELEASE,
|
|
"prod": MIGRATION_TRACK_RELEASE,
|
|
"production": MIGRATION_TRACK_RELEASE,
|
|
"releases": MIGRATION_TRACK_RELEASE,
|
|
"detailed": MIGRATION_TRACK_DEV,
|
|
"development": MIGRATION_TRACK_DEV,
|
|
}
|
|
track = aliases.get(track, track)
|
|
if track not in MIGRATION_TRACKS:
|
|
raise ValueError(
|
|
"Unsupported migration track "
|
|
f"{raw_value!r}; expected one of: {', '.join(MIGRATION_TRACKS)}"
|
|
)
|
|
return track
|
|
|
|
|
|
def _core_version_location(root: Path, migration_track: str) -> Path:
|
|
if migration_track == MIGRATION_TRACK_DEV:
|
|
dev_versions = root / "alembic" / "dev_versions"
|
|
if dev_versions.is_dir():
|
|
return dev_versions
|
|
return root / "alembic" / "versions"
|
|
|
|
|
|
def _script_location_for_track(location: str | None, migration_track: str) -> str | None:
|
|
if not location or migration_track != MIGRATION_TRACK_DEV:
|
|
return location
|
|
path = Path(location)
|
|
if path.name != "versions":
|
|
return location
|
|
dev_location = path.with_name("dev_versions")
|
|
if dev_location.is_dir():
|
|
return str(dev_location)
|
|
return location
|
|
|
|
|
|
def _migration_plan_for_track(plan: MigrationMetadataPlan, migration_track: str) -> MigrationMetadataPlan:
|
|
if migration_track == MIGRATION_TRACK_RELEASE:
|
|
return plan
|
|
script_locations = tuple(
|
|
location
|
|
for location in (
|
|
_script_location_for_track(location, migration_track)
|
|
for location in plan.script_locations
|
|
)
|
|
if location
|
|
)
|
|
modules = tuple(
|
|
replace(
|
|
item,
|
|
script_location=_script_location_for_track(item.script_location, migration_track),
|
|
)
|
|
for item in plan.modules
|
|
)
|
|
return replace(plan, script_locations=script_locations, modules=modules)
|
|
|
|
|
|
def alembic_config(
|
|
*,
|
|
database_url: str | None = None,
|
|
enabled_modules: tuple[str, ...] | list[str] | None = None,
|
|
manifest_factories: tuple[ManifestFactory, ...] = (),
|
|
migration_track: str | None = None,
|
|
) -> Config:
|
|
root = _repo_root()
|
|
active_migration_track = _normalize_migration_track(migration_track)
|
|
config = Config(str(root / "alembic.ini"))
|
|
config.set_main_option("script_location", str(root / "alembic"))
|
|
|
|
active_database_url = database_url or settings.database_url
|
|
plan = registered_module_migration_plan(
|
|
active_database_url,
|
|
enabled_modules=enabled_modules,
|
|
manifest_factories=manifest_factories,
|
|
migration_track=active_migration_track,
|
|
)
|
|
version_locations = [str(_core_version_location(root, active_migration_track))]
|
|
version_locations.extend(location for location in plan.script_locations if location)
|
|
config.set_main_option("version_locations", os.pathsep.join(dict.fromkeys(version_locations)))
|
|
config.set_main_option("path_separator", "os")
|
|
|
|
config.attributes["database_url"] = active_database_url
|
|
config.attributes["migration_track"] = active_migration_track
|
|
if enabled_modules is not None:
|
|
config.attributes["enabled_modules"] = tuple(enabled_modules)
|
|
if manifest_factories:
|
|
config.attributes["manifest_factories"] = tuple(manifest_factories)
|
|
return config
|
|
|
|
|
|
def database_revision(database_url: str | None = None) -> str | None:
|
|
url = database_url or settings.database_url
|
|
engine = create_engine(url)
|
|
try:
|
|
with engine.connect() as connection:
|
|
heads = MigrationContext.configure(connection).get_current_heads()
|
|
if not heads:
|
|
return None
|
|
if len(heads) == 1:
|
|
return heads[0]
|
|
return ",".join(sorted(heads))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def _has_columns(inspector, table_name: str, required: set[str]) -> bool:
|
|
try:
|
|
actual = {column["name"] for column in inspector.get_columns(table_name)}
|
|
except Exception:
|
|
return False
|
|
return required.issubset(actual)
|
|
|
|
|
|
def _has_indexes(inspector, table_name: str, required: set[str]) -> bool:
|
|
try:
|
|
actual = {index["name"] for index in inspector.get_indexes(table_name)}
|
|
except Exception:
|
|
return False
|
|
return required.issubset(actual)
|
|
|
|
|
|
def _has_create_all_schema_through_hierarchical_settings(inspector, tables: set[str]) -> bool:
|
|
if not _CREATE_ALL_THROUGH_HIERARCHICAL_TABLES.issubset(tables):
|
|
return False
|
|
if not any(
|
|
table_name in tables and _has_columns(inspector, table_name, _SCOPE_TABLE_COLUMNS)
|
|
for table_name in (CORE_SCOPE_TABLE, LEGACY_SCOPE_TABLE)
|
|
):
|
|
return False
|
|
for table_name, required_columns in _CREATE_ALL_THROUGH_HIERARCHICAL_COLUMNS.items():
|
|
if table_name not in tables or not _has_columns(inspector, table_name, required_columns):
|
|
return False
|
|
for table_name, required_indexes in _CREATE_ALL_THROUGH_HIERARCHICAL_INDEXES.items():
|
|
if table_name not in tables or not _has_indexes(inspector, table_name, required_indexes):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _backfill_user_lock_state_for_create_all_schema(database_url: str) -> None:
|
|
engine = create_engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(text("""
|
|
UPDATE campaign_versions
|
|
SET user_lock_state = 'permanent',
|
|
user_locked_at = published_at,
|
|
user_locked_by_user_id = NULL
|
|
WHERE published_at IS NOT NULL
|
|
AND user_lock_state IS NULL
|
|
"""))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def _row_count(connection, table_name: str) -> int:
|
|
quoted = _quoted_table_name(connection, table_name)
|
|
statement = text(f"SELECT COUNT(*) FROM {quoted}") # noqa: S608 # nosec B608 # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
|
|
return int(connection.execute(statement).scalar_one())
|
|
|
|
|
|
def _drop_table(connection, table_name: str) -> None:
|
|
quoted = _quoted_table_name(connection, table_name)
|
|
statement = text(f"DROP TABLE {quoted}") # nosec B608 # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
|
|
connection.execute(statement)
|
|
|
|
|
|
def _rename_table(connection, old_name: str, new_name: str) -> None:
|
|
quoted_old = _quoted_table_name(connection, old_name)
|
|
quoted_new = _quoted_table_name(connection, new_name)
|
|
statement = text(f"ALTER TABLE {quoted_old} RENAME TO {quoted_new}") # nosec B608 # nosemgrep: python.sqlalchemy.security.audit.avoid-sqlalchemy-text.avoid-sqlalchemy-text
|
|
connection.execute(statement)
|
|
|
|
|
|
def _quoted_table_name(connection, table_name: str) -> str:
|
|
if not _SQL_IDENTIFIER_RE.fullmatch(table_name):
|
|
raise ValueError(f"Unsafe table identifier: {table_name!r}")
|
|
return connection.dialect.identifier_preparer.quote(table_name)
|
|
|
|
|
|
def _reconcile_scope_table_names(connection, tables: set[str]) -> bool:
|
|
if LEGACY_SCOPE_TABLE not in tables:
|
|
return False
|
|
|
|
changed = False
|
|
if CORE_SCOPE_TABLE in tables:
|
|
if _row_count(connection, CORE_SCOPE_TABLE) == 0:
|
|
_drop_table(connection, CORE_SCOPE_TABLE)
|
|
tables.remove(CORE_SCOPE_TABLE)
|
|
changed = True
|
|
elif _row_count(connection, LEGACY_SCOPE_TABLE) == 0:
|
|
_drop_table(connection, LEGACY_SCOPE_TABLE)
|
|
tables.remove(LEGACY_SCOPE_TABLE)
|
|
changed = True
|
|
else:
|
|
raise RuntimeError(f"Cannot reconcile non-empty {LEGACY_SCOPE_TABLE} over non-empty {CORE_SCOPE_TABLE}")
|
|
|
|
if LEGACY_SCOPE_TABLE in tables and CORE_SCOPE_TABLE not in tables:
|
|
_rename_table(connection, LEGACY_SCOPE_TABLE, CORE_SCOPE_TABLE)
|
|
tables.remove(LEGACY_SCOPE_TABLE)
|
|
tables.add(CORE_SCOPE_TABLE)
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def reconcile_scope_table_retirement_drift(database_url: str | None = None) -> bool:
|
|
"""Repair databases that are stamped after the scope-table rename.
|
|
|
|
The real Alembic migration performs the same rename. This helper covers
|
|
development databases that were manually stamped or partially created via
|
|
create_all(), where the version marker can be newer than the table name.
|
|
It runs after Alembic upgrades so older migrations can still use the
|
|
historical table name while they execute.
|
|
"""
|
|
|
|
url = database_url or settings.database_url
|
|
changed = False
|
|
engine = create_engine(url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
tables = set(inspect(connection).get_table_names())
|
|
changed = _reconcile_scope_table_names(connection, tables)
|
|
finally:
|
|
engine.dispose()
|
|
return changed
|
|
|
|
|
|
def reconcile_namespace_table_drift(database_url: str | None = None) -> bool:
|
|
"""Repair dev databases stamped past the namespace-table migration.
|
|
|
|
During the repository split some development databases were stamped at the
|
|
newer Alembic heads while still carrying the old platform table names. The
|
|
real migration only renames tables, so replay that idempotent operation
|
|
before startup bootstrap code touches the ORM.
|
|
"""
|
|
|
|
url = database_url or settings.database_url
|
|
changed = False
|
|
engine = create_engine(url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
schema = inspect(connection)
|
|
tables = set(schema.get_table_names())
|
|
if not any(old_name in tables and new_name not in tables for old_name, new_name in _NAMESPACE_TABLE_RENAMES):
|
|
return False
|
|
|
|
old_tables_to_drop: set[str] = set()
|
|
new_tables_to_drop: set[str] = set()
|
|
for old_name, new_name in _NAMESPACE_TABLE_RENAMES:
|
|
if old_name not in tables or new_name not in tables:
|
|
continue
|
|
if _row_count(connection, old_name) == 0:
|
|
old_tables_to_drop.add(old_name)
|
|
elif _row_count(connection, new_name) == 0:
|
|
new_tables_to_drop.add(new_name)
|
|
else:
|
|
raise RuntimeError(f"Cannot reconcile non-empty {old_name} over non-empty {new_name}")
|
|
|
|
for old_name, _new_name in reversed(_NAMESPACE_TABLE_RENAMES):
|
|
if old_name in old_tables_to_drop:
|
|
_drop_table(connection, old_name)
|
|
tables.remove(old_name)
|
|
changed = True
|
|
for _old_name, new_name in reversed(_NAMESPACE_TABLE_RENAMES):
|
|
if new_name in new_tables_to_drop:
|
|
_drop_table(connection, new_name)
|
|
tables.remove(new_name)
|
|
changed = True
|
|
|
|
for old_name, new_name in _NAMESPACE_TABLE_RENAMES:
|
|
if old_name not in tables or new_name in tables:
|
|
continue
|
|
_rename_table(connection, old_name, new_name)
|
|
tables.remove(old_name)
|
|
tables.add(new_name)
|
|
changed = True
|
|
finally:
|
|
engine.dispose()
|
|
return changed
|
|
|
|
|
|
def reconcile_change_sequence_retention_floor_drift(database_url: str | None = None) -> bool:
|
|
"""Repair databases stamped after the change-sequence migration changed.
|
|
|
|
Early development databases may have applied the change-sequence revision
|
|
before the retention-floor table was added to that migration. The main
|
|
change-sequence table is enough for full snapshots, but incremental delta
|
|
requests need the retention floor to decide whether a watermark is stale.
|
|
"""
|
|
|
|
url = database_url or settings.database_url
|
|
changed = False
|
|
engine = create_engine(url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
schema = inspect(connection)
|
|
tables = set(schema.get_table_names())
|
|
if ChangeSequenceEntry.__tablename__ not in tables:
|
|
return False
|
|
if ChangeSequenceRetentionFloor.__tablename__ not in tables:
|
|
ChangeSequenceRetentionFloor.__table__.create(bind=connection, checkfirst=True)
|
|
changed = True
|
|
else:
|
|
indexes = {
|
|
index["name"]
|
|
for index in schema.get_indexes(ChangeSequenceRetentionFloor.__tablename__)
|
|
}
|
|
for index in ChangeSequenceRetentionFloor.__table__.indexes:
|
|
if index.name not in indexes:
|
|
index.create(bind=connection)
|
|
changed = True
|
|
finally:
|
|
engine.dispose()
|
|
return changed
|
|
|
|
|
|
def reconcile_covered_alembic_dependency_heads(
|
|
database_url: str | None = None,
|
|
*,
|
|
config: Config | None = None,
|
|
) -> tuple[str, ...]:
|
|
"""Remove stale parent/dependency rows from ``alembic_version``.
|
|
|
|
Alembic's version table represents current graph heads, not every owner
|
|
head that contributed to the schema. After the v0.1.7 baseline squash,
|
|
development databases may still carry dependency parents such as the core
|
|
baseline beside child module heads. That shape is semantically already
|
|
upgraded, but Alembic rejects it as an overlapping current revision set.
|
|
"""
|
|
|
|
url = database_url or settings.database_url
|
|
active_config = config or alembic_config(database_url=url)
|
|
script = ScriptDirectory.from_config(active_config)
|
|
engine = create_engine(url)
|
|
removed: set[str] = set()
|
|
try:
|
|
with engine.begin() as connection:
|
|
tables = set(inspect(connection).get_table_names())
|
|
if "alembic_version" not in tables:
|
|
return ()
|
|
current = {
|
|
str(row[0])
|
|
for row in connection.execute(text("SELECT version_num FROM alembic_version")).all()
|
|
if row[0]
|
|
}
|
|
if len(current) < 2:
|
|
return ()
|
|
covered: set[str] = set()
|
|
for revision_id in current:
|
|
try:
|
|
revision = script.get_revision(revision_id)
|
|
except Exception as exc:
|
|
logger.debug("Skipping Alembic revision %s while pruning dependency heads: %s", revision_id, exc, exc_info=True)
|
|
continue
|
|
ancestors = {
|
|
item.revision
|
|
for item in script.revision_map._get_ancestor_nodes( # noqa: SLF001 - Alembic has no public dependency-ancestor API.
|
|
[revision],
|
|
include_dependencies=True,
|
|
)
|
|
}
|
|
covered.update(ancestors - {revision_id})
|
|
removed = current.intersection(covered)
|
|
for revision_id in sorted(removed):
|
|
connection.execute(
|
|
text("DELETE FROM alembic_version WHERE version_num = :revision"),
|
|
{"revision": revision_id},
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
return tuple(sorted(removed))
|
|
|
|
|
|
def reconcile_legacy_create_all_schema(
|
|
database_url: str | None = None,
|
|
*,
|
|
migration_track: str | None = None,
|
|
) -> str | None:
|
|
"""Repair the known Alembic/create_all drift without modifying application data.
|
|
|
|
Returns the revision stamped during reconciliation, or ``None`` when no
|
|
repair was necessary. A partial/unknown schema is intentionally left alone
|
|
so Alembic can fail visibly instead of guessing.
|
|
"""
|
|
|
|
url = database_url or settings.database_url
|
|
state = _legacy_create_all_schema_state(url)
|
|
target = _legacy_create_all_reconciliation_target(state)
|
|
if target is None:
|
|
return None
|
|
if target == REVISION_HIERARCHICAL_SETTINGS:
|
|
_backfill_user_lock_state_for_create_all_schema(url)
|
|
|
|
command.stamp(alembic_config(database_url=url, migration_track=migration_track), target)
|
|
return target
|
|
|
|
|
|
def _legacy_create_all_schema_state(database_url: str) -> _LegacyCreateAllSchemaState:
|
|
engine = create_engine(database_url)
|
|
try:
|
|
with engine.connect() as connection:
|
|
heads = MigrationContext.configure(connection).get_current_heads()
|
|
schema = inspect(connection)
|
|
tables = set(schema.get_table_names())
|
|
return _legacy_create_all_schema_state_from_inspection(heads, schema, tables)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def _legacy_create_all_schema_state_from_inspection(
|
|
heads: tuple[str, ...],
|
|
schema: object,
|
|
tables: set[str],
|
|
) -> _LegacyCreateAllSchemaState:
|
|
has_file_storage = _FILE_STORAGE_TABLES.issubset(tables) and all(
|
|
_has_columns(schema, table, _FILE_STORAGE_COLUMNS[table])
|
|
for table in _FILE_STORAGE_TABLES
|
|
)
|
|
has_file_folders = _FILE_FOLDER_TABLES.issubset(tables) and _has_columns(
|
|
schema,
|
|
"file_folders",
|
|
_FILE_STORAGE_COLUMNS["file_folders"],
|
|
)
|
|
return _LegacyCreateAllSchemaState(
|
|
current=heads[0] if len(heads) == 1 else None,
|
|
has_no_revision=len(heads) == 0,
|
|
has_file_storage=has_file_storage,
|
|
has_file_folders=has_file_folders,
|
|
has_create_all_hierarchical_schema=_has_create_all_schema_through_hierarchical_settings(schema, tables),
|
|
)
|
|
|
|
|
|
def _legacy_create_all_reconciliation_target(state: _LegacyCreateAllSchemaState) -> str | None:
|
|
if state.current == REVISION_AUTH_RBAC and state.has_file_storage and state.has_file_folders:
|
|
return REVISION_FILE_FOLDERS
|
|
if state.current == REVISION_AUTH_RBAC and state.has_file_storage:
|
|
return REVISION_FILE_STORAGE
|
|
if state.current == REVISION_FILE_STORAGE and state.has_file_folders:
|
|
return REVISION_FILE_FOLDERS
|
|
if state.current == REVISION_FILE_FOLDERS and state.has_create_all_hierarchical_schema:
|
|
return REVISION_HIERARCHICAL_SETTINGS
|
|
if state.has_no_revision and state.has_create_all_hierarchical_schema:
|
|
return REVISION_HIERARCHICAL_SETTINGS
|
|
if state.has_no_revision and state.has_file_storage and state.has_file_folders:
|
|
return REVISION_FILE_FOLDERS
|
|
return None
|
|
|
|
|
|
def migrate_database(
|
|
*,
|
|
database_url: str | None = None,
|
|
reconcile_legacy_schema: bool = True,
|
|
enabled_modules: tuple[str, ...] | list[str] | None = None,
|
|
migration_module_order: tuple[str, ...] | list[str] | None = None,
|
|
manifest_factories: tuple[ManifestFactory, ...] = (),
|
|
migration_track: str | None = None,
|
|
) -> MigrationResult:
|
|
url = database_url or settings.database_url
|
|
active_migration_track = _normalize_migration_track(migration_track)
|
|
if reconcile_legacy_schema:
|
|
reconcile_namespace_table_drift(url)
|
|
reconcile_change_sequence_retention_floor_drift(url)
|
|
previous = database_revision(url)
|
|
reconciled = (
|
|
reconcile_legacy_create_all_schema(url, migration_track=active_migration_track)
|
|
if reconcile_legacy_schema
|
|
else None
|
|
)
|
|
config = alembic_config(
|
|
database_url=url,
|
|
enabled_modules=enabled_modules,
|
|
manifest_factories=manifest_factories,
|
|
migration_track=active_migration_track,
|
|
)
|
|
reconcile_covered_alembic_dependency_heads(url, config=config)
|
|
if migration_module_order:
|
|
plan = registered_module_migration_plan(
|
|
url,
|
|
enabled_modules=enabled_modules,
|
|
manifest_factories=manifest_factories,
|
|
migration_track=active_migration_track,
|
|
)
|
|
_upgrade_ordered_module_heads(config, plan, tuple(migration_module_order))
|
|
command.upgrade(config, "heads")
|
|
if reconcile_legacy_schema:
|
|
reconcile_scope_table_retirement_drift(url)
|
|
current = database_revision(url)
|
|
return MigrationResult(
|
|
previous_revision=previous,
|
|
reconciled_revision=reconciled,
|
|
current_revision=current,
|
|
)
|
|
|
|
|
|
def _upgrade_ordered_module_heads(
|
|
config: Config,
|
|
plan: MigrationMetadataPlan,
|
|
module_order: tuple[str, ...],
|
|
) -> None:
|
|
module_heads = _module_heads_by_id(config, plan)
|
|
upgraded: set[str] = set()
|
|
for module_id in dict.fromkeys(module_order):
|
|
for head in module_heads.get(module_id, ()):
|
|
if head in upgraded:
|
|
continue
|
|
command.upgrade(config, head)
|
|
upgraded.add(head)
|
|
|
|
|
|
def _module_heads_by_id(config: Config, plan: MigrationMetadataPlan) -> dict[str, tuple[str, ...]]:
|
|
script = ScriptDirectory.from_config(config)
|
|
heads = script.get_heads()
|
|
module_locations = {
|
|
item.module_id: Path(item.script_location).resolve()
|
|
for item in plan.modules
|
|
if item.script_location
|
|
}
|
|
grouped: dict[str, list[str]] = {module_id: [] for module_id in module_locations}
|
|
for head in heads:
|
|
revision = script.get_revision(head)
|
|
path_text = getattr(revision, "path", None)
|
|
if revision is None or not path_text:
|
|
continue
|
|
revision_path = Path(path_text).resolve()
|
|
for module_id, location in module_locations.items():
|
|
try:
|
|
revision_path.relative_to(location)
|
|
except ValueError:
|
|
continue
|
|
grouped[module_id].append(head)
|
|
break
|
|
return {module_id: tuple(sorted(values)) for module_id, values in grouped.items() if values}
|