A R R R Associates
126 lines
6.3 KiB
Python
126 lines
6.3 KiB
Python
"""Phase 7H common alerts foundation
|
|
|
|
Revision ID: 20260529_phase_7h_common_alerts
|
|
Revises: af48d99d7321
|
|
Create Date: 2026-05-20
|
|
|
|
This migration is intentionally idempotent for SQLite/dev environments.
|
|
If a previous failed Alembic run already created user_alerts before the
|
|
revision was recorded, re-running `alembic upgrade head` will safely continue.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "20260529_phase_7h_common_alerts"
|
|
down_revision = "af48d99d7321"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
TABLE_NAME = "user_alerts"
|
|
|
|
|
|
def _table_exists(inspector: sa.Inspector, table_name: str) -> bool:
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def _existing_columns(inspector: sa.Inspector, table_name: str) -> set[str]:
|
|
if not _table_exists(inspector, table_name):
|
|
return set()
|
|
return {col["name"] for col in inspector.get_columns(table_name)}
|
|
|
|
|
|
def _existing_indexes(inspector: sa.Inspector, table_name: str) -> set[str]:
|
|
if not _table_exists(inspector, table_name):
|
|
return set()
|
|
return {idx["name"] for idx in inspector.get_indexes(table_name)}
|
|
|
|
|
|
def _add_column_if_missing(inspector: sa.Inspector, column_name: str, column: sa.Column) -> None:
|
|
existing = _existing_columns(inspector, TABLE_NAME)
|
|
if column_name not in existing:
|
|
op.add_column(TABLE_NAME, column)
|
|
|
|
|
|
def _create_index_if_missing(inspector: sa.Inspector, index_name: str, columns: list[str]) -> None:
|
|
existing = _existing_indexes(inspector, TABLE_NAME)
|
|
if index_name not in existing:
|
|
op.create_index(index_name, TABLE_NAME, columns)
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
|
|
if not _table_exists(inspector, TABLE_NAME):
|
|
op.create_table(
|
|
TABLE_NAME,
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
|
|
sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True),
|
|
sa.Column("branch_id", sa.Integer(), sa.ForeignKey("branches.id", ondelete="SET NULL"), nullable=True),
|
|
sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
|
|
sa.Column("role_context", sa.String(length=50), nullable=True),
|
|
sa.Column("alert_type", sa.String(length=80), nullable=False, server_default="general"),
|
|
sa.Column("priority", sa.String(length=20), nullable=False, server_default="normal"),
|
|
sa.Column("title", sa.String(length=255), nullable=False),
|
|
sa.Column("message", sa.Text(), nullable=True),
|
|
sa.Column("target_url", sa.String(length=500), nullable=True),
|
|
sa.Column("is_read", sa.Boolean(), nullable=False, server_default=sa.false()),
|
|
sa.Column("read_at_utc", sa.DateTime(timezone=True), nullable=True),
|
|
sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
|
sa.Column("created_by_user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
|
|
)
|
|
inspector = sa.inspect(bind)
|
|
else:
|
|
# Safety path for a failed/partial migration run. SQLite cannot add
|
|
# foreign-key constraints after table creation, but it can add any
|
|
# missing columns so the model and UI continue to work.
|
|
_add_column_if_missing(inspector, "tenant_id", sa.Column("tenant_id", sa.Integer(), nullable=True))
|
|
_add_column_if_missing(inspector, "branch_id", sa.Column("branch_id", sa.Integer(), nullable=True))
|
|
_add_column_if_missing(inspector, "user_id", sa.Column("user_id", sa.Integer(), nullable=False, server_default="0"))
|
|
_add_column_if_missing(inspector, "role_context", sa.Column("role_context", sa.String(length=50), nullable=True))
|
|
_add_column_if_missing(inspector, "alert_type", sa.Column("alert_type", sa.String(length=80), nullable=False, server_default="general"))
|
|
_add_column_if_missing(inspector, "priority", sa.Column("priority", sa.String(length=20), nullable=False, server_default="normal"))
|
|
_add_column_if_missing(inspector, "title", sa.Column("title", sa.String(length=255), nullable=False, server_default="Alert"))
|
|
_add_column_if_missing(inspector, "message", sa.Column("message", sa.Text(), nullable=True))
|
|
_add_column_if_missing(inspector, "target_url", sa.Column("target_url", sa.String(length=500), nullable=True))
|
|
_add_column_if_missing(inspector, "is_read", sa.Column("is_read", sa.Boolean(), nullable=False, server_default=sa.false()))
|
|
_add_column_if_missing(inspector, "read_at_utc", sa.Column("read_at_utc", sa.DateTime(timezone=True), nullable=True))
|
|
_add_column_if_missing(inspector, "created_at_utc", sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()))
|
|
_add_column_if_missing(inspector, "created_by_user_id", sa.Column("created_by_user_id", sa.Integer(), nullable=True))
|
|
inspector = sa.inspect(bind)
|
|
|
|
_create_index_if_missing(inspector, "ix_user_alerts_tenant_id", ["tenant_id"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_branch_id", ["branch_id"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_user_id", ["user_id"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_role_context", ["role_context"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_alert_type", ["alert_type"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_priority", ["priority"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_is_read", ["is_read"])
|
|
_create_index_if_missing(inspector, "ix_user_alerts_created_at_utc", ["created_at_utc"])
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
if not _table_exists(inspector, TABLE_NAME):
|
|
return
|
|
|
|
existing_indexes = _existing_indexes(inspector, TABLE_NAME)
|
|
for index_name in [
|
|
"ix_user_alerts_created_at_utc",
|
|
"ix_user_alerts_is_read",
|
|
"ix_user_alerts_priority",
|
|
"ix_user_alerts_alert_type",
|
|
"ix_user_alerts_role_context",
|
|
"ix_user_alerts_user_id",
|
|
"ix_user_alerts_branch_id",
|
|
"ix_user_alerts_tenant_id",
|
|
]:
|
|
if index_name in existing_indexes:
|
|
op.drop_index(index_name, table_name=TABLE_NAME)
|
|
|
|
op.drop_table(TABLE_NAME)
|