47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
"""channel table
|
|
|
|
Revision ID: 7506a3320699
|
|
Revises: 69ef23ef1ed1
|
|
Create Date: 2025-09-05 14:12:37.430983
|
|
|
|
"""
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from sqlalchemy.sql import table, column
|
|
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = '7506a3320699'
|
|
down_revision: Union[str, Sequence[str], None] = '69ef23ef1ed1'
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""Upgrade schema."""
|
|
# Для SQLite: создаём новую таблицу, копируем данные, удаляем старую, переименовываем новую
|
|
conn = op.get_bind()
|
|
# 1. Создать новую таблицу
|
|
op.create_table(
|
|
'channels_new',
|
|
sa.Column('id', sa.Integer(), nullable=False),
|
|
sa.Column('name', sa.String(length=255), nullable=True),
|
|
sa.Column('link', sa.String(length=255), nullable=True),
|
|
sa.Column('admin_id', sa.Integer(), nullable=True),
|
|
sa.PrimaryKeyConstraint('id')
|
|
)
|
|
# 2. Если старая таблица есть, скопировать данные
|
|
result = conn.execute(sa.text("SELECT name FROM sqlite_master WHERE type='table' AND name='channels'")).fetchone()
|
|
if result:
|
|
conn.execute(sa.text("INSERT INTO channels_new (id, name, link, admin_id) SELECT id, name, link, admin_id FROM channels"))
|
|
op.drop_table('channels')
|
|
# 3. Переименовать новую таблицу
|
|
conn.execute(sa.text("ALTER TABLE channels_new RENAME TO channels"))
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""Downgrade schema."""
|
|
op.drop_table('channels')
|