Simplify string expressions & Use f-string (#12216)

* Simplify string expressions & Use f-string

This is a follow-up clean-up work for the minor issues caused in the process of introducing Black

* Fixup

Author: XD-DENG

airflow/cli/cli_parser.py

@@ -363,7 +363,7 @@ def positive_int(value):
 # instead.
 ARG_INTERACTIVE = Arg(
     ('-N', '--interactive'),
-    help='Do not capture standard output and error streams ' '(useful for interactive debugging)',
+    help='Do not capture standard output and error streams (useful for interactive debugging)',
     action='store_true',
 )
 ARG_FORCE = Arg(
@@ -573,15 +573,15 @@ def positive_int(value):
 ARG_LASTNAME = Arg(('-l', '--lastname'), help='Last name of the user', required=True, type=str)
 ARG_ROLE = Arg(
     ('-r', '--role'),
-    help='Role of the user. Existing roles include Admin, ' 'User, Op, Viewer, and Public',
+    help='Role of the user. Existing roles include Admin, User, Op, Viewer, and Public',
     required=True,
     type=str,
 )
 ARG_EMAIL = Arg(('-e', '--email'), help='Email of the user', required=True, type=str)
 ARG_EMAIL_OPTIONAL = Arg(('-e', '--email'), help='Email of the user', type=str)
 ARG_PASSWORD = Arg(
     ('-p', '--password'),
-    help='Password of the user, required to create a user ' 'without --use-random-password',
+    help='Password of the user, required to create a user without --use-random-password',
     type=str,
 )
 ARG_USE_RANDOM_PASSWORD = Arg(
@@ -628,7 +628,7 @@ def positive_int(value):
 # info
 ARG_ANONYMIZE = Arg(
     ('--anonymize',),
-    help=('Minimize any personal identifiable information. ' 'Use it when sharing output with others.'),
+    help='Minimize any personal identifiable information. Use it when sharing output with others.',
     action='store_true',
 )
 ARG_FILE_IO = Arg(

airflow/cli/commands/user_command.py

@@ -92,7 +92,7 @@ def users_manage_role(args, remove=False):
         raise SystemExit('Missing args: must supply one of --username or --email')
 
     if args.username and args.email:
-        raise SystemExit('Conflicting args: must supply either --username' ' or --email, but not both')
+        raise SystemExit('Conflicting args: must supply either --username or --email, but not both')
 
     appbuilder = cached_app().appbuilder  # pylint: disable=no-member
     user = appbuilder.sm.find_user(username=args.username) or appbuilder.sm.find_user(email=args.email)

airflow/executors/kubernetes_executor.py

@@ -112,7 +112,7 @@ def run(self) -> None:
                 raise
             else:
                 self.log.warning(
-                    'Watch died gracefully, starting back up with: ' 'last resource_version: %s',
+                    'Watch died gracefully, starting back up with: last resource_version: %s',
                     self.resource_version,
                 )
 
@@ -263,7 +263,7 @@ def _health_check_kube_watcher(self):
             self.log.debug("KubeJobWatcher alive, continuing")
         else:
             self.log.error(
-                'Error while health checking kube watcher process. ' 'Process died for unknown reasons'
+                'Error while health checking kube watcher process. Process died for unknown reasons'
             )
             self.kube_watcher = self._make_kube_watcher()
 
@@ -478,7 +478,7 @@ def clear_not_launched_queued_tasks(self, session=None) -> None:
             pod_list = self.kube_client.list_namespaced_pod(self.kube_config.kube_namespace, **kwargs)
             if not pod_list.items:
                 self.log.info(
-                    'TaskInstance: %s found in queued state but was not launched, ' 'rescheduling', task
+                    'TaskInstance: %s found in queued state but was not launched, rescheduling', task
                 )
                 session.query(TaskInstance).filter(
                     TaskInstance.dag_id == task.dag_id,
@@ -509,7 +509,7 @@ def _create_or_update_secret(secret_name, secret_path):
                         **self.kube_config.kube_client_request_args,
                     )
                 self.log.exception(
-                    'Exception while trying to inject secret. ' 'Secret name: %s, error details: %s',
+                    'Exception while trying to inject secret. Secret name: %s, error details: %s',
                     secret_name,
                     e,
                 )
@@ -605,7 +605,7 @@ def sync(self) -> None:
                         self.change_state(key, State.FAILED, e)
                     else:
                         self.log.warning(
-                            'ApiException when attempting to run task, re-queueing. ' 'Message: %s',
+                            'ApiException when attempting to run task, re-queueing. Message: %s',
                             json.loads(e.body)['message'],
                         )
                         self.task_queue.put(task)
@@ -729,7 +729,7 @@ def _flush_result_queue(self) -> None:
                         self._change_state(key, state, pod_id, namespace)
                     except Exception as e:  # pylint: disable=broad-except
                         self.log.exception(
-                            'Ignoring exception: %s when attempting to change state of %s ' 'to %s.',
+                            'Ignoring exception: %s when attempting to change state of %s to %s.',
                             e,
                             results,
                             state,

airflow/models/baseoperator.py

@@ -1194,7 +1194,7 @@ def _set_relatives(
 
         if len(dags) > 1:
             raise AirflowException(
-                'Tried to set relationships between tasks in ' 'more than one DAG: {}'.format(dags.values())
+                f'Tried to set relationships between tasks in more than one DAG: {dags.values()}'
             )
         elif len(dags) == 1:
             dag = dags.popitem()[1]

airflow/models/dagrun.py

@@ -135,7 +135,7 @@ def __init__(
 
     def __repr__(self):
         return (
-            '<DagRun {dag_id} @ {execution_date}: {run_id}, ' 'externally triggered: {external_trigger}>'
+            '<DagRun {dag_id} @ {execution_date}: {run_id}, externally triggered: {external_trigger}>'
         ).format(
             dag_id=self.dag_id,
             execution_date=self.execution_date,

airflow/providers/apache/druid/hooks/druid.py

@@ -93,7 +93,7 @@ def submit_indexing_job(self, json_index_spec: Dict[str, Any]) -> None:
         self.log.info("Druid ingestion spec: %s", json_index_spec)
         req_index = requests.post(url, data=json_index_spec, headers=self.header, auth=self.get_auth())
         if req_index.status_code != 200:
-            raise AirflowException('Did not get 200 when ' 'submitting the Druid job to {}'.format(url))
+            raise AirflowException(f'Did not get 200 when submitting the Druid job to {url}')
 
         req_json = req_index.json()
         # Wait until the job is completed

airflow/providers/google/cloud/hooks/bigquery.py

@@ -2067,7 +2067,7 @@ def run_query(
             _validate_value("api_resource_configs['query']", configuration['query'], dict)
 
         if sql is None and not configuration['query'].get('query', None):
-            raise TypeError('`BigQueryBaseCursor.run_query` ' 'missing 1 required positional argument: `sql`')
+            raise TypeError('`BigQueryBaseCursor.run_query` missing 1 required positional argument: `sql`')
 
         # BigQuery also allows you to define how you want a table's schema to change
         # as a side effect of a query job
@@ -2801,9 +2801,7 @@ def _split_tablename(
 ) -> Tuple[str, str, str]:
 
     if '.' not in table_input:
-        raise ValueError(
-            'Expected target table name in the format of ' '<dataset>.<table>. Got: {}'.format(table_input)
-        )
+        raise ValueError(f'Expected table name in the format of <dataset>.<table>. Got: {table_input}')
 
     if not default_project_id:
         raise ValueError("INTERNAL: No default project is specified")

airflow/providers/google/cloud/hooks/gcs.py

@@ -630,7 +630,7 @@ def get_crc32c(self, bucket_name: str, object_name: str):
         :type object_name: str
         """
         self.log.info(
-            'Retrieving the crc32c checksum of ' 'object_name: %s in bucket_name: %s',
+            'Retrieving the crc32c checksum of object_name: %s in bucket_name: %s',
             object_name,
             bucket_name,
         )
@@ -651,7 +651,7 @@ def get_md5hash(self, bucket_name: str, object_name: str) -> str:
             storage bucket_name.
         :type object_name: str
         """
-        self.log.info('Retrieving the MD5 hash of ' 'object: %s in bucket: %s', object_name, bucket_name)
+        self.log.info('Retrieving the MD5 hash of object: %s in bucket: %s', object_name, bucket_name)
         client = self.get_conn()
         bucket = client.bucket(bucket_name)
         blob = bucket.get_blob(blob_name=object_name)

airflow/providers/google/cloud/hooks/mlengine.py

@@ -140,7 +140,7 @@ def create_job(self, job: dict, project_id: str, use_existing_job_fn: Optional[C
                     existing_job = self._get_job(project_id, job_id)
                     if not use_existing_job_fn(existing_job):
                         self.log.error(
-                            'Job with job_id %s already exist, but it does ' 'not match our expectation: %s',
+                            'Job with job_id %s already exist, but it does not match our expectation: %s',
                             job_id,
                             existing_job,
                         )

airflow/providers/microsoft/azure/hooks/azure_batch.py

@@ -148,7 +148,7 @@ def configure_pool(  # pylint: disable=too-many-arguments
 
         elif os_family:
             self.log.info(
-                'Using cloud service configuration to create pool, ' 'virtual machine configuration ignored'
+                'Using cloud service configuration to create pool, virtual machine configuration ignored'
             )
             pool = batch_models.PoolAddParameter(
                 id=pool_id,