42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
"""add generic settings hierarchy columns
|
|
|
|
Revision ID: f5a6b7c8d9e0
|
|
Revises: e4f5a6b7c8d9
|
|
Create Date: 2026-06-16 15:15:00.000000
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "f5a6b7c8d9e0"
|
|
down_revision = "e4f5a6b7c8d9"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
_SETTINGS_DEFAULT = sa.text("'{}'")
|
|
|
|
|
|
def _columns(inspector: sa.Inspector, table_name: str) -> set[str]:
|
|
return {column["name"] for column in inspector.get_columns(table_name)}
|
|
|
|
|
|
def upgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
tables = set(inspector.get_table_names())
|
|
for table_name in ("access_users", "access_groups", "campaigns"):
|
|
if table_name not in tables:
|
|
continue
|
|
if "settings" not in _columns(inspector, table_name):
|
|
op.add_column(table_name, sa.Column("settings", sa.JSON(), nullable=False, server_default=_SETTINGS_DEFAULT))
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = sa.inspect(bind)
|
|
tables = set(inspector.get_table_names())
|
|
for table_name in ("campaigns", "access_groups", "access_users"):
|
|
if table_name in tables and "settings" in _columns(inspector, table_name):
|
|
op.drop_column(table_name, "settings")
|