code: def upgrade():
      conn = op.get_bind()
      session = sa.orm.Session(bind=conn)
  
      with op.batch_alter_table("integration", schema=None) as batch_op:
          batch_op.add_column(sa.Column("engine", sa.String()))
  
      integrations = conn.execute(
          text(
              """
          select id, name, data from integration
      """
          )
      ).fetchall()
  
      for row in integrations:
          try:
              data = json.loads(row["data"])
          except Exception:
              data = {}
  
          if "test" in data:
              del data["test"]
          if "publish" in data:
              del data["publish"]
          if "enabled" in data:
              del data["enabled"]
          if "database_name" in data:
              if row["name"] is None:
                  row["name"] = data["database_name"]
              del data["database_name"]
          integration_type = data.get("type")
          if integration_type is None:
              if row["name"] == "files":
                  integration_type = "files"
              if row["name"] == "views":
                  integration_type = "views"
          if "type" in data:
              del data["type"]
  
          conn.execute(
              text(
                  """
                  update integration
                  set engine = :integration_type,
                      data = :integration_data
                  where id = :integration_id
              """
              ),
              {
                  "integration_type": integration_type,
                  "integration_data": json.dumps(data),
                  "integration_id": row["id"],
              },
          )
  
      session.commit()
  
imports: import json
  from alembic import op
  import sqlalchemy
  from sqlalchemy.sql import text
  import mindsdb.interfaces.storage.db
label_desc: Use image node_id 1 for calling a global function with example usage: upgrade() without return types