casacade delete tasks

This commit is contained in:
IluaAir
2025-08-06 23:58:29 +03:00
parent 93cf7b2d24
commit c61db4bc9d
4 changed files with 126 additions and 2 deletions

View File

@@ -0,0 +1,65 @@
"""add_cascade_delete_to_tasks
Revision ID: 197b195208e8
Revises: a2fdd0ec4a96
Create Date: 2025-08-06 23:41:56.778423
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '197b195208e8'
down_revision: Union[str, None] = 'a2fdd0ec4a96'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade():
"""Upgrade schema."""
op.execute("PRAGMA foreign_keys=ON")
with op.batch_alter_table('tasks', schema=None) as batch_op:
connection = op.get_bind()
inspector = sa.inspect(connection)
foreign_keys = inspector.get_foreign_keys('tasks')
constraint_name = None
for fk in foreign_keys:
if 'user_id' in fk['constrained_columns']:
constraint_name = fk['name']
break
if constraint_name:
try:
batch_op.drop_constraint(constraint_name, type_='foreignkey')
except:
pass
batch_op.create_foreign_key(
'fk_tasks_user_id_users',
'users',
['user_id'],
['id'],
ondelete='CASCADE'
)
def downgrade():
"""Downgrade schema."""
with op.batch_alter_table('tasks', schema=None) as batch_op:
try:
batch_op.drop_constraint('fk_tasks_user_id_users', type_='foreignkey')
except:
pass
batch_op.create_foreign_key(
'fk_tasks_user_id_users',
'users',
['user_id'],
['id']
)

View File

@@ -0,0 +1,59 @@
"""fix_duplicate_foreign_keys
Revision ID: 4b0f3ea2fd26
Revises: 197b195208e8
Create Date: 2025-08-06 23:54:24.308488
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '4b0f3ea2fd26'
down_revision: Union[str, None] = '197b195208e8'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade():
"""Upgrade schema."""
op.execute("PRAGMA foreign_keys=ON")
with op.batch_alter_table('tasks', schema=None) as batch_op:
connection = op.get_bind()
inspector = sa.inspect(connection)
foreign_keys = inspector.get_foreign_keys('tasks')
for fk in foreign_keys:
if 'user_id' in fk['constrained_columns']:
try:
batch_op.drop_constraint(fk['name'], type_='foreignkey')
except:
pass
batch_op.create_foreign_key(
'fk_tasks_user_id_users',
'users',
['user_id'],
['id'],
ondelete='CASCADE'
)
def downgrade():
"""Downgrade schema."""
with op.batch_alter_table('tasks', schema=None) as batch_op:
try:
batch_op.drop_constraint('fk_tasks_user_id_users', type_='foreignkey')
except:
pass
batch_op.create_foreign_key(
'fk_tasks_user_id_users',
'users',
['user_id'],
['id']
)