Refactor: Use f-strings

airflow/cli/cli_parser.py

@@ -101,12 +101,12 @@ def _iter_indented_subactions(self, action: Action):
             action_subcommands, group_subcommands = partition(
                 lambda d: isinstance(ALL_COMMANDS_DICT[d.dest], GroupCommand), subactions
             )
-            yield Action([], "\n%*s%s:" % (self._current_indent, "", "Groups"), nargs=0)
+            yield Action([], f"\n{' ':{self._current_indent}}Groups", nargs=0)
             self._indent()
             yield from group_subcommands
             self._dedent()
 
-            yield Action([], "\n%*s%s:" % (self._current_indent, "", "Commands"), nargs=0)
+            yield Action([], f"\n{' ':{self._current_indent}}Commands:", nargs=0)
             self._indent()
             yield from action_subcommands
             self._dedent()

airflow/cli/commands/standalone_command.py

@@ -139,7 +139,7 @@ def print_output(self, name: str, output):
             "triggerer": "cyan",
             "standalone": "white",
         }.get(name, "white")
-        colorised_name = colored("%10s" % name, color)
+        colorised_name = colored(f"{name:10}", color)
         for line in output.splitlines():
             print(f"{colorised_name} | {line.strip()}")
 

airflow/providers/google/cloud/sensors/tasks.py

@@ -83,6 +83,6 @@ def poke(self, context: Context) -> bool:
             # page_size=1
         )
 
-        self.log.info("tasks exhausted in cloud task queue?: %s" % (len(tasks) == 0))
+        self.log.info("tasks exhausted in cloud task queue?: %s", (len(tasks) == 0))
 
         return len(tasks) == 0

airflow/providers/oracle/hooks/oracle.py

@@ -346,7 +346,7 @@ def bulk_insert_rows(
         prepared_stm = "insert into {tablename} {columns} values ({values})".format(
             tablename=table,
             columns="({})".format(", ".join(target_fields)) if target_fields else "",
-            values=", ".join(":%s" % i for i in range(1, len(values_base) + 1)),
+            values=", ".join(f":{i}" for i in range(1, len(values_base) + 1)),
         )
         row_count = 0
         # Chunk the rows

tests/api_connexion/endpoints/test_user_endpoint.py

@@ -617,7 +617,7 @@ def test_required_fields(
             environ_overrides={"REMOTE_USER": "test"},
         )
         assert response.status_code == 400, response.json
-        assert response.json["detail"] == "{'%s': ['Missing data for required field.']}" % field
+        assert response.json["detail"] == f"{{'{field}': ['Missing data for required field.']}}"
 
     @pytest.mark.usefixtures("autoclean_admin_user")
     def test_username_can_be_updated(self, autoclean_user_payload, autoclean_username):

tests/api_experimental/common/experimental/test_pool.py

@@ -100,7 +100,7 @@ def test_create_pool_name_too_long(self):
         long_name = "".join(random.choices(string.ascii_lowercase, k=300))
         column_length = models.Pool.pool.property.columns[0].type.length
         with pytest.raises(
-            AirflowBadRequest, match="^Pool name can't be more than %d characters$" % column_length
+            AirflowBadRequest, match=f"^Pool name can't be more than {column_length} characters$"
         ):
             pool_api.create_pool(
                 name=long_name,

tests/operators/test_bash.py

@@ -204,7 +204,7 @@ def test_bash_operator_multi_byte_output(self):
     def test_bash_operator_kill(self, dag_maker):
         import psutil
 
-        sleep_time = "100%d" % os.getpid()
+        sleep_time = f"100{os.getpid()}"
         with dag_maker():
             op = BashOperator(
                 task_id="test_bash_operator_kill",

tests/providers/oracle/operators/test_oracle.py

@@ -90,7 +90,7 @@ def test_push_oracle_exit_to_xcom(self, mock_callproc, request, dag_maker):
         oracle_conn_id = "oracle_default"
         parameters = {"parameter": "value"}
         task_id = "test_push"
-        ora_exit_code = "%05d" % randrange(10**5)
+        ora_exit_code = f"{randrange(10**5):05}"
         error = f"ORA-{ora_exit_code}: This is a five-digit ORA error code"
         mock_callproc.side_effect = oracledb.DatabaseError(error)
 

tests/sensors/test_external_task_sensor.py

@@ -1344,7 +1344,7 @@ def dag_bag_multiple():
             task_id=f"{daily_task.task_id}_{i}",
             external_dag_id=daily_dag.dag_id,
             external_task_id=daily_task.task_id,
-            execution_date="{{ macros.ds_add(ds, -1 * %s) }}" % i,
+            execution_date=f"{{{{ macros.ds_add(ds, -1 * {i}) }}}}",
             dag=agg_dag,
         )
         begin >> task

tests/utils/test_edgemodifier.py

@@ -45,7 +45,7 @@ def f(task_id):
         return f"OP:{task_id}"
 
     with DAG(dag_id="test_xcom_dag", default_args=DEFAULT_ARGS) as dag:
-        operators = [PythonOperator(python_callable=f, task_id="test_op_%i" % i) for i in range(4)]
+        operators = [PythonOperator(python_callable=f, task_id=f"test_op_{i}") for i in range(4)]
         return dag, operators