Files
arrr-erp/alembic/versions/20260609_phase_7s1c_email_preferences.py
T
2026-06-20 15:01:44 +05:30

51 lines
2.1 KiB
Python

"""Phase 7S.1C email notification preferences
Revision ID: 20260609_phase_7s1c_email_preferences
Revises: 20260608_phase_7s1_email_notifications
Create Date: 2026-06-09
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "20260609_phase_7s1c_email_preferences"
down_revision = "20260608_phase_7s1_email_notifications"
branch_labels = None
depends_on = None
_NEW_COLUMNS: list[tuple[str, sa.Column]] = [
("send_invoice_emails", sa.Column("send_invoice_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_payment_emails", sa.Column("send_payment_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_client_emails", sa.Column("send_client_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_document_emails", sa.Column("send_document_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_consultant_emails", sa.Column("send_consultant_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_partner_review_emails", sa.Column("send_partner_review_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_leave_attendance_emails", sa.Column("send_leave_attendance_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
("send_online_payment_emails", sa.Column("send_online_payment_emails", sa.Boolean(), nullable=False, server_default=sa.true())),
]
def _existing_columns(table_name: str) -> set[str]:
bind = op.get_bind()
inspector = sa.inspect(bind)
return {col["name"] for col in inspector.get_columns(table_name)}
def upgrade() -> None:
existing = _existing_columns("email_settings")
with op.batch_alter_table("email_settings") as batch_op:
for name, column in _NEW_COLUMNS:
if name not in existing:
batch_op.add_column(column)
def downgrade() -> None:
existing = _existing_columns("email_settings")
with op.batch_alter_table("email_settings") as batch_op:
for name, _column in reversed(_NEW_COLUMNS):
if name in existing:
batch_op.drop_column(name)