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

chore: fix types in Alembic migration script

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2024-06-24 10:10:29 -03:00
commit a539c90c31
1 changed files with 0 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

91
src/backend/base/langflow/alembic/versions/bc804d8e7a18_fix_types.py
View file

@ -1,91 +0,0 @@
"""Fix types
Revision ID: bc804d8e7a18
Revises: bc2f01c40e4a
Create Date: 2024-04-22 19:33:02.242116
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
from sqlalchemy.engine.reflection import Inspector
# revision identifiers, used by Alembic.
revision: str = 'bc804d8e7a18'
down_revision: Union[str, None] = 'bc2f01c40e4a'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
conn = op.get_bind()
inspector = Inspector.from_engine(conn) # type: ignore
# ### commands auto generated by Alembic - please adjust! ###
table_names = inspector.get_table_names()
if "apikey" in table_names:
column_names = [col["name"] for col in inspector.get_columns("apikey")]
with op.batch_alter_table('apikey', schema=None) as batch_op:
if "created_at" in column_names:
batch_op.alter_column('created_at',
existing_type=postgresql.TIMESTAMP(),
type_=sa.DateTime(timezone=True),
nullable=True)
if "last_used_at" in column_names:
batch_op.alter_column('last_used_at',
existing_type=postgresql.TIMESTAMP(),
type_=sa.DateTime(timezone=True),
existing_nullable=True)
if "credential" in table_names:
column_names = [col["name"] for col in inspector.get_columns("credential")]
with op.batch_alter_table('credential', schema=None) as batch_op:
if "created_at" in column_names:
batch_op.alter_column('created_at',
existing_type=postgresql.TIMESTAMP(),
type_=sa.DateTime(timezone=True),
nullable=True)
if "updated_at" in column_names:
batch_op.alter_column('updated_at',
existing_type=postgresql.TIMESTAMP(),
type_=sa.DateTime(timezone=True),
existing_nullable=True)
# ### end Alembic commands ###
def downgrade() -> None:
conn = op.get_bind()
inspector = Inspector.from_engine(conn) # type: ignore
# ### commands auto generated by Alembic - please adjust! ###
table_names = inspector.get_table_names()
if "credential" in table_names:
column_names = [col["name"] for col in inspector.get_columns("credential")]
with op.batch_alter_table('credential', schema=None) as batch_op:
if "updated_at" in column_names:
batch_op.alter_column('updated_at',
existing_type=sa.DateTime(timezone=True),
type_=postgresql.TIMESTAMP(),
existing_nullable=True)
if "created_at" in column_names:
batch_op.alter_column('created_at',
existing_type=sa.DateTime(timezone=True),
type_=postgresql.TIMESTAMP(),
nullable=False)
if "apikey" in table_names:
column_names = [col["name"] for col in inspector.get_columns("apikey")]
with op.batch_alter_table('apikey', schema=None) as batch_op:
if "last_used_at" in column_names:
batch_op.alter_column('last_used_at',
existing_type=sa.DateTime(timezone=True),
type_=postgresql.TIMESTAMP(),
existing_nullable=True)
if "created_at" in column_names:
batch_op.alter_column('created_at',
existing_type=sa.DateTime(timezone=True),
type_=postgresql.TIMESTAMP(),
nullable=False)
# ### end Alembic commands ###