code: def upgrade():
      with op.batch_alter_table("predictor", schema=None) as batch_op:
          batch_op.add_column(
              sa.Column(
                  "data_integration_ref", db.Json(), nullable=True
              )
          )
  
      conn = op.get_bind()
      session = sa.orm.Session(bind=conn)
  
      view_integration = conn.execute(
          text(
              """
          select id from integration where name = 'views'
      """
          )
      ).fetchone()
      if view_integration is not None:
          views_integration_id = view_integration["id"]
  
          predictors = conn.execute(
              text(
                  """
              select id, data_integration_id from predictor
          """
              )
          ).fetchall()
  
          for predictor in predictors:
              data_integration_ref = None
              if predictor["data_integration_id"] is not None:
                  data_integration_ref = {
                      "type": "integration",
                      "id": predictor["data_integration_id"],
                  }
              if (
                  predictor["data_integration_id"]
                  == views_integration_id
              ):
                  data_integration_ref = {"type": "view"}
              if isinstance(data_integration_ref, dict):
                  data_integration_ref = json.dumps(
                      data_integration_ref
                  )
              conn.execute(
                  text(
                      """
                  update predictor set data_integration_ref = :data_integration_ref where id = :id
              """
                  ),
                  {
                      "data_integration_ref": data_integration_ref,
                      "id": predictor["id"],
                  },
              )
  
      session.commit()
  
      with op.batch_alter_table("predictor", schema=None) as batch_op:
          batch_op.drop_constraint(
              "fk_data_integration_id", type_="foreignkey"
          )
          batch_op.drop_column("data_integration_id")
  
      conn.execute(
          text(
              """
          delete from integration where name = 'views'
      """
          )
      )
      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