diff --git a/.env.example b/.env.example index a91aa2e..d0519a7 100644 --- a/.env.example +++ b/.env.example @@ -50,7 +50,9 @@ NETX_UME_NOTIFICATION_TOPIC=ALARM # API docs (/docs, /redoc, /openapi.json); default false. Lab: set true. # NETX_DOCS_ENABLED=true # NETX_ALLOW_INSECURE_DEFAULTS=false -# NETX_SKIP_LEGACY_STARTUP_DDL=true +# Schema: Alembic auto-upgrades on API start (defaults on). Override only if needed: +# NETX_ALEMBIC_UPGRADE_ON_START=false +# NETX_SKIP_LEGACY_STARTUP_DDL=false # NETX_SQL_READONLY_DATABASE_URL=postgresql+psycopg://netx_ro:xxx@127.0.0.1:5432/netx # NETX_RUN_INLINE_SCHEDULERS=true # NETX_AUDIT_ASYNC=true diff --git a/PROD_MIN_CHECKLIST.md b/PROD_MIN_CHECKLIST.md index c774e1a..b26754e 100644 --- a/PROD_MIN_CHECKLIST.md +++ b/PROD_MIN_CHECKLIST.md @@ -11,7 +11,7 @@ ## Runtime - Ensure PostgreSQL backup policy exists (daily logical backup + retention). -- Prefer Alembic: see [docs/ALEMBIC.md](docs/ALEMBIC.md). After `alembic upgrade head`, set `NETX_SKIP_LEGACY_STARTUP_DDL=true`. +- Schema: API auto-runs `alembic upgrade head` on start (see [docs/ALEMBIC.md](docs/ALEMBIC.md)). No manual migrate flag required for normal deploys. - Optional: `NETX_RUN_INLINE_SCHEDULERS=false` and run `python -m netx_api.worker` for collectors. - Run `oclaw` and `netx` under process managers (systemd/Windows service/pm2 equivalent). - Enable auto-restart and startup-at-boot for both services. diff --git a/alembic/versions/20260802_legacy_schema.py b/alembic/versions/20260802_legacy_schema.py new file mode 100644 index 0000000..9b3f10b --- /dev/null +++ b/alembic/versions/20260802_legacy_schema.py @@ -0,0 +1,40 @@ +"""Apply shared brownfield schema patches (legacy startup DDL). + +Revision ID: 20260802_legacy +Revises: 20260802_scopes +Create Date: 2026-08-02 + +Idempotent: safe on DBs that already received startup ALTER TABLE. +Fresh installs: prefer ``create_all`` then ``alembic upgrade head``. +Brownfield already patched: ``alembic stamp head`` then set +``NETX_SKIP_LEGACY_STARTUP_DDL=true``. +""" + +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op + +revision: str = "20260802_legacy" +down_revision: Union[str, Sequence[str], None] = "20260802_scopes" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + from netx_api.schema_patches import ( + apply_auth_schema_patches, + apply_domain_schema_patches, + apply_key_alert_schema_patches, + ) + + bind = op.get_bind() + apply_auth_schema_patches(bind) + apply_key_alert_schema_patches(conn=bind) + apply_domain_schema_patches(bind) + + +def downgrade() -> None: + # Brownfield additive patches are not safely reversible. + pass diff --git a/docs/ALEMBIC.md b/docs/ALEMBIC.md index fd921bf..f6fc8a7 100644 --- a/docs/ALEMBIC.md +++ b/docs/ALEMBIC.md @@ -1,22 +1,45 @@ # Alembic (schema migrations) -netx historically evolved the schema with startup `ALTER TABLE … IF NOT EXISTS`. -Alembic is the preferred path going forward. +Schema evolution uses Alembic. On API startup, netx **automatically** runs +`alembic upgrade head` (same patches as [`netx_api/schema_patches.py`](../netx_api/schema_patches.py)). +You normally do **not** need extra env flags. -## Commands +## What happens on boot + +1. `create_all` — create any missing tables from ORM metadata +2. `alembic upgrade head` — apply revisions (idempotent brownfield patches) +3. Auth column safety-net ensures (before admin bootstrap) +4. Legacy inline ALTER block is **skipped** by default (already covered by Alembic) + +## Commands (optional / CI) ```powershell cd netx .\.venv\Scripts\alembic.exe upgrade head +.\.venv\Scripts\alembic.exe current +.\.venv\Scripts\alembic.exe history ``` -## Env +### Brownfield DB that already received old startup DDL -| Variable | Meaning | +First boot after this change will run `upgrade head`. Patches are idempotent. +If Alembic history was never stamped and you prefer to mark current without re-running: + +```powershell +.\.venv\Scripts\alembic.exe stamp head +``` + +## Env overrides (usually leave defaults) + +| Variable | Default | Meaning | +|----------|---------|---------| +| `NETX_ALEMBIC_UPGRADE_ON_START` | `true` | Run `alembic upgrade head` on API start. Set `false` only if a separate migrate job owns upgrades. | +| `NETX_SKIP_LEGACY_STARTUP_DDL` | `true` | Skip the duplicate inline ALTER path. Set `false` only as emergency fallback. | +| `NETX_DATABASE_URL` | — | Same URL Alembic reads via `netx_api.config.settings`. | + +## Revisions + +| Revision | Purpose | |----------|---------| -| `NETX_SKIP_LEGACY_STARTUP_DDL=true` | Skip the large ad-hoc ALTER block in API startup (keep auth `scopes` column ensures). Use after `alembic upgrade head`. | -| `NETX_DATABASE_URL` | Same URL Alembic reads via `netx_api.config.settings`. | - -Fresh lab installs can keep the legacy startup DDL (`false`, default) until you adopt Alembic in your deploy checklist. - -Revision for capability scopes: `alembic/versions/20260802_scopes.py`. +| `20260802_scopes` | `app_user.scopes` / `api_token.scopes` | +| `20260802_legacy` | Shared brownfield patches (alarms, inventory, managed_ne, topology, port traffic, key-alert, …) | diff --git a/netx_api/config.py b/netx_api/config.py index 7a31f9c..c467120 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -133,8 +133,11 @@ class Settings(BaseSettings): # Async audit writer; sample_n>1 keeps 1/N of generic http.* events. audit_async: bool = True audit_sample_n: int = 1 - # Prefer Alembic; when true, skip ad-hoc ALTER TABLE on startup (create_all still runs). - skip_legacy_startup_ddl: bool = False + # Prefer Alembic on API start; brownfield patches live in schema_patches + revisions. + # Auth column ensures still run as a safety net before bootstrap. + skip_legacy_startup_ddl: bool = True + # Run `alembic upgrade head` during API startup (recommended default). + alembic_upgrade_on_start: bool = True # Optional dedicated SQLAlchemy URL for /v1/sql/* (read-only DB role recommended). sql_readonly_database_url: str = "" # When false, API skips config_sync / lldp / port_traffic schedulers (run `python -m netx_api.worker`). diff --git a/netx_api/main.py b/netx_api/main.py index d3d1fd1..236d3ed 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -90,21 +90,6 @@ from .ume_alarm_ws import ( start_ume_alarm_ws_consumer, ) from .ume_sync_service import sync_alarms_current, sync_alarms_history_full, sync_inventory_full -from .key_alert_config import ( - get_key_alert_monitor_config, - invalidate_key_alert_config_cache, - set_key_alert_monitor_config, -) -from .key_alert_matcher import ( - invalidate_key_alert_rule_cache, - normalize_match_type, - parse_rule_ne_types_payload, - rule_match_type, - rule_match_value, - rule_ne_types, - rule_storage_key, - serialize_rule_ne_types, -) from .runtime_task_messages import ( RT_ALARMS_SYNC_IN_PROGRESS_SKIP, RT_OCLAW_FWD_DISABLED, @@ -185,115 +170,43 @@ def _configure_ume_diag_logging() -> None: lg.propagate = False -def _migrate_key_alert_rule_schema() -> None: - """Evolve ume_key_alert_rule in isolated transactions (psycopg3 rejects unescaped % in SQL).""" - steps = [ - ( - "add match_type", - "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS match_type VARCHAR(32) DEFAULT 'notification_id'", - ), - ( - "add match_value", - "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS match_value VARCHAR(256) DEFAULT ''", - ), - ( - "create monitor_config", - """ - CREATE TABLE IF NOT EXISTS ume_key_alert_monitor_config ( - id INTEGER PRIMARY KEY, - forward_on_clear INTEGER DEFAULT 0, - updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW() - ) - """, - ), - ( - "seed monitor_config", - "INSERT INTO ume_key_alert_monitor_config (id, forward_on_clear, updated_at) " - "VALUES (1, 0, NOW()) ON CONFLICT (id) DO NOTHING", - ), - ( - "backfill match_value from notification_id", - "UPDATE ume_key_alert_rule SET match_value = notification_id " - "WHERE (match_value IS NULL OR match_value = '') " - "AND NOT starts_with(notification_id, 'kw:')", - ), - ( - "backfill keyword rules", - "UPDATE ume_key_alert_rule SET match_type = 'keyword', match_value = SUBSTRING(notification_id FROM 4) " - "WHERE starts_with(notification_id, 'kw:') " - "AND (match_type IS NULL OR match_type = '' OR match_type = 'notification_id')", - ), - ( - "migrate forward_on_clear to global config", - "UPDATE ume_key_alert_monitor_config SET forward_on_clear = 1, updated_at = NOW() " - "WHERE id = 1 AND EXISTS (SELECT 1 FROM ume_key_alert_rule WHERE forward_on_clear = 1)", - ), - ( - "add forward_log rule_key", - "ALTER TABLE ume_key_alert_forward_log ADD COLUMN IF NOT EXISTS rule_key VARCHAR(128) DEFAULT ''", - ), - ( - "add rule ne_types", - "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS ne_types TEXT DEFAULT '[]'", - ), - ] - for label, sql in steps: - try: - with engine.begin() as conn: - conn.exec_driver_sql(sql) - except Exception: - _schedule_log.exception("startup: ume_key_alert_rule schema migration failed at %s", label) - invalidate_key_alert_config_cache() - - @app.on_event("startup") def on_startup() -> None: assert_secure_defaults_or_exit() _configure_ume_diag_logging() Base.metadata.create_all(bind=engine) - skip_ddl = bool(getattr(settings, "skip_legacy_startup_ddl", False)) - if not skip_ddl: - _migrate_key_alert_rule_schema() - # Auth columns must exist before bootstrap / flag_default_password_users. + from .schema_patches import ( + apply_all_legacy_startup_ddl, + apply_auth_schema_patches, + run_alembic_upgrade_to_head, + ) + + alembic_ok = True + if bool(getattr(settings, "alembic_upgrade_on_start", True)): + try: + run_alembic_upgrade_to_head() + except Exception: + alembic_ok = False + _schedule_log.exception("startup: alembic upgrade head failed") + + skip_ddl = bool(getattr(settings, "skip_legacy_startup_ddl", True)) + # Auth columns must exist before bootstrap even when legacy DDL is skipped. try: with engine.begin() as conn: - conn.exec_driver_sql( - "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE" - ) - # JSON works on Postgres/SQLite; create_all also defines ORM column. - try: - conn.exec_driver_sql( - "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'" - ) - except Exception: - pass - conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") - try: - conn.exec_driver_sql( - "ALTER TABLE api_token ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'" - ) - except Exception: - pass - if not skip_ddl: - from .port_traffic_migrate import ensure_port_traffic_series_schema - from .topology_migrate import ensure_topology_schema - - ensure_port_traffic_series_schema(conn) - ensure_topology_schema(conn) - conn.exec_driver_sql( - "ALTER TABLE ne_collection_run ADD COLUMN IF NOT EXISTS ne_source VARCHAR(16) DEFAULT 'managed'" - ) - try: - conn.exec_driver_sql( - "ALTER TABLE ne_collection_run ALTER COLUMN ne_id TYPE VARCHAR(128)" - ) - except Exception: - pass - conn.exec_driver_sql( - "ALTER TABLE config_sync_policy ADD COLUMN IF NOT EXISTS cycle_keep INTEGER DEFAULT 30" - ) + apply_auth_schema_patches(conn) except Exception: - _schedule_log.exception("startup: auth/port_traffic/topology schema migration failed") + _schedule_log.exception("startup: auth schema patches failed") + if skip_ddl and alembic_ok: + _schedule_log.info("startup: schema via Alembic (legacy inline DDL skipped)") + else: + if skip_ddl and not alembic_ok: + _schedule_log.warning( + "startup: Alembic failed — falling back to legacy schema patches" + ) + try: + apply_all_legacy_startup_ddl(engine) + except Exception: + _schedule_log.exception("startup: legacy schema patches failed") ume_support._reset_runtime_pause_flags() ume_support._fail_stale_running_sync_jobs_on_startup() try: @@ -367,218 +280,6 @@ def on_startup() -> None: "startup: inline schedulers disabled — run `python -m netx_api.worker` for " "config_sync / lldp_collect / port_traffic" ) - # Best-effort schema evolution for new columns (no migrations framework). - # Safe for Postgres (IF NOT EXISTS); ignored on failure. - # When NETX_SKIP_LEGACY_STARTUP_DDL=1, rely on Alembic instead. - if bool(getattr(settings, "skip_legacy_startup_ddl", False)): - _schedule_log.info("startup: skipping legacy ALTER TABLE DDL (Alembic mode)") - else: - try: - with engine.begin() as conn: - # Removed from ORM: drop legacy holder table if present (was optional nested UME data). - conn.exec_driver_sql("DROP TABLE IF EXISTS ume_inventory_equipment_holder") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS relevancy VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS l3vpn_peer_ne VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS service VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS affected_client_service_number INTEGER DEFAULT 0" - ) - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS intermittence_count INTEGER DEFAULT 0") - conn.exec_driver_sql("ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS me_level VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_owner VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_expires_at_epoch_s INTEGER DEFAULT 0") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS device_level VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS location VARCHAR(512) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS ipv6_address VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS hardware_version VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS loopback VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS consistent_state VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS interface_version VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS mac VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS admin_status VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS address_type VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS connection_status VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS maintain_status VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS net_mask VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS create_time VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS creator VARCHAR(128) DEFAULT ''") - # Allow long UME alarm fields; avoid StringDataRightTruncation on large payloads. - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN alarm_key TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN object_name TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN event_type TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN native_probable_cause TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN perceived_severity TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN is_cleared TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN time_created TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ALTER COLUMN root_cause_alarm_indication TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_notification_id ON ume_alarms_current (notification_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_notification_id ON ume_alarms_history (notification_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_host_name ON ume_alarms_current (host_name)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_host_name ON ume_alarms_history (host_name)" - ) - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN alarm_key TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN object_name TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN event_type TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN native_probable_cause TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN perceived_severity TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN is_cleared TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN time_created TYPE TEXT") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history ALTER COLUMN root_cause_alarm_indication TYPE TEXT") - # Simplify alarm tables: display fields come from runtime join with inventory table. - conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS ne_name") - conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name") - conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label") - conn.exec_driver_sql("ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") - conn.exec_driver_sql( - "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE" - ) - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_enabled BOOLEAN DEFAULT FALSE") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vendor VARCHAR(32) DEFAULT 'zte'") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_host VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_port INTEGER DEFAULT 22") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_protocol VARCHAR(16) DEFAULT 'ssh'") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_username VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_password_enc TEXT DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_command_template TEXT DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vrf VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql( - "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed'" - ) - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source VARCHAR(64) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source_ref VARCHAR(128) DEFAULT ''") - conn.exec_driver_sql("ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS connect_detail TEXT DEFAULT ''") - # WebCRT sessions may share a host IP; uniqueness is enforced in ne_service for inventory only. - for stmt in ( - "ALTER TABLE managed_ne DROP CONSTRAINT IF EXISTS managed_ne_ip_address_key", - "DROP INDEX IF EXISTS managed_ne_ip_address_key", - "DROP INDEX IF EXISTS ix_managed_ne_ip_address", - "DROP INDEX IF EXISTS sqlite_autoindex_managed_ne_1", - ): - try: - conn.exec_driver_sql(stmt) - except Exception: - pass - try: - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_managed_ne_ip_address ON managed_ne (ip_address)" - ) - except Exception: - pass - conn.exec_driver_sql( - "ALTER TABLE ne_collection_job ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP" - ) - conn.exec_driver_sql( - "UPDATE ne_collection_job SET last_run_at = COALESCE(ended_at, started_at, created_at) " - "WHERE last_run_at IS NULL" - ) - conn.exec_driver_sql( - """ - CREATE TABLE IF NOT EXISTS cli_connect_profile ( - id VARCHAR(64) PRIMARY KEY, - name VARCHAR(256) DEFAULT '', - is_default BOOLEAN DEFAULT FALSE, - username VARCHAR(128) DEFAULT '', - password_enc TEXT DEFAULT '', - port INTEGER DEFAULT 22, - protocol VARCHAR(16) DEFAULT 'ssh', - device_type_default VARCHAR(128) DEFAULT 'zte_zxros', - vendor_default VARCHAR(64) DEFAULT 'ZTE', - ne_type_rules TEXT DEFAULT '', - hop_enabled BOOLEAN DEFAULT FALSE, - hop_vendor VARCHAR(32) DEFAULT 'zte', - hop_host VARCHAR(128) DEFAULT '', - hop_port INTEGER DEFAULT 22, - hop_protocol VARCHAR(16) DEFAULT 'ssh', - hop_username VARCHAR(128) DEFAULT '', - hop_password_enc TEXT DEFAULT '', - hop_command_template TEXT DEFAULT '', - hop_vrf VARCHAR(128) DEFAULT '', - hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed', - created_at TIMESTAMP, - updated_at TIMESTAMP - ) - """ - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_cli_connect_profile_is_default ON cli_connect_profile (is_default)" - ) - conn.exec_driver_sql( - """ - CREATE TABLE IF NOT EXISTS ume_cli_override ( - ume_ne_id VARCHAR(128) PRIMARY KEY, - profile_id VARCHAR(64), - username_override VARCHAR(128) DEFAULT '', - device_type_override VARCHAR(128) DEFAULT '', - vendor_override VARCHAR(64) DEFAULT '', - connect_status VARCHAR(32) DEFAULT 'unknown', - connect_message VARCHAR(512) DEFAULT '', - connect_detail TEXT DEFAULT '', - connect_tested_at TIMESTAMP, - updated_at TIMESTAMP - ) - """ - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_ume_cli_override_connect_status ON ume_cli_override (connect_status)" - ) - conn.exec_driver_sql("COMMENT ON TABLE ume_inventory_ne IS '网元对象详细信息'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_id IS '网元uuid'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_name IS '资源名称'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_type IS '网元类型'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.user_label IS '用户标签'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.address_type IS '管理地址类型(1:IPv4,2:IPv6)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ip_address IS '网元IPv4地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.net_mask IS '管理IPv4掩码(点分十进制)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ipv6_address IS 'IPv6地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.admin_status IS '管理状态(0-离线,1-在线)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.connection_status IS '连接状态(0-断链,1-正常)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.consistent_state IS '数据一致性状态(1一致,2不一致,3冲突)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.maintain_status IS '工程状态(0普通,1调测,2新建)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.vendor IS '网元提供商'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.interface_version IS '网元接口版本号'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.hardware_version IS '硬件版本'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.mac IS '设备机架MAC地址'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.loopback IS '业务环回IP(IPv4)'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.device_level IS '网元层次'") - conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.host_name IS '主机名称'") - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_color VARCHAR(32) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_width INTEGER DEFAULT 0" - ) - conn.exec_driver_sql( - "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS line_style VARCHAR(16) DEFAULT ''" - ) - except Exception: - pass try: if bool(getattr(settings, "ume_keepalive_enabled", True)): interval_keepalive_s = int(getattr(settings, "ume_keepalive_interval_s", 600) or 600) diff --git a/netx_api/schema_patches.py b/netx_api/schema_patches.py new file mode 100644 index 0000000..2671711 --- /dev/null +++ b/netx_api/schema_patches.py @@ -0,0 +1,334 @@ +"""Idempotent brownfield schema patches (shared by API startup and Alembic). + +Historically these lived inline in ``main.on_startup``. Prefer: + +1. ``alembic upgrade head`` (applies this module via revision) +2. ``NETX_SKIP_LEGACY_STARTUP_DDL=true`` so API startup does not re-run the big ALTER block + +``create_all`` still creates missing tables from ORM metadata; these patches add/alter +columns and indexes that ``create_all`` will not evolve on existing DBs. +""" + +from __future__ import annotations + +import logging + +from sqlalchemy import text +from sqlalchemy.engine import Connection, Engine + +_log = logging.getLogger("netx.schema_patches") + + +def _dialect_name(conn: Connection) -> str: + return str(getattr(conn.dialect, "name", "") or "").lower() + + +def _run_sql(conn: Connection, sql: str, *, quiet: bool = True) -> None: + stmt = str(sql or "").strip() + if not stmt: + return + try: + if hasattr(conn, "exec_driver_sql"): + conn.exec_driver_sql(stmt) + else: + conn.execute(text(stmt)) + except Exception: + if quiet: + _log.debug("schema patch skipped/failed: %s", stmt[:120], exc_info=True) + else: + raise + + +def apply_auth_schema_patches(conn: Connection) -> None: + """Columns required before auth bootstrap (safe to run even in Alembic mode).""" + _run_sql( + conn, + "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE", + ) + _run_sql(conn, "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'") + _run_sql(conn, "ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP") + _run_sql(conn, "ALTER TABLE api_token ADD COLUMN IF NOT EXISTS scopes JSON DEFAULT '[]'") + + +def apply_key_alert_schema_patches( + engine: Engine | None = None, + *, + conn: Connection | None = None, +) -> None: + """Evolve ume_key_alert_rule (isolated tx when using engine; in-place when given conn).""" + from .key_alert_config import invalidate_key_alert_config_cache + + steps = [ + ( + "add match_type", + "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS match_type VARCHAR(32) DEFAULT 'notification_id'", + ), + ( + "add match_value", + "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS match_value VARCHAR(256) DEFAULT ''", + ), + ( + "create monitor_config", + """ + CREATE TABLE IF NOT EXISTS ume_key_alert_monitor_config ( + id INTEGER PRIMARY KEY, + forward_on_clear INTEGER DEFAULT 0, + updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW() + ) + """, + ), + ( + "seed monitor_config", + "INSERT INTO ume_key_alert_monitor_config (id, forward_on_clear, updated_at) " + "VALUES (1, 0, NOW()) ON CONFLICT (id) DO NOTHING", + ), + ( + "backfill match_value from notification_id", + "UPDATE ume_key_alert_rule SET match_value = notification_id " + "WHERE (match_value IS NULL OR match_value = '') " + "AND NOT starts_with(notification_id, 'kw:')", + ), + ( + "backfill keyword rules", + "UPDATE ume_key_alert_rule SET match_type = 'keyword', match_value = SUBSTRING(notification_id FROM 4) " + "WHERE starts_with(notification_id, 'kw:') " + "AND (match_type IS NULL OR match_type = '' OR match_type = 'notification_id')", + ), + ( + "migrate forward_on_clear to global config", + "UPDATE ume_key_alert_monitor_config SET forward_on_clear = 1, updated_at = NOW() " + "WHERE id = 1 AND EXISTS (SELECT 1 FROM ume_key_alert_rule WHERE forward_on_clear = 1)", + ), + ( + "add forward_log rule_key", + "ALTER TABLE ume_key_alert_forward_log ADD COLUMN IF NOT EXISTS rule_key VARCHAR(128) DEFAULT ''", + ), + ( + "add rule ne_types", + "ALTER TABLE ume_key_alert_rule ADD COLUMN IF NOT EXISTS ne_types TEXT DEFAULT '[]'", + ), + ] + if conn is not None: + for _label, sql in steps: + _run_sql(conn, sql) + else: + if engine is None: + raise ValueError("engine or conn required") + for label, sql in steps: + try: + with engine.begin() as c: + _run_sql(c, sql, quiet=False) + except Exception: + _log.exception("ume_key_alert_rule schema migration failed at %s", label) + invalidate_key_alert_config_cache() + + +def apply_domain_schema_patches(conn: Connection) -> None: + """Port traffic, topology, collection, config_sync, and historical column evolves.""" + from .port_traffic_migrate import ensure_port_traffic_series_schema + from .topology_migrate import ensure_topology_schema + + try: + ensure_port_traffic_series_schema(conn) + except Exception: + _log.debug("port_traffic schema ensure failed", exc_info=True) + try: + ensure_topology_schema(conn) + except Exception: + _log.debug("topology schema ensure failed", exc_info=True) + + _run_sql( + conn, + "ALTER TABLE ne_collection_run ADD COLUMN IF NOT EXISTS ne_source VARCHAR(16) DEFAULT 'managed'", + ) + _run_sql(conn, "ALTER TABLE ne_collection_run ALTER COLUMN ne_id TYPE VARCHAR(128)") + _run_sql( + conn, + "ALTER TABLE config_sync_policy ADD COLUMN IF NOT EXISTS cycle_keep INTEGER DEFAULT 30", + ) + + pg = _dialect_name(conn).startswith("postgres") + + _run_sql(conn, "DROP TABLE IF EXISTS ume_inventory_equipment_holder") + + for sql in ( + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS relevancy VARCHAR(128) DEFAULT ''", + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS l3vpn_peer_ne VARCHAR(256) DEFAULT ''", + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS service VARCHAR(256) DEFAULT ''", + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS affected_client_service_number INTEGER DEFAULT 0", + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS intermittence_count INTEGER DEFAULT 0", + "ALTER TABLE alarms_norm ADD COLUMN IF NOT EXISTS me_level VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_owner VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_token_cache ADD COLUMN IF NOT EXISTS lock_expires_at_epoch_s INTEGER DEFAULT 0", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS device_level VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS location VARCHAR(512) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS ipv6_address VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS hardware_version VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS loopback VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS consistent_state VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS interface_version VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS mac VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS admin_status VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS address_type VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS connection_status VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS maintain_status VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS net_mask VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS create_time VARCHAR(64) DEFAULT ''", + "ALTER TABLE ume_inventory_ne ADD COLUMN IF NOT EXISTS creator VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''", + "ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS host_name VARCHAR(256) DEFAULT ''", + "ALTER TABLE ume_alarms_current ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''", + "ALTER TABLE ume_alarms_history ADD COLUMN IF NOT EXISTS notification_id VARCHAR(128) DEFAULT ''", + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_notification_id ON ume_alarms_current (notification_id)", + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_notification_id ON ume_alarms_history (notification_id)", + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_current_host_name ON ume_alarms_current (host_name)", + "CREATE INDEX IF NOT EXISTS ix_ume_alarms_history_host_name ON ume_alarms_history (host_name)", + "ALTER TABLE api_token ADD COLUMN IF NOT EXISTS expires_at TIMESTAMP", + "ALTER TABLE app_user ADD COLUMN IF NOT EXISTS must_change_password BOOLEAN DEFAULT FALSE", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_enabled BOOLEAN DEFAULT FALSE", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vendor VARCHAR(32) DEFAULT 'zte'", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_host VARCHAR(128) DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_port INTEGER DEFAULT 22", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_protocol VARCHAR(16) DEFAULT 'ssh'", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_username VARCHAR(128) DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_password_enc TEXT DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_command_template TEXT DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_vrf VARCHAR(128) DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed'", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source VARCHAR(64) DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS source_ref VARCHAR(128) DEFAULT ''", + "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS connect_detail TEXT DEFAULT ''", + "ALTER TABLE ne_collection_job ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP", + "UPDATE ne_collection_job SET last_run_at = COALESCE(ended_at, started_at, created_at) " + "WHERE last_run_at IS NULL", + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_color VARCHAR(32) DEFAULT ''", + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS stroke_width INTEGER DEFAULT 0", + "ALTER TABLE topology_edge ADD COLUMN IF NOT EXISTS line_style VARCHAR(16) DEFAULT ''", + ): + _run_sql(conn, sql) + + if pg: + for sql in ( + "ALTER TABLE ume_alarms_current ALTER COLUMN alarm_key TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN object_name TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN event_type TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN native_probable_cause TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN perceived_severity TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN is_cleared TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN time_created TYPE TEXT", + "ALTER TABLE ume_alarms_current ALTER COLUMN root_cause_alarm_indication TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN alarm_key TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN object_name TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN event_type TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN native_probable_cause TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN perceived_severity TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN is_cleared TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN time_created TYPE TEXT", + "ALTER TABLE ume_alarms_history ALTER COLUMN root_cause_alarm_indication TYPE TEXT", + "ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS ne_name", + "ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label", + "ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name", + "ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label", + "ALTER TABLE managed_ne DROP CONSTRAINT IF EXISTS managed_ne_ip_address_key", + "DROP INDEX IF EXISTS managed_ne_ip_address_key", + "DROP INDEX IF EXISTS ix_managed_ne_ip_address", + "CREATE INDEX IF NOT EXISTS ix_managed_ne_ip_address ON managed_ne (ip_address)", + """ + CREATE TABLE IF NOT EXISTS cli_connect_profile ( + id VARCHAR(64) PRIMARY KEY, + name VARCHAR(256) DEFAULT '', + is_default BOOLEAN DEFAULT FALSE, + username VARCHAR(128) DEFAULT '', + password_enc TEXT DEFAULT '', + port INTEGER DEFAULT 22, + protocol VARCHAR(16) DEFAULT 'ssh', + device_type_default VARCHAR(128) DEFAULT 'zte_zxros', + vendor_default VARCHAR(64) DEFAULT 'ZTE', + ne_type_rules TEXT DEFAULT '', + hop_enabled BOOLEAN DEFAULT FALSE, + hop_vendor VARCHAR(32) DEFAULT 'zte', + hop_host VARCHAR(128) DEFAULT '', + hop_port INTEGER DEFAULT 22, + hop_protocol VARCHAR(16) DEFAULT 'ssh', + hop_username VARCHAR(128) DEFAULT '', + hop_password_enc TEXT DEFAULT '', + hop_command_template TEXT DEFAULT '', + hop_vrf VARCHAR(128) DEFAULT '', + hop_target_auth_mode VARCHAR(32) DEFAULT 'bastion_managed', + created_at TIMESTAMP, + updated_at TIMESTAMP + ) + """, + "CREATE INDEX IF NOT EXISTS ix_cli_connect_profile_is_default ON cli_connect_profile (is_default)", + """ + CREATE TABLE IF NOT EXISTS ume_cli_override ( + ume_ne_id VARCHAR(128) PRIMARY KEY, + profile_id VARCHAR(64), + username_override VARCHAR(128) DEFAULT '', + device_type_override VARCHAR(128) DEFAULT '', + vendor_override VARCHAR(64) DEFAULT '', + connect_status VARCHAR(32) DEFAULT 'unknown', + connect_message VARCHAR(512) DEFAULT '', + connect_detail TEXT DEFAULT '', + connect_tested_at TIMESTAMP, + updated_at TIMESTAMP + ) + """, + "CREATE INDEX IF NOT EXISTS ix_ume_cli_override_connect_status ON ume_cli_override (connect_status)", + "COMMENT ON TABLE ume_inventory_ne IS '网元对象详细信息'", + "COMMENT ON COLUMN ume_inventory_ne.ne_id IS '网元uuid'", + "COMMENT ON COLUMN ume_inventory_ne.ne_name IS '资源名称'", + "COMMENT ON COLUMN ume_inventory_ne.ne_type IS '网元类型'", + "COMMENT ON COLUMN ume_inventory_ne.user_label IS '用户标签'", + "COMMENT ON COLUMN ume_inventory_ne.address_type IS '管理地址类型(1:IPv4,2:IPv6)'", + "COMMENT ON COLUMN ume_inventory_ne.ip_address IS '网元IPv4地址'", + "COMMENT ON COLUMN ume_inventory_ne.net_mask IS '管理IPv4掩码(点分十进制)'", + "COMMENT ON COLUMN ume_inventory_ne.ipv6_address IS 'IPv6地址'", + "COMMENT ON COLUMN ume_inventory_ne.admin_status IS '管理状态(0-离线,1-在线)'", + "COMMENT ON COLUMN ume_inventory_ne.connection_status IS '连接状态(0-断链,1-正常)'", + "COMMENT ON COLUMN ume_inventory_ne.consistent_state IS '数据一致性状态(1一致,2不一致,3冲突)'", + "COMMENT ON COLUMN ume_inventory_ne.maintain_status IS '工程状态(0普通,1调测,2新建)'", + "COMMENT ON COLUMN ume_inventory_ne.vendor IS '网元提供商'", + "COMMENT ON COLUMN ume_inventory_ne.interface_version IS '网元接口版本号'", + "COMMENT ON COLUMN ume_inventory_ne.hardware_version IS '硬件版本'", + "COMMENT ON COLUMN ume_inventory_ne.mac IS '设备机架MAC地址'", + "COMMENT ON COLUMN ume_inventory_ne.loopback IS '业务环回IP(IPv4)'", + "COMMENT ON COLUMN ume_inventory_ne.device_level IS '网元层次'", + "COMMENT ON COLUMN ume_inventory_ne.host_name IS '主机名称'", + ): + _run_sql(conn, sql) + else: + for sql in ( + "DROP INDEX IF EXISTS managed_ne_ip_address_key", + "DROP INDEX IF EXISTS ix_managed_ne_ip_address", + "DROP INDEX IF EXISTS sqlite_autoindex_managed_ne_1", + "CREATE INDEX IF NOT EXISTS ix_managed_ne_ip_address ON managed_ne (ip_address)", + "ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS ne_name", + "ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label", + "ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name", + "ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label", + ): + _run_sql(conn, sql) + + +def apply_all_legacy_startup_ddl(engine: Engine) -> None: + """Full brownfield patch set previously inlined in ``main.on_startup``.""" + apply_key_alert_schema_patches(engine) + with engine.begin() as conn: + apply_auth_schema_patches(conn) + apply_domain_schema_patches(conn) + + +def run_alembic_upgrade_to_head() -> None: + """Programmatic ``alembic upgrade head`` (optional on API start).""" + from pathlib import Path + + from alembic import command + from alembic.config import Config + + root = Path(__file__).resolve().parents[1] + cfg = Config(str(root / "alembic.ini")) + # env.py reads settings.database_url; keep ini placeholder overwritten there. + command.upgrade(cfg, "head") + _log.info("alembic upgrade head completed") diff --git a/tests/test_schema_patches.py b/tests/test_schema_patches.py new file mode 100644 index 0000000..281b817 --- /dev/null +++ b/tests/test_schema_patches.py @@ -0,0 +1,57 @@ +"""Schema patch / Alembic wiring smoke tests.""" + +from __future__ import annotations + +import unittest +from pathlib import Path + +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.orm import sessionmaker + +from netx_api.db import Base +import netx_api.models # noqa: F401 +from netx_api.schema_patches import apply_auth_schema_patches, apply_domain_schema_patches + + +class SchemaPatchesTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(bind=self.engine) + + def tearDown(self) -> None: + self.engine.dispose() + + def test_auth_patches_idempotent(self) -> None: + with self.engine.begin() as conn: + apply_auth_schema_patches(conn) + apply_auth_schema_patches(conn) + insp = inspect(self.engine) + user_cols = {c["name"] for c in insp.get_columns("app_user")} + token_cols = {c["name"] for c in insp.get_columns("api_token")} + self.assertIn("scopes", user_cols) + self.assertIn("must_change_password", user_cols) + self.assertIn("scopes", token_cols) + self.assertIn("expires_at", token_cols) + + def test_domain_patches_do_not_raise(self) -> None: + with self.engine.begin() as conn: + apply_domain_schema_patches(conn) + apply_domain_schema_patches(conn) + + def test_alembic_auto_defaults(self) -> None: + from netx_api.config import Settings + + s = Settings(_env_file=None) + self.assertTrue(s.alembic_upgrade_on_start) + self.assertTrue(s.skip_legacy_startup_ddl) + versions = Path(__file__).resolve().parents[1] / "alembic" / "versions" + files = sorted(p.name for p in versions.glob("*.py") if p.name != "__init__.py") + self.assertIn("20260802_scopes.py", files) + self.assertIn("20260802_legacy_schema.py", files) + text_legacy = (versions / "20260802_legacy_schema.py").read_text(encoding="utf-8") + self.assertIn('down_revision', text_legacy) + self.assertIn("20260802_scopes", text_legacy) + + +if __name__ == "__main__": + unittest.main()