"""add invoice activities and reminders Revision ID: 20260309001a Revises: d4e5f6a7b8c9 Create Date: 2026-03-09 00:00:00.000000 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy import inspect revision: str = '20260309001a' down_revision: Union[str, Sequence[str], None] = 'd4e5f6a7b8c9' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # Helper to check if column exists conn = op.get_bind() inspector = inspect(conn) reminders_columns = {col['name'] for col in inspector.get_columns('reminders')} # ── 1. Extend existing `reminders` table ──────────────────────────────── # Only add columns if they don't exist if 'reminder_id' not in reminders_columns: op.add_column('reminders', sa.Column('reminder_id', sa.String(64), nullable=True)) if 'invoice_id' not in reminders_columns: op.add_column('reminders', sa.Column('invoice_id', sa.Integer(), nullable=True)) if 'preset' not in reminders_columns: op.add_column('reminders', sa.Column('preset', sa.String(32), nullable=True)) if 'scheduled_at' not in reminders_columns: op.add_column('reminders', sa.Column('scheduled_at', sa.DateTime(timezone=True), nullable=True)) if 'channel' not in reminders_columns: op.add_column('reminders', sa.Column('channel', sa.String(10), nullable=True)) if 'reminder_status' not in reminders_columns: op.add_column( 'reminders', sa.Column('reminder_status', sa.String(20), nullable=True, server_default='pending'), ) if 'sent_at' not in reminders_columns: op.add_column('reminders', sa.Column('sent_at', sa.DateTime(timezone=True), nullable=True)) if 'failure_reason' not in reminders_columns: op.add_column('reminders', sa.Column('failure_reason', sa.Text(), nullable=True)) if 'merchant_id' not in reminders_columns: op.add_column('reminders', sa.Column('merchant_id', sa.Integer(), nullable=True)) # Check and create constraints/indexes only if they don't exist reminders_indexes = {idx['name'] for idx in inspector.get_indexes('reminders')} reminders_constraints = {con['name'] for con in inspector.get_unique_constraints('reminders')} if 'uq_reminders_reminder_id' not in reminders_constraints: op.create_unique_constraint('uq_reminders_reminder_id', 'reminders', ['reminder_id']) if 'ix_reminders_reminder_id' not in reminders_indexes: op.create_index('ix_reminders_reminder_id', 'reminders', ['reminder_id']) if 'ix_reminders_invoice_id' not in reminders_indexes: op.create_index('ix_reminders_invoice_id', 'reminders', ['invoice_id']) if 'ix_reminders_scheduled_at' not in reminders_indexes: op.create_index('ix_reminders_scheduled_at', 'reminders', ['scheduled_at']) if 'ix_reminders_reminder_status' not in reminders_indexes: op.create_index('ix_reminders_reminder_status', 'reminders', ['reminder_status']) if 'ix_reminders_merchant_id' not in reminders_indexes: op.create_index('ix_reminders_merchant_id', 'reminders', ['merchant_id']) if 'ix_reminders_status_scheduled_at' not in reminders_indexes: op.create_index( 'ix_reminders_status_scheduled_at', 'reminders', ['reminder_status', 'scheduled_at'], ) # Check and create foreign keys reminders_fks = {fk['name'] for fk in inspector.get_foreign_keys('reminders')} if 'fk_reminders_invoice_id' not in reminders_fks: op.create_foreign_key( 'fk_reminders_invoice_id', 'reminders', 'invoices', ['invoice_id'], ['id'], ) if 'fk_reminders_merchant_id' not in reminders_fks: op.create_foreign_key( 'fk_reminders_merchant_id', 'reminders', 'merchants', ['merchant_id'], ['id'], ) # ── 2. Create `invoice_activities` table ──────────────────────────────── # Check if table exists existing_tables = inspector.get_table_names() if 'invoice_activities' not in existing_tables: op.create_table( 'invoice_activities', sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('invoice_id', sa.Integer(), nullable=False), sa.Column('activity_type', sa.String(64), nullable=False), sa.Column('description', sa.Text(), nullable=True), sa.Column('actor_type', sa.String(20), nullable=True), sa.Column('actor_id', sa.Integer(), nullable=True), sa.Column('metadata', sa.JSON(), nullable=True), sa.Column( 'created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False, ), sa.ForeignKeyConstraint(['invoice_id'], ['invoices.id'], name='fk_invoice_activities_invoice_id'), sa.PrimaryKeyConstraint('id'), ) op.create_index('ix_invoice_activities_id', 'invoice_activities', ['id']) op.create_index('ix_invoice_activities_invoice_id', 'invoice_activities', ['invoice_id']) op.create_index('ix_invoice_activities_activity_type', 'invoice_activities', ['activity_type']) op.create_index('ix_invoice_activities_created_at', 'invoice_activities', ['created_at']) op.create_index( 'ix_invoice_activities_invoice_id_created_at', 'invoice_activities', ['invoice_id', 'created_at'], ) def downgrade() -> None: # Helper to check what exists conn = op.get_bind() inspector = inspect(conn) existing_tables = inspector.get_table_names() # Drop invoice_activities if it exists if 'invoice_activities' in existing_tables: invoice_activities_indexes = {idx['name'] for idx in inspector.get_indexes('invoice_activities')} if 'ix_invoice_activities_invoice_id_created_at' in invoice_activities_indexes: op.drop_index('ix_invoice_activities_invoice_id_created_at', table_name='invoice_activities') if 'ix_invoice_activities_created_at' in invoice_activities_indexes: op.drop_index('ix_invoice_activities_created_at', table_name='invoice_activities') if 'ix_invoice_activities_activity_type' in invoice_activities_indexes: op.drop_index('ix_invoice_activities_activity_type', table_name='invoice_activities') if 'ix_invoice_activities_invoice_id' in invoice_activities_indexes: op.drop_index('ix_invoice_activities_invoice_id', table_name='invoice_activities') if 'ix_invoice_activities_id' in invoice_activities_indexes: op.drop_index('ix_invoice_activities_id', table_name='invoice_activities') op.drop_table('invoice_activities') # Drop reminders extensions if they exist if 'reminders' in existing_tables: reminders_fks = {fk['name'] for fk in inspector.get_foreign_keys('reminders')} reminders_indexes = {idx['name'] for idx in inspector.get_indexes('reminders')} reminders_constraints = {con['name'] for con in inspector.get_unique_constraints('reminders')} reminders_columns = {col['name'] for col in inspector.get_columns('reminders')} if 'fk_reminders_merchant_id' in reminders_fks: op.drop_constraint('fk_reminders_merchant_id', 'reminders', type_='foreignkey') if 'fk_reminders_invoice_id' in reminders_fks: op.drop_constraint('fk_reminders_invoice_id', 'reminders', type_='foreignkey') if 'ix_reminders_status_scheduled_at' in reminders_indexes: op.drop_index('ix_reminders_status_scheduled_at', table_name='reminders') if 'ix_reminders_merchant_id' in reminders_indexes: op.drop_index('ix_reminders_merchant_id', table_name='reminders') if 'ix_reminders_reminder_status' in reminders_indexes: op.drop_index('ix_reminders_reminder_status', table_name='reminders') if 'ix_reminders_scheduled_at' in reminders_indexes: op.drop_index('ix_reminders_scheduled_at', table_name='reminders') if 'ix_reminders_invoice_id' in reminders_indexes: op.drop_index('ix_reminders_invoice_id', table_name='reminders') if 'ix_reminders_reminder_id' in reminders_indexes: op.drop_index('ix_reminders_reminder_id', table_name='reminders') if 'uq_reminders_reminder_id' in reminders_constraints: op.drop_constraint('uq_reminders_reminder_id', 'reminders', type_='unique') if 'merchant_id' in reminders_columns: op.drop_column('reminders', 'merchant_id') if 'failure_reason' in reminders_columns: op.drop_column('reminders', 'failure_reason') if 'sent_at' in reminders_columns: op.drop_column('reminders', 'sent_at') if 'reminder_status' in reminders_columns: op.drop_column('reminders', 'reminder_status') if 'channel' in reminders_columns: op.drop_column('reminders', 'channel') if 'scheduled_at' in reminders_columns: op.drop_column('reminders', 'scheduled_at') if 'preset' in reminders_columns: op.drop_column('reminders', 'preset') if 'invoice_id' in reminders_columns: op.drop_column('reminders', 'invoice_id') if 'reminder_id' in reminders_columns: op.drop_column('reminders', 'reminder_id')