A R R R Associates
96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
"""add recurring engagement period support
|
|
|
|
Revision ID: 20260805_periodic_engagements
|
|
Revises: 20260805_performing_partner
|
|
"""
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy import inspect
|
|
|
|
revision = "20260805_periodic_engagements"
|
|
down_revision = "20260805_performing_partner"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
TABLE = "client_service_subscriptions"
|
|
OLD_UQ = "uq_client_service_subscription_tenant_client_service_year"
|
|
NEW_UQ = "uq_css_tenant_client_service_fy_period"
|
|
INDEX = "ix_client_service_subscriptions_period_label"
|
|
|
|
|
|
def _inspector():
|
|
return inspect(op.get_bind())
|
|
|
|
|
|
def _columns() -> set[str]:
|
|
return {row["name"] for row in _inspector().get_columns(TABLE)}
|
|
|
|
|
|
def _unique_constraints() -> set[str]:
|
|
return {row.get("name") for row in _inspector().get_unique_constraints(TABLE) if row.get("name")}
|
|
|
|
|
|
def _indexes() -> set[str]:
|
|
return {row.get("name") for row in _inspector().get_indexes(TABLE) if row.get("name")}
|
|
|
|
|
|
def upgrade():
|
|
if "period_label" not in _columns():
|
|
op.add_column(TABLE, sa.Column("period_label", sa.String(length=30), nullable=False, server_default=""))
|
|
|
|
op.execute("UPDATE client_service_subscriptions SET period_label = '' WHERE period_label IS NULL")
|
|
|
|
uniques = _unique_constraints()
|
|
dialect = op.get_bind().dialect.name
|
|
if dialect == "sqlite":
|
|
with op.batch_alter_table(TABLE, recreate="always") as batch:
|
|
if OLD_UQ in uniques:
|
|
batch.drop_constraint(OLD_UQ, type_="unique")
|
|
if NEW_UQ not in uniques:
|
|
batch.create_unique_constraint(
|
|
NEW_UQ,
|
|
["tenant_id", "client_id", "service_catalogue_id", "financial_year", "period_label"],
|
|
)
|
|
else:
|
|
if OLD_UQ in uniques:
|
|
op.drop_constraint(OLD_UQ, TABLE, type_="unique")
|
|
if NEW_UQ not in _unique_constraints():
|
|
op.create_unique_constraint(
|
|
NEW_UQ,
|
|
TABLE,
|
|
["tenant_id", "client_id", "service_catalogue_id", "financial_year", "period_label"],
|
|
)
|
|
|
|
if INDEX not in _indexes():
|
|
op.create_index(INDEX, TABLE, ["period_label"], unique=False)
|
|
|
|
|
|
def downgrade():
|
|
if INDEX in _indexes():
|
|
op.drop_index(INDEX, table_name=TABLE)
|
|
|
|
uniques = _unique_constraints()
|
|
dialect = op.get_bind().dialect.name
|
|
if dialect == "sqlite":
|
|
with op.batch_alter_table(TABLE, recreate="always") as batch:
|
|
if NEW_UQ in uniques:
|
|
batch.drop_constraint(NEW_UQ, type_="unique")
|
|
if OLD_UQ not in uniques:
|
|
batch.create_unique_constraint(
|
|
OLD_UQ,
|
|
["tenant_id", "client_id", "service_catalogue_id", "financial_year"],
|
|
)
|
|
if "period_label" in _columns():
|
|
batch.drop_column("period_label")
|
|
else:
|
|
if NEW_UQ in uniques:
|
|
op.drop_constraint(NEW_UQ, TABLE, type_="unique")
|
|
if OLD_UQ not in _unique_constraints():
|
|
op.create_unique_constraint(
|
|
OLD_UQ,
|
|
TABLE,
|
|
["tenant_id", "client_id", "service_catalogue_id", "financial_year"],
|
|
)
|
|
if "period_label" in _columns():
|
|
op.drop_column(TABLE, "period_label")
|