A R R R Associates
111 lines
5.5 KiB
Python
111 lines
5.5 KiB
Python
"""phase_2_client_acceptance_controls
|
|
|
|
Revision ID: 20260622_phase_2_client_acceptance_controls
|
|
Revises: 20260621_phase_1_udin_final_documents
|
|
Create Date: 2026-06-22
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy import inspect, text
|
|
|
|
revision = "20260622_phase_2_client_acceptance_controls"
|
|
down_revision = "20260621_phase_1_udin_final_documents"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
CLIENT_TABLE = "clients"
|
|
|
|
|
|
def _columns(table_name: str) -> set[str]:
|
|
bind = op.get_bind()
|
|
inspector = inspect(bind)
|
|
if table_name not in inspector.get_table_names():
|
|
return set()
|
|
return {col["name"] for col in inspector.get_columns(table_name)}
|
|
|
|
|
|
def _add_column_if_missing(existing: set[str], column: sa.Column) -> None:
|
|
if column.name in existing:
|
|
return
|
|
op.add_column(CLIENT_TABLE, column)
|
|
existing.add(column.name)
|
|
|
|
|
|
def _create_index_if_missing(index_name: str, columns: list[str]) -> None:
|
|
bind = op.get_bind()
|
|
inspector = inspect(bind)
|
|
existing_indexes = {idx["name"] for idx in inspector.get_indexes(CLIENT_TABLE)}
|
|
if index_name not in existing_indexes:
|
|
op.create_index(index_name, CLIENT_TABLE, columns)
|
|
|
|
|
|
def upgrade() -> None:
|
|
existing = _columns(CLIENT_TABLE)
|
|
|
|
_add_column_if_missing(existing, sa.Column("acceptance_status", sa.String(length=30), nullable=False, server_default="pending_review"))
|
|
_add_column_if_missing(existing, sa.Column("acceptance_required", sa.Boolean(), nullable=False, server_default=sa.true()))
|
|
_add_column_if_missing(existing, sa.Column("independence_check_completed", sa.Boolean(), nullable=False, server_default=sa.false()))
|
|
_add_column_if_missing(existing, sa.Column("conflict_check_completed", sa.Boolean(), nullable=False, server_default=sa.false()))
|
|
_add_column_if_missing(existing, sa.Column("kyc_completed", sa.Boolean(), nullable=False, server_default=sa.false()))
|
|
_add_column_if_missing(existing, sa.Column("engagement_letter_required", sa.Boolean(), nullable=False, server_default=sa.true()))
|
|
_add_column_if_missing(existing, sa.Column("engagement_letter_received", sa.Boolean(), nullable=False, server_default=sa.false()))
|
|
_add_column_if_missing(existing, sa.Column("acceptance_approved_by_user_id", sa.Integer(), nullable=True))
|
|
_add_column_if_missing(existing, sa.Column("acceptance_approved_at_utc", sa.DateTime(timezone=True), nullable=True))
|
|
_add_column_if_missing(existing, sa.Column("acceptance_review_notes", sa.Text(), nullable=True))
|
|
_add_column_if_missing(existing, sa.Column("acceptance_rejection_reason", sa.Text(), nullable=True))
|
|
|
|
bind = op.get_bind()
|
|
dialect = bind.dialect.name
|
|
if dialect == "postgresql":
|
|
op.execute(text("UPDATE clients SET acceptance_status = 'pending_review' WHERE acceptance_status IS NULL"))
|
|
op.execute(text("UPDATE clients SET acceptance_required = TRUE WHERE acceptance_required IS NULL"))
|
|
op.execute(text("UPDATE clients SET independence_check_completed = FALSE WHERE independence_check_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET conflict_check_completed = FALSE WHERE conflict_check_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET kyc_completed = FALSE WHERE kyc_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET engagement_letter_required = TRUE WHERE engagement_letter_required IS NULL"))
|
|
op.execute(text("UPDATE clients SET engagement_letter_received = FALSE WHERE engagement_letter_received IS NULL"))
|
|
else:
|
|
op.execute(text("UPDATE clients SET acceptance_status = 'pending_review' WHERE acceptance_status IS NULL"))
|
|
op.execute(text("UPDATE clients SET acceptance_required = 1 WHERE acceptance_required IS NULL"))
|
|
op.execute(text("UPDATE clients SET independence_check_completed = 0 WHERE independence_check_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET conflict_check_completed = 0 WHERE conflict_check_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET kyc_completed = 0 WHERE kyc_completed IS NULL"))
|
|
op.execute(text("UPDATE clients SET engagement_letter_required = 1 WHERE engagement_letter_required IS NULL"))
|
|
op.execute(text("UPDATE clients SET engagement_letter_received = 0 WHERE engagement_letter_received IS NULL"))
|
|
|
|
_create_index_if_missing("ix_clients_acceptance_status", ["acceptance_status"])
|
|
_create_index_if_missing("ix_clients_acceptance_approved_by_user_id", ["acceptance_approved_by_user_id"])
|
|
|
|
|
|
def downgrade() -> None:
|
|
bind = op.get_bind()
|
|
inspector = inspect(bind)
|
|
if CLIENT_TABLE not in inspector.get_table_names():
|
|
return
|
|
|
|
existing_indexes = {idx["name"] for idx in inspector.get_indexes(CLIENT_TABLE)}
|
|
if "ix_clients_acceptance_approved_by_user_id" in existing_indexes:
|
|
op.drop_index("ix_clients_acceptance_approved_by_user_id", table_name=CLIENT_TABLE)
|
|
if "ix_clients_acceptance_status" in existing_indexes:
|
|
op.drop_index("ix_clients_acceptance_status", table_name=CLIENT_TABLE)
|
|
|
|
existing = _columns(CLIENT_TABLE)
|
|
for col in [
|
|
"acceptance_rejection_reason",
|
|
"acceptance_review_notes",
|
|
"acceptance_approved_at_utc",
|
|
"acceptance_approved_by_user_id",
|
|
"engagement_letter_received",
|
|
"engagement_letter_required",
|
|
"kyc_completed",
|
|
"conflict_check_completed",
|
|
"independence_check_completed",
|
|
"acceptance_required",
|
|
"acceptance_status",
|
|
]:
|
|
if col in existing:
|
|
op.drop_column(CLIENT_TABLE, col)
|