A R R R Associates
143 lines
7.8 KiB
Python
143 lines
7.8 KiB
Python
"""services phase 4A due date engine foundation
|
|
|
|
Revision ID: 20260504_due_date_engine_foundation
|
|
Revises: 20260503_assurance_review_partner
|
|
Create Date: 2026-05-04
|
|
"""
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision: str = "20260504_due_date_engine_foundation"
|
|
down_revision: Union[str, None] = "20260503_assurance_review_partner"
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def _insp():
|
|
return sa.inspect(op.get_bind())
|
|
|
|
|
|
def _has_table(table_name: str) -> bool:
|
|
return table_name in _insp().get_table_names()
|
|
|
|
|
|
def _has_column(table_name: str, column_name: str) -> bool:
|
|
if not _has_table(table_name):
|
|
return False
|
|
return any(c["name"] == column_name for c in _insp().get_columns(table_name))
|
|
|
|
|
|
def _has_index(table_name: str, index_name: str) -> bool:
|
|
if not _has_table(table_name):
|
|
return False
|
|
return any(ix.get("name") == index_name for ix in _insp().get_indexes(table_name))
|
|
|
|
|
|
def _create_index_if_missing(table_name: str, index_name: str, columns: list[str]) -> None:
|
|
if _has_table(table_name) and not _has_index(table_name, index_name):
|
|
op.create_index(index_name, table_name, columns)
|
|
|
|
|
|
def upgrade() -> None:
|
|
if not _has_table("service_due_date_rules"):
|
|
op.create_table(
|
|
"service_due_date_rules",
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
|
|
sa.Column("service_catalogue_id", sa.Integer(), nullable=False),
|
|
sa.Column("rule_name", sa.String(length=150), nullable=False),
|
|
sa.Column("period_type", sa.String(length=30), nullable=False, server_default="yearly"),
|
|
sa.Column("due_year_basis", sa.String(length=40), nullable=False, server_default="assessment_year_start"),
|
|
sa.Column("due_day", sa.Integer(), nullable=True),
|
|
sa.Column("due_month", sa.Integer(), nullable=True),
|
|
sa.Column("due_month_offset", sa.Integer(), nullable=False, server_default="0"),
|
|
sa.Column("days_offset_after_event", sa.Integer(), nullable=True),
|
|
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="100"),
|
|
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
|
|
sa.Column("remarks", sa.Text(), nullable=True),
|
|
sa.Column("created_by_user_id", sa.Integer(), nullable=True),
|
|
sa.Column("updated_by_user_id", sa.Integer(), nullable=True),
|
|
sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
|
sa.Column("updated_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
|
sa.ForeignKeyConstraint(["service_catalogue_id"], ["service_catalogues.id"], ondelete="CASCADE"),
|
|
sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"]),
|
|
sa.ForeignKeyConstraint(["updated_by_user_id"], ["users.id"]),
|
|
sa.UniqueConstraint("service_catalogue_id", "rule_name", name="uq_service_due_date_rules_catalogue_name"),
|
|
)
|
|
_create_index_if_missing("service_due_date_rules", "ix_service_due_date_rules_service_catalogue_id", ["service_catalogue_id"])
|
|
_create_index_if_missing("service_due_date_rules", "ix_service_due_date_rules_period_type", ["period_type"])
|
|
_create_index_if_missing("service_due_date_rules", "ix_service_due_date_rules_is_active", ["is_active"])
|
|
|
|
if not _has_table("service_due_date_extensions"):
|
|
op.create_table(
|
|
"service_due_date_extensions",
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
|
|
sa.Column("tenant_id", sa.Integer(), nullable=False),
|
|
sa.Column("service_catalogue_id", sa.Integer(), nullable=False),
|
|
sa.Column("due_date_rule_id", sa.Integer(), nullable=True),
|
|
sa.Column("financial_year", sa.String(length=9), nullable=False),
|
|
sa.Column("assessment_year", sa.String(length=9), nullable=True),
|
|
sa.Column("period_label", sa.String(length=30), nullable=True),
|
|
sa.Column("previous_due_date", sa.Date(), nullable=True),
|
|
sa.Column("extended_due_date", sa.Date(), nullable=False),
|
|
sa.Column("extension_sequence", sa.Integer(), nullable=False, server_default="1"),
|
|
sa.Column("notification_reference", sa.String(length=200), nullable=True),
|
|
sa.Column("notification_date", sa.Date(), nullable=True),
|
|
sa.Column("remarks", sa.Text(), nullable=True),
|
|
sa.Column("created_by_user_id", sa.Integer(), nullable=True),
|
|
sa.Column("updated_by_user_id", sa.Integer(), nullable=True),
|
|
sa.Column("created_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
|
sa.Column("updated_at_utc", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
|
sa.ForeignKeyConstraint(["tenant_id"], ["tenants.id"], ondelete="CASCADE"),
|
|
sa.ForeignKeyConstraint(["service_catalogue_id"], ["service_catalogues.id"], ondelete="CASCADE"),
|
|
sa.ForeignKeyConstraint(["due_date_rule_id"], ["service_due_date_rules.id"], ondelete="SET NULL"),
|
|
sa.ForeignKeyConstraint(["created_by_user_id"], ["users.id"]),
|
|
sa.ForeignKeyConstraint(["updated_by_user_id"], ["users.id"]),
|
|
)
|
|
for index_name, columns in [
|
|
("ix_service_due_date_extensions_tenant_id", ["tenant_id"]),
|
|
("ix_service_due_date_extensions_service_catalogue_id", ["service_catalogue_id"]),
|
|
("ix_service_due_date_extensions_due_date_rule_id", ["due_date_rule_id"]),
|
|
("ix_service_due_date_extensions_financial_year", ["financial_year"]),
|
|
("ix_service_due_date_extensions_assessment_year", ["assessment_year"]),
|
|
("ix_service_due_date_extensions_period_label", ["period_label"]),
|
|
("ix_service_due_date_extensions_extended_due_date", ["extended_due_date"]),
|
|
]:
|
|
_create_index_if_missing("service_due_date_extensions", index_name, columns)
|
|
|
|
subscription_columns = [
|
|
("due_date_rule_id", sa.Column("due_date_rule_id", sa.Integer(), nullable=True)),
|
|
("original_due_date", sa.Column("original_due_date", sa.Date(), nullable=True)),
|
|
("current_due_date", sa.Column("current_due_date", sa.Date(), nullable=True)),
|
|
("due_date_source", sa.Column("due_date_source", sa.String(length=30), nullable=True)),
|
|
]
|
|
for column_name, column in subscription_columns:
|
|
if not _has_column("client_service_subscriptions", column_name):
|
|
op.add_column("client_service_subscriptions", column)
|
|
|
|
for index_name, columns in [
|
|
("ix_client_service_subscriptions_due_date_rule_id", ["due_date_rule_id"]),
|
|
("ix_client_service_subscriptions_original_due_date", ["original_due_date"]),
|
|
("ix_client_service_subscriptions_current_due_date", ["current_due_date"]),
|
|
]:
|
|
_create_index_if_missing("client_service_subscriptions", index_name, columns)
|
|
|
|
|
|
def downgrade() -> None:
|
|
for index_name in [
|
|
"ix_client_service_subscriptions_current_due_date",
|
|
"ix_client_service_subscriptions_original_due_date",
|
|
"ix_client_service_subscriptions_due_date_rule_id",
|
|
]:
|
|
if _has_index("client_service_subscriptions", index_name):
|
|
op.drop_index(index_name, table_name="client_service_subscriptions")
|
|
for column_name in ["due_date_source", "current_due_date", "original_due_date", "due_date_rule_id"]:
|
|
if _has_column("client_service_subscriptions", column_name):
|
|
op.drop_column("client_service_subscriptions", column_name)
|
|
if _has_table("service_due_date_extensions"):
|
|
op.drop_table("service_due_date_extensions")
|
|
if _has_table("service_due_date_rules"):
|
|
op.drop_table("service_due_date_rules")
|