A R R R Associates
98 lines
4.7 KiB
Python
98 lines
4.7 KiB
Python
"""Phase 7R.3 payment tracking and receipts
|
|
|
|
Revision ID: 20260603_phase_7r3_payments
|
|
Revises: 20260602_phase_7r2_gst_invoice
|
|
Create Date: 2026-06-03
|
|
|
|
Adds invoice collection totals and a payment/receipt ledger. The migration
|
|
is intentionally idempotent for SQLite development databases used in this
|
|
project.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
revision = "20260603_phase_7r3_payments"
|
|
down_revision = "20260602_phase_7r2_gst_invoice"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def _tables() -> set[str]:
|
|
return set(sa.inspect(op.get_bind()).get_table_names())
|
|
|
|
|
|
def _columns(table_name: str) -> set[str]:
|
|
inspector = sa.inspect(op.get_bind())
|
|
if table_name not in inspector.get_table_names():
|
|
return set()
|
|
return {c["name"] for c in inspector.get_columns(table_name)}
|
|
|
|
|
|
def _add_if_missing(table: str, column: sa.Column) -> None:
|
|
if column.name not in _columns(table):
|
|
op.add_column(table, column)
|
|
|
|
|
|
def _index_exists(index_name: str) -> bool:
|
|
inspector = sa.inspect(op.get_bind())
|
|
for table_name in inspector.get_table_names():
|
|
for idx in inspector.get_indexes(table_name):
|
|
if idx.get("name") == index_name:
|
|
return True
|
|
return False
|
|
|
|
|
|
def upgrade() -> None:
|
|
_add_if_missing("billing_invoices", sa.Column("amount_received", sa.Numeric(14, 2), nullable=False, server_default="0.00"))
|
|
_add_if_missing("billing_invoices", sa.Column("tds_deducted", sa.Numeric(14, 2), nullable=False, server_default="0.00"))
|
|
_add_if_missing("billing_invoices", sa.Column("bank_charges", sa.Numeric(14, 2), nullable=False, server_default="0.00"))
|
|
_add_if_missing("billing_invoices", sa.Column("balance_amount", sa.Numeric(14, 2), nullable=False, server_default="0.00"))
|
|
|
|
if "billing_payments" not in _tables():
|
|
op.create_table(
|
|
"billing_payments",
|
|
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
|
|
sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False),
|
|
sa.Column("branch_id", sa.Integer(), sa.ForeignKey("branches.id", ondelete="SET NULL"), nullable=True),
|
|
sa.Column("invoice_id", sa.Integer(), sa.ForeignKey("billing_invoices.id", ondelete="CASCADE"), nullable=False),
|
|
sa.Column("client_id", sa.Integer(), sa.ForeignKey("clients.id", ondelete="RESTRICT"), nullable=False),
|
|
sa.Column("receipt_no", sa.String(length=60), nullable=False),
|
|
sa.Column("receipt_date", sa.Date(), nullable=False),
|
|
sa.Column("payment_date", sa.Date(), nullable=False),
|
|
sa.Column("amount_received", sa.Numeric(14, 2), nullable=False, server_default="0.00"),
|
|
sa.Column("tds_deducted", sa.Numeric(14, 2), nullable=False, server_default="0.00"),
|
|
sa.Column("bank_charges", sa.Numeric(14, 2), nullable=False, server_default="0.00"),
|
|
sa.Column("mode", sa.String(length=30), nullable=False, server_default="BANK"),
|
|
sa.Column("reference_no", sa.String(length=120), nullable=True),
|
|
sa.Column("payment_gateway", sa.String(length=50), nullable=True),
|
|
sa.Column("gateway_transaction_id", sa.String(length=120), nullable=True),
|
|
sa.Column("remarks", sa.Text(), nullable=True),
|
|
sa.Column("status", sa.String(length=20), nullable=False, server_default="RECEIVED"),
|
|
sa.Column("created_by_user_id", sa.Integer(), sa.ForeignKey("users.id"), 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.UniqueConstraint("tenant_id", "receipt_no", name="uq_billing_payments_tenant_receipt_no"),
|
|
)
|
|
for name, cols in {
|
|
"ix_billing_payments_tenant_id": ["tenant_id"],
|
|
"ix_billing_payments_branch_id": ["branch_id"],
|
|
"ix_billing_payments_invoice_id": ["invoice_id"],
|
|
"ix_billing_payments_client_id": ["client_id"],
|
|
"ix_billing_payments_receipt_no": ["receipt_no"],
|
|
"ix_billing_payments_payment_date": ["payment_date"],
|
|
"ix_billing_payments_status": ["status"],
|
|
}.items():
|
|
if not _index_exists(name):
|
|
op.create_index(name, "billing_payments", cols)
|
|
|
|
bind = op.get_bind()
|
|
if "billing_invoices" in _tables():
|
|
bind.execute(sa.text("UPDATE billing_invoices SET balance_amount = COALESCE(total_amount, 0) WHERE COALESCE(balance_amount, 0) = 0 AND COALESCE(amount_received, 0) = 0 AND COALESCE(tds_deducted, 0) = 0"))
|
|
|
|
|
|
def downgrade() -> None:
|
|
# No-op downgrade for SQLite/dev safety.
|
|
pass
|