Add periodic engagements and statutory due dates

2026-08-05 12:31:39 +05:30
parent dd2a69dbaa
commit c35c89982e
9 changed files with 415 additions and 94 deletions
@@ -0,0 +1,95 @@
+"""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_client_service_subscription_tenant_client_service_year_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")