jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

refactor: Suppress SQLite foreign key warnings in Alembic migration (#6809)

Browse source 
Fix: Suppress SQLite foreign key warnings in Alembic migration

Handle SQLite-specific warnings during foreign key inspection in the Alembic migration script by using warnings.catch_warnings() to ignore foreign key constraint warnings
This commit is contained in:
Gabriel Luiz Freitas Almeida 2025-02-26 17:16:19 -03:00 • committed by GitHub
commit bca49a3335
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 11 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

13
src/backend/base/langflow/alembic/versions/bc2f01c40e4a_new_fixes.py
View file

@ -7,6 +7,7 @@ Create Date: 2024-01-26 13:34:14.496769
"""
from typing import Sequence, Union
import warnings
import sqlalchemy as sa
import sqlmodel
@ -25,7 +26,11 @@ def upgrade() -> None:
inspector = sa.inspect(conn) # type: ignore
flow_columns = {column["name"] for column in inspector.get_columns("flow")}
flow_indexes = {index["name"] for index in inspector.get_indexes("flow")}
flow_fks = {fk["name"] for fk in inspector.get_foreign_keys("flow")}
# Suppress the SQLite foreign key warning
with warnings.catch_warnings():
warnings.filterwarnings("ignore", message=".*SQL-parsed foreign key constraint.*")
flow_fks = {fk["name"] for fk in inspector.get_foreign_keys("flow")}
with op.batch_alter_table("flow", schema=None) as batch_op:
if "is_component" not in flow_columns:
@ -47,7 +52,11 @@ def downgrade() -> None:
inspector = sa.inspect(conn) # type: ignore
flow_columns = {column["name"] for column in inspector.get_columns("flow")}
flow_indexes = {index["name"] for index in inspector.get_indexes("flow")}
flow_fks = {fk["name"] for fk in inspector.get_foreign_keys("flow")}
# Suppress the SQLite foreign key warning
with warnings.catch_warnings():
warnings.filterwarnings("ignore", message=".*SQL-parsed foreign key constraint.*")
flow_fks = {fk["name"] for fk in inspector.get_foreign_keys("flow")}
with op.batch_alter_table("flow", schema=None) as batch_op:
if "flow_user_id_fkey" in flow_fks: