Simplify strings previously split across lines (#18679)

airflow/jobs/backfill_job.py

@@ -462,7 +462,7 @@ def _per_task_process(key, ti: TaskInstance, session=None):
                 # in case max concurrency has been reached at task runtime
                 elif ti.state == State.NONE:
                     self.log.warning(
-                        "FIXME: task instance {} state was set to None " "externally. This should not happen"
+                        "FIXME: Task instance %s state was set to None externally. This should not happen", ti
                     )
                     ti.set_state(State.SCHEDULED, session=session)
                 if self.rerun_failed_tasks:
@@ -502,9 +502,7 @@ def _per_task_process(key, ti: TaskInstance, session=None):
                     dep_context=backfill_context, session=session, verbose=self.verbose
                 ):
                     if executor.has_task(ti):
-                        self.log.debug(
-                            "Task Instance %s already in executor " "waiting for queue to clear", ti
-                        )
+                        self.log.debug("Task Instance %s already in executor waiting for queue to clear", ti)
                     else:
                         self.log.debug('Sending %s to executor', ti)
                         # Skip scheduled state, we are executing immediately
@@ -544,15 +542,15 @@ def _per_task_process(key, ti: TaskInstance, session=None):
 
                 # special case
                 if ti.state == State.UP_FOR_RETRY:
-                    self.log.debug("Task instance %s retry period not " "expired yet", ti)
+                    self.log.debug("Task instance %s retry period not expired yet", ti)
                     if key in ti_status.running:
                         ti_status.running.pop(key)
                     ti_status.to_run[key] = ti
                     return
 
                 # special case
                 if ti.state == State.UP_FOR_RESCHEDULE:
-                    self.log.debug("Task instance %s reschedule period not " "expired yet", ti)
+                    self.log.debug("Task instance %s reschedule period not expired yet", ti)
                     if key in ti_status.running:
                         ti_status.running.pop(key)
                     ti_status.to_run[key] = ti

airflow/jobs/local_task_job.py

@@ -189,7 +189,7 @@ def heartbeat_callback(self, session=None):
             same_hostname = fqdn == ti.hostname
             if not same_hostname:
                 self.log.warning(
-                    "The recorded hostname %s " "does not match this instance's hostname " "%s",
+                    "The recorded hostname %s does not match this instance's hostname %s",
                     ti.hostname,
                     fqdn,
                 )
@@ -209,7 +209,7 @@ def heartbeat_callback(self, session=None):
                 raise AirflowException("PID of job runner does not match")
         elif self.task_runner.return_code() is None and hasattr(self.task_runner, 'process'):
             self.log.warning(
-                "State of this instance has been externally set to %s. " "Terminating instance.", ti.state
+                "State of this instance has been externally set to %s. Terminating instance.", ti.state
             )
             self.task_runner.terminate()
             if ti.state == State.SUCCESS:

airflow/providers/amazon/aws/hooks/ses.py

@@ -66,7 +66,7 @@ def send_email(
         :param return_path: The email address to which replies will be sent. By default, replies
             are sent to the original sender's email address.
         :param reply_to: The email address to which message bounces and complaints should be sent.
-            "Return-Path" is sometimes called "envelope from," "envelope sender," or "MAIL FROM."
+            "Return-Path" is sometimes called "envelope from", "envelope sender", or "MAIL FROM".
         :param custom_headers: Additional headers to add to the MIME message.
             No validations are run on these values and they should be able to be encoded.
         :return: Response from Amazon SES service with unique message identifier.

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

@@ -182,7 +182,7 @@ def create_transfer_job(self, body: dict) -> dict:
                 if transfer_job.get(STATUS) == GcpTransferJobsStatus.DELETED:
                     body[JOB_NAME] = gen_job_name(job_name)
                     self.log.info(
-                        "Job `%s` has been soft deleted. Creating job with " "new name `%s`",
+                        "Job `%s` has been soft deleted. Creating job with new name `%s`",
                         job_name,
                         {body[JOB_NAME]},
                     )

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

@@ -77,7 +77,7 @@ def get_conn(self) -> container_v1.ClusterManagerClient:
     # TODO: remove one day
     def get_client(self) -> container_v1.ClusterManagerClient:
         warnings.warn(
-            "The get_client method has been deprecated. " "You should use the get_conn method.",
+            "The get_client method has been deprecated. You should use the get_conn method.",
             DeprecationWarning,
         )
         return self.get_conn()

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

@@ -411,7 +411,7 @@ def create_model(
         """
         hook = self.get_conn()
         if 'name' not in model or not model['name']:
-            raise ValueError("Model name must be provided and " "could not be an empty string")
+            raise ValueError("Model name must be provided and could not be an empty string")
         project = f'projects/{project_id}'
 
         self._append_label(model)
@@ -463,7 +463,7 @@ def get_model(
         """
         hook = self.get_conn()
         if not model_name:
-            raise ValueError("Model name must be provided and " "it could not be an empty string")
+            raise ValueError("Model name must be provided and it could not be an empty string")
         full_model_name = f'projects/{project_id}/models/{model_name}'
         request = hook.projects().models().get(name=full_model_name)
         try:

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

@@ -331,7 +331,7 @@ def update_database(
         except AlreadyExists as e:
             if e.code == 409 and operation_id in e.message:
                 self.log.info(
-                    "Replayed update_ddl message - the operation id %s " "was already done before.",
+                    "Replayed update_ddl message - the operation id %s was already done before.",
                     operation_id,
                 )
                 return

airflow/utils/helpers.py

@@ -63,7 +63,7 @@ def validate_group_key(k: str, max_length: int = 200):
         raise AirflowException(f"The key has to be less than {max_length} characters")
     if not GROUP_KEY_REGEX.match(k):
         raise AirflowException(
-            f"The key ({k}) has to be made of alphanumeric characters, dashes " "and underscores exclusively"
+            f"The key ({k}) has to be made of alphanumeric characters, dashes and underscores exclusively"
         )
 
 

docs/publish_docs.py

@@ -62,7 +62,7 @@ def _get_parser():
     available_packages_list = " * " + "\n * ".join(get_available_packages())
     parser = argparse.ArgumentParser(
         description='Copies the built documentation to airflow-site repository.',
-        epilog=f"List of supported documentation packages:\n{available_packages_list}" "",
+        epilog=f"List of supported documentation packages:\n{available_packages_list}",
     )
     parser.formatter_class = argparse.RawTextHelpFormatter
     parser.add_argument(

tests/providers/apache/drill/operators/test_drill.py

@@ -56,8 +56,8 @@ def test_drill_operator_single(self):
 
     def test_drill_operator_multi(self):
         sql = [
-            "create table dfs.tmp.test_airflow as" "select * from cp.`employee.json` limit 10",
-            "select sum(employee_id), any_value(full_name)" "from dfs.tmp.test_airflow",
+            "create table dfs.tmp.test_airflow as select * from cp.`employee.json` limit 10",
+            "select sum(employee_id), any_value(full_name) from dfs.tmp.test_airflow",
         ]
         op = DrillOperator(task_id='drill_operator_test_multi', sql=sql, dag=self.dag)
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE, ignore_ti_state=True)

tests/providers/google/cloud/hooks/test_natural_language.py

@@ -45,7 +45,7 @@ def setUp(self):
         new_callable=mock.PropertyMock,
     )
     @mock.patch(
-        "airflow.providers.google.cloud.hooks.natural_language.CloudNaturalLanguageHook." "_get_credentials"
+        "airflow.providers.google.cloud.hooks.natural_language.CloudNaturalLanguageHook._get_credentials"
     )
     @mock.patch("airflow.providers.google.cloud.hooks.natural_language.LanguageServiceClient")
     def test_language_service_client_creation(self, mock_client, mock_get_creds, mock_client_info):

tests/providers/microsoft/azure/operators/test_azure_container_instances.py

@@ -68,7 +68,7 @@ def make_mock_cg_with_missing_events(container_state):
 
 class TestACIOperator(unittest.TestCase):
     @mock.patch(
-        "airflow.providers.microsoft.azure.operators." "azure_container_instances.AzureContainerInstanceHook"
+        "airflow.providers.microsoft.azure.operators.azure_container_instances.AzureContainerInstanceHook"
     )
     def test_execute(self, aci_mock):
         expected_c_state = ContainerState(state='Terminated', exit_code=0, detail_status='test')
@@ -106,7 +106,7 @@ def test_execute(self, aci_mock):
         assert aci_mock.return_value.delete.call_count == 1
 
     @mock.patch(
-        "airflow.providers.microsoft.azure.operators." "azure_container_instances.AzureContainerInstanceHook"
+        "airflow.providers.microsoft.azure.operators.azure_container_instances.AzureContainerInstanceHook"
     )
     def test_execute_with_failures(self, aci_mock):
         expected_c_state = ContainerState(state='Terminated', exit_code=1, detail_status='test')
@@ -130,7 +130,7 @@ def test_execute_with_failures(self, aci_mock):
         assert aci_mock.return_value.delete.call_count == 1
 
     @mock.patch(
-        "airflow.providers.microsoft.azure.operators." "azure_container_instances.AzureContainerInstanceHook"
+        "airflow.providers.microsoft.azure.operators.azure_container_instances.AzureContainerInstanceHook"
     )
     def test_execute_with_tags(self, aci_mock):
         expected_c_state = ContainerState(state='Terminated', exit_code=0, detail_status='test')
@@ -171,7 +171,7 @@ def test_execute_with_tags(self, aci_mock):
         assert aci_mock.return_value.delete.call_count == 1
 
     @mock.patch(
-        "airflow.providers.microsoft.azure.operators." "azure_container_instances.AzureContainerInstanceHook"
+        "airflow.providers.microsoft.azure.operators.azure_container_instances.AzureContainerInstanceHook"
     )
     def test_execute_with_messages_logs(self, aci_mock):
         events = [Event(message="test"), Event(message="messages")]