forked from andika/membership-be
82 lines
3.5 KiB
Python
82 lines
3.5 KiB
Python
"""fix_subscriptions_and_storage
|
|
|
|
Revision ID: 005_fix_subs_storage
|
|
Revises: 004_add_document_file_sizes
|
|
Create Date: 2026-01-04
|
|
|
|
Fixes:
|
|
- Add missing columns to subscriptions table
|
|
- Rename storage_usage.last_calculated_at to last_updated
|
|
"""
|
|
from typing import Sequence, Union
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = '005_fix_subs_storage'
|
|
down_revision: Union[str, None] = '004_add_document_file_sizes'
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""Add missing columns and fix naming (skip if already exists)"""
|
|
from sqlalchemy import inspect
|
|
|
|
conn = op.get_bind()
|
|
inspector = inspect(conn)
|
|
|
|
# Check existing columns in subscriptions table
|
|
existing_columns = {col['name'] for col in inspector.get_columns('subscriptions')}
|
|
|
|
# Add missing columns to subscriptions table only if they don't exist
|
|
if 'start_date' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('start_date', sa.DateTime(timezone=True), nullable=True))
|
|
if 'end_date' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('end_date', sa.DateTime(timezone=True), nullable=True))
|
|
if 'amount_paid_cents' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('amount_paid_cents', sa.Integer(), nullable=True))
|
|
if 'manual_payment_notes' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('manual_payment_notes', sa.Text(), nullable=True))
|
|
if 'manual_payment_admin_id' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('manual_payment_admin_id', UUID(as_uuid=True), nullable=True))
|
|
if 'manual_payment_date' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('manual_payment_date', sa.DateTime(timezone=True), nullable=True))
|
|
if 'payment_method' not in existing_columns:
|
|
op.add_column('subscriptions', sa.Column('payment_method', sa.String(50), nullable=True))
|
|
|
|
# Add foreign key for manual_payment_admin_id if it doesn't exist
|
|
existing_fks = [fk['name'] for fk in inspector.get_foreign_keys('subscriptions')]
|
|
if 'subscriptions_manual_payment_admin_id_fkey' not in existing_fks:
|
|
op.create_foreign_key(
|
|
'subscriptions_manual_payment_admin_id_fkey',
|
|
'subscriptions', 'users',
|
|
['manual_payment_admin_id'], ['id']
|
|
)
|
|
|
|
# Rename storage_usage.last_calculated_at to last_updated (only if needed)
|
|
storage_columns = {col['name'] for col in inspector.get_columns('storage_usage')}
|
|
if 'last_calculated_at' in storage_columns and 'last_updated' not in storage_columns:
|
|
op.alter_column('storage_usage', 'last_calculated_at', new_column_name='last_updated')
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""Remove added columns (rollback)"""
|
|
|
|
# Rename back
|
|
op.alter_column('storage_usage', 'last_updated', new_column_name='last_calculated_at')
|
|
|
|
# Drop foreign key first
|
|
op.drop_constraint('subscriptions_manual_payment_admin_id_fkey', 'subscriptions', type_='foreignkey')
|
|
|
|
# Drop columns from subscriptions
|
|
op.drop_column('subscriptions', 'payment_method')
|
|
op.drop_column('subscriptions', 'manual_payment_date')
|
|
op.drop_column('subscriptions', 'manual_payment_admin_id')
|
|
op.drop_column('subscriptions', 'manual_payment_notes')
|
|
op.drop_column('subscriptions', 'amount_paid_cents')
|
|
op.drop_column('subscriptions', 'end_date')
|
|
op.drop_column('subscriptions', 'start_date')
|