A R R R Associates
108 lines
4.8 KiB
Python
108 lines
4.8 KiB
Python
"""Phase v2.0.4-D - Billing financial year isolation
|
|
|
|
Revision ID: 20260618_phase_204d_billing_fy
|
|
Revises: 20260617_phase_204a_financial_year
|
|
Create Date: 2026-06-18
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
revision = "20260618_phase_204d_billing_fy"
|
|
down_revision = "20260617_phase_204a_financial_year"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _fy_label(value) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
value = value.date()
|
|
if isinstance(value, str):
|
|
try:
|
|
value = date.fromisoformat(value[:10])
|
|
except Exception:
|
|
return None
|
|
if not isinstance(value, date):
|
|
return None
|
|
start_year = value.year if value.month >= 4 else value.year - 1
|
|
return f"{start_year}-{str(start_year + 1)[-2:]}"
|
|
|
|
|
|
def _has_table(conn, table_name: str) -> bool:
|
|
return sa.inspect(conn).has_table(table_name)
|
|
|
|
|
|
def _has_column(conn, table_name: str, column_name: str) -> bool:
|
|
if not _has_table(conn, table_name):
|
|
return False
|
|
return any(col["name"] == column_name for col in sa.inspect(conn).get_columns(table_name))
|
|
|
|
|
|
def _create_index_if_missing(conn, index_name: str, table_name: str, columns: list[str]) -> None:
|
|
if not _has_table(conn, table_name):
|
|
return
|
|
existing = {idx["name"] for idx in sa.inspect(conn).get_indexes(table_name)}
|
|
if index_name not in existing:
|
|
op.create_index(index_name, table_name, columns)
|
|
|
|
|
|
def _drop_index_if_exists(conn, index_name: str, table_name: str) -> None:
|
|
if not _has_table(conn, table_name):
|
|
return
|
|
existing = {idx["name"] for idx in sa.inspect(conn).get_indexes(table_name)}
|
|
if index_name in existing:
|
|
op.drop_index(index_name, table_name=table_name)
|
|
|
|
|
|
def upgrade() -> None:
|
|
conn = op.get_bind()
|
|
|
|
if _has_table(conn, "billing_invoice_generation_batches") and not _has_column(conn, "billing_invoice_generation_batches", "financial_year"):
|
|
op.add_column("billing_invoice_generation_batches", sa.Column("financial_year", sa.String(length=20), nullable=True))
|
|
rows = conn.execute(sa.text("SELECT id, billing_period_from FROM billing_invoice_generation_batches")).mappings().all()
|
|
for row in rows:
|
|
fy = _fy_label(row["billing_period_from"])
|
|
if fy:
|
|
conn.execute(sa.text("UPDATE billing_invoice_generation_batches SET financial_year = :fy WHERE id = :id"), {"fy": fy, "id": row["id"]})
|
|
_create_index_if_missing(conn, "ix_billing_invoice_generation_batches_financial_year", "billing_invoice_generation_batches", ["financial_year"])
|
|
|
|
if _has_table(conn, "billing_invoices") and not _has_column(conn, "billing_invoices", "financial_year"):
|
|
op.add_column("billing_invoices", sa.Column("financial_year", sa.String(length=20), nullable=True))
|
|
rows = conn.execute(sa.text("SELECT id, billing_period_from, invoice_date FROM billing_invoices")).mappings().all()
|
|
for row in rows:
|
|
fy = _fy_label(row["billing_period_from"]) or _fy_label(row["invoice_date"])
|
|
if fy:
|
|
conn.execute(sa.text("UPDATE billing_invoices SET financial_year = :fy WHERE id = :id"), {"fy": fy, "id": row["id"]})
|
|
_create_index_if_missing(conn, "ix_billing_invoices_financial_year", "billing_invoices", ["financial_year"])
|
|
|
|
if _has_table(conn, "billing_payments") and not _has_column(conn, "billing_payments", "financial_year"):
|
|
op.add_column("billing_payments", sa.Column("financial_year", sa.String(length=20), nullable=True))
|
|
rows = conn.execute(sa.text("SELECT id, payment_date FROM billing_payments")).mappings().all()
|
|
for row in rows:
|
|
fy = _fy_label(row["payment_date"])
|
|
if fy:
|
|
conn.execute(sa.text("UPDATE billing_payments SET financial_year = :fy WHERE id = :id"), {"fy": fy, "id": row["id"]})
|
|
_create_index_if_missing(conn, "ix_billing_payments_financial_year", "billing_payments", ["financial_year"])
|
|
|
|
|
|
def downgrade() -> None:
|
|
conn = op.get_bind()
|
|
|
|
if _has_column(conn, "billing_payments", "financial_year"):
|
|
_drop_index_if_exists(conn, "ix_billing_payments_financial_year", "billing_payments")
|
|
op.drop_column("billing_payments", "financial_year")
|
|
|
|
if _has_column(conn, "billing_invoices", "financial_year"):
|
|
_drop_index_if_exists(conn, "ix_billing_invoices_financial_year", "billing_invoices")
|
|
op.drop_column("billing_invoices", "financial_year")
|
|
|
|
if _has_column(conn, "billing_invoice_generation_batches", "financial_year"):
|
|
_drop_index_if_exists(conn, "ix_billing_invoice_generation_batches_financial_year", "billing_invoice_generation_batches")
|
|
op.drop_column("billing_invoice_generation_batches", "financial_year")
|