101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
|
"""Add bonus assignment disputes support
|
||
|
|
|
||
|
|
Revision ID: 021_add_bonus_disputes
|
||
|
|
Revises: 020_add_game_types
|
||
|
|
Create Date: 2024-12-29
|
||
|
|
|
||
|
|
"""
|
||
|
|
from typing import Sequence, Union
|
||
|
|
|
||
|
|
from alembic import op
|
||
|
|
import sqlalchemy as sa
|
||
|
|
from sqlalchemy import inspect
|
||
|
|
|
||
|
|
|
||
|
|
# revision identifiers, used by Alembic.
|
||
|
|
revision: str = '021_add_bonus_disputes'
|
||
|
|
down_revision: Union[str, None] = '020_add_game_types'
|
||
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
||
|
|
depends_on: Union[str, Sequence[str], None] = None
|
||
|
|
|
||
|
|
|
||
|
|
def column_exists(table_name: str, column_name: str) -> bool:
|
||
|
|
bind = op.get_bind()
|
||
|
|
inspector = inspect(bind)
|
||
|
|
columns = [col['name'] for col in inspector.get_columns(table_name)]
|
||
|
|
return column_name in columns
|
||
|
|
|
||
|
|
|
||
|
|
def constraint_exists(table_name: str, constraint_name: str) -> bool:
|
||
|
|
bind = op.get_bind()
|
||
|
|
inspector = inspect(bind)
|
||
|
|
constraints = inspector.get_unique_constraints(table_name)
|
||
|
|
return any(c['name'] == constraint_name for c in constraints)
|
||
|
|
|
||
|
|
|
||
|
|
def upgrade() -> None:
|
||
|
|
bind = op.get_bind()
|
||
|
|
|
||
|
|
# Add bonus_assignment_id column to disputes
|
||
|
|
if not column_exists('disputes', 'bonus_assignment_id'):
|
||
|
|
op.add_column('disputes', sa.Column(
|
||
|
|
'bonus_assignment_id',
|
||
|
|
sa.Integer(),
|
||
|
|
nullable=True
|
||
|
|
))
|
||
|
|
op.create_foreign_key(
|
||
|
|
'fk_disputes_bonus_assignment_id',
|
||
|
|
'disputes',
|
||
|
|
'bonus_assignments',
|
||
|
|
['bonus_assignment_id'],
|
||
|
|
['id'],
|
||
|
|
ondelete='CASCADE'
|
||
|
|
)
|
||
|
|
op.create_index('ix_disputes_bonus_assignment_id', 'disputes', ['bonus_assignment_id'])
|
||
|
|
|
||
|
|
# Drop the unique index on assignment_id first (required before making nullable)
|
||
|
|
if bind.dialect.name != 'sqlite':
|
||
|
|
try:
|
||
|
|
op.drop_index('ix_disputes_assignment_id', 'disputes')
|
||
|
|
except Exception:
|
||
|
|
pass # Index might not exist
|
||
|
|
|
||
|
|
# Make assignment_id nullable (PostgreSQL only, SQLite doesn't support ALTER COLUMN)
|
||
|
|
if bind.dialect.name != 'sqlite':
|
||
|
|
op.alter_column('disputes', 'assignment_id', nullable=True)
|
||
|
|
|
||
|
|
# Create a non-unique index on assignment_id
|
||
|
|
try:
|
||
|
|
op.create_index('ix_disputes_assignment_id_non_unique', 'disputes', ['assignment_id'])
|
||
|
|
except Exception:
|
||
|
|
pass # Index might already exist
|
||
|
|
|
||
|
|
|
||
|
|
def downgrade() -> None:
|
||
|
|
bind = op.get_bind()
|
||
|
|
|
||
|
|
# Remove non-unique index
|
||
|
|
try:
|
||
|
|
op.drop_index('ix_disputes_assignment_id_non_unique', table_name='disputes')
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# Make assignment_id not nullable again
|
||
|
|
if bind.dialect.name != 'sqlite':
|
||
|
|
op.alter_column('disputes', 'assignment_id', nullable=False)
|
||
|
|
|
||
|
|
# Recreate unique index
|
||
|
|
try:
|
||
|
|
op.create_index('ix_disputes_assignment_id', 'disputes', ['assignment_id'], unique=True)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# Remove foreign key, index and column
|
||
|
|
if column_exists('disputes', 'bonus_assignment_id'):
|
||
|
|
try:
|
||
|
|
op.drop_constraint('fk_disputes_bonus_assignment_id', 'disputes', type_='foreignkey')
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
op.drop_index('ix_disputes_bonus_assignment_id', table_name='disputes')
|
||
|
|
op.drop_column('disputes', 'bonus_assignment_id')
|