Update SQLAlchemy select() to new style (#30515)

SQLAlchemy has a new style for `select() that is standard for 2.0. This
updates our uses of it to avoid `RemovedIn20Warning` warnings.

https://docs.sqlalchemy.org/en/20/errors.html#select-construct-created-in-legacy-mode-keyword-arguments-etc

Author: jedcunningham

airflow/migrations/versions/0093_2_2_0_taskinstance_keyed_to_dagrun.py

@@ -403,7 +403,7 @@ def _multi_table_update(dialect_name, target, column):
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select([dag_run.c[column.name]]).where(condition)
+        sub_q = select(dag_run.c[column.name]).where(condition)
 
         return target.update().values({column: sub_q})
     else:

airflow/migrations/versions/0104_2_3_0_migrate_rtif_to_use_run_id_and_map_index.py

@@ -99,7 +99,7 @@ def _multi_table_update(dialect_name, target, column):
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select([dag_run.c[column.name]]).where(condition)
+        sub_q = select(dag_run.c[column.name]).where(condition)
 
         return target.update().values({column: sub_q})
     else:

airflow/migrations/versions/0105_2_3_0_add_map_index_to_taskfail.py

@@ -95,7 +95,7 @@ def _update_value_from_dag_run(
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select([dag_run.c[target_column.name]]).where(condition)
+        sub_q = select(dag_run.c[target_column.name]).where(condition)
 
         return target_table.update().values({target_column: sub_q})
     else:

airflow/models/dagrun.py

@@ -128,7 +128,7 @@ class DagRun(Base, LoggingMixin):
     log_template_id = Column(
         Integer,
         ForeignKey("log_template.id", name="task_instance_log_template_id_fkey", ondelete="NO ACTION"),
-        default=select([func.max(LogTemplate.__table__.c.id)]),
+        default=select(func.max(LogTemplate.__table__.c.id)),
     )
     updated_at = Column(UtcDateTime, default=timezone.utcnow, onupdate=timezone.utcnow)
 

airflow/utils/db.py

@@ -1161,7 +1161,7 @@ def _create_table_as(
     if dialect_name == "mssql":
         cte = source_query.cte("source")
         moved_data_tbl = table(target_table_name, *(column(c.name) for c in cte.columns))
-        ins = moved_data_tbl.insert().from_select(list(cte.columns), select([cte]))
+        ins = moved_data_tbl.insert().from_select(list(cte.columns), select(cte))
 
         stmt = ins.compile(bind=session.get_bind())
         cte_sql = stmt.ctes[cte]
@@ -1200,7 +1200,7 @@ def _move_dangling_data_to_new_table(
 
     target_table = source_table.to_metadata(source_table.metadata, name=target_table_name)
     log.debug("checking whether rows were moved for table %s", target_table_name)
-    moved_rows_exist_query = select([1]).select_from(target_table).limit(1)
+    moved_rows_exist_query = select(1).select_from(target_table).limit(1)
     first_moved_row = session.execute(moved_rows_exist_query).all()
     session.commit()
 

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-a6ec7b232fa919d38d4ef9f48a1db2968b68e878994554b15f311bd11ff1abb6
+b409a0826f10dc5ef934616fe2d5e72fff322bb1aebe321aad475e4d72c25a4f