You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2021/10/17 16:34:26 UTC
[airflow] branch main updated: More f-strings (#18855)
This is an automated email from the ASF dual-hosted git repository.
kamilbregula pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 86a2a19 More f-strings (#18855)
86a2a19 is described below
commit 86a2a19ad2bdc87a9ad14bb7fde9313b2d7489bb
Author: Kamil BreguĊa <mi...@users.noreply.github.com>
AuthorDate: Sun Oct 17 18:34:06 2021 +0200
More f-strings (#18855)
---
.pre-commit-config.yaml | 6 ++
airflow/api/common/experimental/trigger_dag.py | 5 +-
airflow/cli/commands/task_command.py | 13 ++--
airflow/cli/commands/user_command.py | 5 +-
airflow/cli/commands/webserver_command.py | 18 ++----
airflow/config_templates/default_celery.py | 5 +-
airflow/configuration.py | 26 +++-----
.../example_passing_params_via_test_command.py | 7 +--
airflow/executors/kubernetes_executor.py | 19 +++---
airflow/jobs/backfill_job.py | 13 ++--
airflow/jobs/local_task_job.py | 4 +-
airflow/logging_config.py | 14 ++---
...867_change_datetime_to_datetime2_6_on_mssql_.py | 6 +-
.../7b2661a43ba3_taskinstance_keyed_to_dagrun.py | 6 +-
.../83f031fd9f1c_improve_mssql_compatibility.py | 6 +-
...606e2_add_scheduling_decision_to_dagrun_and_.py | 9 ++-
.../bbf4a7ad0465_remove_id_column_from_xcom.py | 6 +-
airflow/models/baseoperator.py | 44 +++++--------
airflow/models/connection.py | 36 ++++-------
airflow/models/dag.py | 8 +--
airflow/models/taskinstance.py | 8 +--
airflow/operators/subdag.py | 6 +-
airflow/operators/trigger_dagrun.py | 3 +-
airflow/providers/amazon/aws/hooks/batch_client.py | 4 +-
airflow/providers/amazon/aws/hooks/s3.py | 5 +-
.../amazon/aws/log/cloudwatch_task_handler.py | 9 +--
airflow/providers/amazon/aws/operators/athena.py | 11 ++--
airflow/providers/amazon/aws/operators/datasync.py | 7 +--
.../amazon/aws/sensors/s3_keys_unchanged.py | 4 +-
airflow/providers/apache/hive/hooks/hive.py | 17 ++---
.../providers/apache/kylin/operators/kylin_cube.py | 4 +-
airflow/providers/apache/livy/hooks/livy.py | 14 ++---
airflow/providers/apache/spark/hooks/spark_sql.py | 5 +-
.../providers/apache/spark/hooks/spark_submit.py | 22 +++----
airflow/providers/dingding/hooks/dingding.py | 3 +-
.../google/cloud/example_dags/example_functions.py | 13 ++--
airflow/providers/google/cloud/hooks/bigquery.py | 29 +++------
airflow/providers/google/cloud/hooks/cloud_sql.py | 26 +++-----
.../cloud/hooks/cloud_storage_transfer_service.py | 6 +-
airflow/providers/google/cloud/hooks/dataflow.py | 9 ++-
airflow/providers/google/cloud/hooks/dlp.py | 5 +-
airflow/providers/google/cloud/hooks/gcs.py | 8 +--
airflow/providers/google/cloud/hooks/pubsub.py | 4 +-
.../providers/google/cloud/operators/bigtable.py | 8 +--
.../providers/google/cloud/operators/cloud_sql.py | 15 ++---
.../providers/google/cloud/operators/compute.py | 4 +-
.../providers/google/cloud/operators/dataflow.py | 10 ++-
.../providers/google/cloud/operators/dataproc.py | 21 +++----
.../providers/google/cloud/operators/functions.py | 10 +--
.../providers/google/cloud/operators/mlengine.py | 4 +-
.../providers/google/cloud/operators/spanner.py | 6 +-
airflow/providers/google/cloud/sensors/gcs.py | 27 ++++----
.../providers/google/cloud/transfers/gcs_to_gcs.py | 9 +--
.../google/cloud/transfers/gcs_to_sftp.py | 2 +-
.../google/cloud/transfers/sftp_to_gcs.py | 2 +-
.../google/cloud/utils/field_validator.py | 55 ++++++++---------
airflow/providers/google/suite/hooks/sheets.py | 7 +--
.../google/suite/transfers/gcs_to_gdrive.py | 2 +-
airflow/providers/grpc/hooks/grpc.py | 5 +-
.../jenkins/operators/jenkins_job_trigger.py | 5 +-
.../microsoft/azure/operators/azure_batch.py | 17 +++--
.../providers/microsoft/winrm/operators/winrm.py | 5 +-
.../providers/mysql/transfers/vertica_to_mysql.py | 6 +-
.../providers/pagerduty/hooks/pagerduty_events.py | 5 +-
airflow/providers/plexus/hooks/plexus.py | 4 +-
airflow/providers/plexus/operators/job.py | 8 +--
airflow/providers/postgres/hooks/postgres.py | 5 +-
airflow/providers/qubole/operators/qubole_check.py | 6 +-
airflow/providers/qubole/sensors/qubole.py | 4 +-
airflow/providers/sftp/operators/sftp.py | 5 +-
airflow/sensors/base.py | 8 +--
airflow/sensors/external_task.py | 4 +-
airflow/serialization/serialized_objects.py | 10 ++-
airflow/stats.py | 18 +-----
airflow/task/task_runner/cgroup_task_runner.py | 4 +-
airflow/ti_deps/deps/dag_ti_slots_available_dep.py | 6 +-
airflow/ti_deps/deps/dagrun_exists_dep.py | 3 +-
.../ti_deps/deps/exec_date_after_start_date_dep.py | 12 ++--
airflow/ti_deps/deps/not_in_retry_period_dep.py | 8 ++-
airflow/ti_deps/deps/prev_dagrun_dep.py | 13 ++--
airflow/ti_deps/deps/ready_to_reschedule.py | 8 ++-
airflow/ti_deps/deps/runnable_exec_date_dep.py | 18 ++++--
airflow/ti_deps/deps/trigger_rule_dep.py | 71 +++++++++++----------
airflow/ti_deps/deps/valid_state_dep.py | 6 +-
airflow/utils/cli.py | 10 ++-
airflow/utils/compression.py | 3 +-
airflow/utils/helpers.py | 6 +-
airflow/utils/log/file_task_handler.py | 4 +-
airflow/utils/operator_resources.py | 4 +-
airflow/utils/orm_event_handlers.py | 4 +-
airflow/www/api/experimental/endpoints.py | 16 ++---
airflow/www/fab_security/manager.py | 11 ++--
airflow/www/security.py | 10 +--
airflow/www/views.py | 17 ++---
dev/airflow-github | 4 +-
docs/exts/docs_build/run_patched_sphinx.py | 3 +-
scripts/ci/pre_commit/pre_commit_yaml_to_cfg.py | 4 +-
.../endpoints/test_extra_link_endpoint.py | 7 +--
tests/api_connexion/endpoints/test_log_endpoint.py | 12 ++--
tests/conftest.py | 44 +++++--------
tests/decorators/test_python.py | 4 +-
tests/jobs/test_backfill_job.py | 4 +-
tests/operators/test_python.py | 4 +-
.../amazon/aws/log/test_cloudwatch_task_handler.py | 16 ++---
tests/providers/apache/hive/hooks/test_hive.py | 4 +-
tests/providers/apache/hive/operators/test_hive.py | 5 +-
.../providers/apache/spark/hooks/test_spark_sql.py | 7 +--
.../elasticsearch/log/test_es_task_handler.py | 5 +-
.../providers/google/cloud/hooks/test_bigquery.py | 2 +-
.../google/cloud/hooks/test_cloud_memorystore.py | 8 +--
.../hooks/test_cloud_storage_transfer_service.py | 4 +-
tests/providers/google/cloud/hooks/test_dlp.py | 12 +---
.../providers/google/cloud/hooks/test_functions.py | 4 +-
tests/providers/google/cloud/hooks/test_kms.py | 4 +-
.../providers/google/cloud/hooks/test_mlengine.py | 12 +---
tests/providers/google/cloud/hooks/test_pubsub.py | 4 +-
tests/providers/google/cloud/hooks/test_vision.py | 13 ++--
.../operators/test_cloud_build_system_helper.py | 4 +-
.../google/cloud/operators/test_cloud_sql.py | 18 ++----
.../google/cloud/transfers/test_gcs_to_gcs.py | 4 +-
.../google/cloud/utils/gcp_authenticator.py | 4 +-
tests/providers/grpc/hooks/test_grpc.py | 12 ++--
tests/providers/imap/hooks/test_imap.py | 8 ++-
tests/security/test_kerberos.py | 5 +-
tests/sensors/test_python.py | 4 +-
tests/test_utils/system_tests_class.py | 10 +--
tests/utils/test_helpers.py | 8 +--
tests/utils/test_log_handlers.py | 8 +--
tests/www/views/test_views_decorators.py | 5 +-
tests/www/views/test_views_extra_links.py | 72 ++++++----------------
130 files changed, 544 insertions(+), 813 deletions(-)
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index bec3db3..e652f95 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -258,6 +258,12 @@ repos:
exclude: |
(?x)
^airflow/_vendor/
+ args:
+ # If flynt detects too long text it ignores it. So we set a very large limit to make it easy
+ # to split the text by hand. Too long lines are detected by flake8 (below),
+ # so the user is informed to take action.
+ - --line-length
+ - '99999'
- repo: https://github.com/codespell-project/codespell
rev: v2.1.0
hooks:
diff --git a/airflow/api/common/experimental/trigger_dag.py b/airflow/api/common/experimental/trigger_dag.py
index 2e64f86..d5f2c02 100644
--- a/airflow/api/common/experimental/trigger_dag.py
+++ b/airflow/api/common/experimental/trigger_dag.py
@@ -62,9 +62,8 @@ def _trigger_dag(
min_dag_start_date = dag.default_args["start_date"]
if min_dag_start_date and execution_date < min_dag_start_date:
raise ValueError(
- "The execution_date [{}] should be >= start_date [{}] from DAG's default_args".format(
- execution_date.isoformat(), min_dag_start_date.isoformat()
- )
+ f"The execution_date [{execution_date.isoformat()}] should be >= start_date "
+ f"[{min_dag_start_date.isoformat()}] from DAG's default_args"
)
run_id = run_id or DagRun.generate_run_id(DagRunType.MANUAL, execution_date)
diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index ffba0e1..565a19e 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -241,13 +241,14 @@ def task_run(args, dag=None):
unsupported_options = [o for o in RAW_TASK_UNSUPPORTED_OPTION if getattr(args, o)]
if unsupported_options:
+ unsupported_raw_task_flags = ', '.join(f'--{o}' for o in RAW_TASK_UNSUPPORTED_OPTION)
+ unsupported_flags = ', '.join(f'--{o}' for o in unsupported_options)
raise AirflowException(
- "Option --raw does not work with some of the other options on this command. You "
- "can't use --raw option and the following options: {}. You provided the option {}. "
- "Delete it to execute the command".format(
- ", ".join(f"--{o}" for o in RAW_TASK_UNSUPPORTED_OPTION),
- ", ".join(f"--{o}" for o in unsupported_options),
- )
+ "Option --raw does not work with some of the other options on this command. "
+ "You can't use --raw option and the following options: "
+ f"{unsupported_raw_task_flags}. "
+ f"You provided the option {unsupported_flags}. "
+ "Delete it to execute the command."
)
if dag and args.pickle:
raise AirflowException("You cannot use the --pickle option when using DAG.cli() method.")
diff --git a/airflow/cli/commands/user_command.py b/airflow/cli/commands/user_command.py
index 88b32b0..15cda54 100644
--- a/airflow/cli/commands/user_command.py
+++ b/airflow/cli/commands/user_command.py
@@ -199,9 +199,8 @@ def _import_users(users_list):
print(f"Found existing user with email '{user['email']}'")
if existing_user.username != user['username']:
raise SystemExit(
- "Error: Changing the username is not allowed - "
- "please delete and recreate the user with "
- "email '{}'".format(user['email'])
+ f"Error: Changing the username is not allowed - please delete and recreate the user with"
+ f" email {user['email']!r}"
)
existing_user.roles = roles
diff --git a/airflow/cli/commands/webserver_command.py b/airflow/cli/commands/webserver_command.py
index 296cd89..7fdc743 100644
--- a/airflow/cli/commands/webserver_command.py
+++ b/airflow/cli/commands/webserver_command.py
@@ -369,24 +369,14 @@ def webserver(args):
print(
textwrap.dedent(
- '''\
+ f'''\
Running the Gunicorn Server with:
- Workers: {num_workers} {workerclass}
- Host: {hostname}:{port}
+ Workers: {num_workers} {args.workerclass}
+ Host: {args.hostname}:{args.port}
Timeout: {worker_timeout}
Logfiles: {access_logfile} {error_logfile}
Access Logformat: {access_logformat}
- =================================================================\
- '''.format(
- num_workers=num_workers,
- workerclass=args.workerclass,
- hostname=args.hostname,
- port=args.port,
- worker_timeout=worker_timeout,
- access_logfile=access_logfile,
- error_logfile=error_logfile,
- access_logformat=access_logformat,
- )
+ ================================================================='''
)
)
diff --git a/airflow/config_templates/default_celery.py b/airflow/config_templates/default_celery.py
index 7231db5..8fedeef 100644
--- a/airflow/config_templates/default_celery.py
+++ b/airflow/config_templates/default_celery.py
@@ -87,9 +87,8 @@ except AirflowConfigException:
)
except Exception as e:
raise AirflowException(
- 'Exception: There was an unknown Celery SSL Error. '
- 'Please ensure you want to use '
- 'SSL and/or have all necessary certs and key ({}).'.format(e)
+ f'Exception: There was an unknown Celery SSL Error. Please ensure you want to use SSL and/or have '
+ f'all necessary certs and key ({e}).'
)
result_backend = DEFAULT_CELERY_CONFIG['result_backend']
diff --git a/airflow/configuration.py b/airflow/configuration.py
index e120b26..b1f0095 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -294,12 +294,9 @@ class AirflowConfigParser(ConfigParser):
@staticmethod
def _create_future_warning(name, section, current_value, new_value, version):
warnings.warn(
- 'The {name} setting in [{section}] has the old default value '
- 'of {current_value!r}. This value has been changed to {new_value!r} in the '
- 'running config, but please update your config before Apache '
- 'Airflow {version}.'.format(
- name=name, section=section, current_value=current_value, new_value=new_value, version=version
- ),
+ f'The {name!r} setting in [{section}] has the old default value of {current_value!r}. '
+ f'This value has been changed to {new_value!r} in the running config, but '
+ f'please update your config before Apache Airflow {version}.',
FutureWarning,
)
@@ -724,24 +721,15 @@ class AirflowConfigParser(ConfigParser):
def _warn_deprecate(section, key, deprecated_section, deprecated_name):
if section == deprecated_section:
warnings.warn(
- 'The {old} option in [{section}] has been renamed to {new} - the old '
- 'setting has been used, but please update your config.'.format(
- old=deprecated_name,
- new=key,
- section=section,
- ),
+ f'The {deprecated_name} option in [{section}] has been renamed to {key} - '
+ f'the old setting has been used, but please update your config.',
DeprecationWarning,
stacklevel=3,
)
else:
warnings.warn(
- 'The {old_key} option in [{old_section}] has been moved to the {new_key} option in '
- '[{new_section}] - the old setting has been used, but please update your config.'.format(
- old_section=deprecated_section,
- old_key=deprecated_name,
- new_key=key,
- new_section=section,
- ),
+ f'The {deprecated_name} option in [{deprecated_section}] has been moved to the {key} option '
+ f'in [{section}] - the old setting has been used, but please update your config.',
DeprecationWarning,
stacklevel=3,
)
diff --git a/airflow/example_dags/example_passing_params_via_test_command.py b/airflow/example_dags/example_passing_params_via_test_command.py
index e3f04c4..7b8742b 100644
--- a/airflow/example_dags/example_passing_params_via_test_command.py
+++ b/airflow/example_dags/example_passing_params_via_test_command.py
@@ -36,10 +36,7 @@ def my_py_command(params, test_mode=None, task=None):
"""
if test_mode:
print(
- " 'foo' was passed in via test={} command : kwargs[params][foo] \
- = {}".format(
- test_mode, task.params["foo"]
- )
+ f" 'foo' was passed in via test={test_mode} command : kwargs[params][foo] = {task.params['foo']}"
)
# Print out the value of "miff", passed in below via the Python Operator
print(f" 'miff' was passed in via task params = {params['miff']}")
@@ -72,7 +69,7 @@ with DAG(
"""
echo " 'foo was passed in via Airflow CLI Test command with value {{ params.foo }} "
echo " 'miff was passed in via BashOperator with value {{ params.miff }} "
- """
+ """
)
also_run_this = BashOperator(
diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
index d1b60cd..9cd4dea 100644
--- a/airflow/executors/kubernetes_executor.py
+++ b/airflow/executors/kubernetes_executor.py
@@ -179,8 +179,8 @@ class KubernetesJobWatcher(multiprocessing.Process, LoggingMixin):
# Return resource version 0
return '0'
raise AirflowException(
- 'Kubernetes failure for %s with code %s and message: %s'
- % (raw_object['reason'], raw_object['code'], raw_object['message'])
+ f"Kubernetes failure for {raw_object['reason']} with code {raw_object['code']} and message: "
+ f"{raw_object['message']}"
)
def process_status(
@@ -468,12 +468,12 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
continue
# Build the pod selector
- dict_string = "dag_id={},task_id={},airflow-worker={}".format(
- pod_generator.make_safe_label_value(task.dag_id),
- pod_generator.make_safe_label_value(task.task_id),
- pod_generator.make_safe_label_value(str(self.scheduler_job_id)),
+ base_label_selector = (
+ f"dag_id={pod_generator.make_safe_label_value(task.dag_id)},"
+ f"task_id={pod_generator.make_safe_label_value(task.task_id)},"
+ f"airflow-worker={pod_generator.make_safe_label_value(str(self.scheduler_job_id))}"
)
- kwargs = dict(label_selector=dict_string)
+ kwargs = dict(label_selector=base_label_selector)
if self.kube_config.kube_client_request_args:
kwargs.update(**self.kube_config.kube_client_request_args)
@@ -483,8 +483,9 @@ class KubernetesExecutor(BaseExecutor, LoggingMixin):
if pod_list.items:
continue
# Fallback to old style of using execution_date
- kwargs['label_selector'] = dict_string + ',execution_date={}'.format(
- pod_generator.datetime_to_label_safe_datestring(task.execution_date)
+ kwargs['label_selector'] = (
+ f'{base_label_selector},'
+ f'execution_date={pod_generator.datetime_to_label_safe_datestring(task.execution_date)}'
)
pod_list = self.kube_client.list_namespaced_pod(self.kube_config.kube_namespace, **kwargs)
if pod_list.items:
diff --git a/airflow/jobs/backfill_job.py b/airflow/jobs/backfill_job.py
index 1bcb151..2d06dc6 100644
--- a/airflow/jobs/backfill_job.py
+++ b/airflow/jobs/backfill_job.py
@@ -282,9 +282,8 @@ class BackfillJob(BaseJob):
if state in (State.FAILED, State.SUCCESS) and ti.state in self.STATES_COUNT_AS_RUNNING:
msg = (
- "Executor reports task instance {} finished ({}) "
- "although the task says its {}. Was the task "
- "killed externally? Info: {}".format(ti, state, ti.state, info)
+ f"Executor reports task instance {ti} finished ({state}) although the task says its "
+ f"{ti.state}. Was the task killed externally? Info: {info}"
)
self.log.error(msg)
ti.handle_failure_with_callback(error=msg)
@@ -579,8 +578,7 @@ class BackfillJob(BaseJob):
open_slots = pool.open_slots(session=session)
if open_slots <= 0:
raise NoAvailablePoolSlot(
- "Not scheduling since there are "
- "{} open slots in pool {}".format(open_slots, task.pool)
+ f"Not scheduling since there are {open_slots} open slots in pool {task.pool}"
)
num_running_task_instances_in_dag = DAG.get_num_task_instances(
@@ -778,9 +776,8 @@ class BackfillJob(BaseJob):
tasks_that_depend_on_past = [t.task_id for t in self.dag.task_dict.values() if t.depends_on_past]
if tasks_that_depend_on_past:
raise AirflowException(
- 'You cannot backfill backwards because one or more tasks depend_on_past: {}'.format(
- ",".join(tasks_that_depend_on_past)
- )
+ f'You cannot backfill backwards because one or more '
+ f'tasks depend_on_past: {",".join(tasks_that_depend_on_past)}'
)
dagrun_infos = dagrun_infos[::-1]
diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index 7b37d83..43eca4a 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -141,8 +141,8 @@ class LocalTaskJob(BaseJob):
Stats.incr('local_task_job_prolonged_heartbeat_failure', 1, 1)
self.log.error("Heartbeat time limit exceeded!")
raise AirflowException(
- "Time since last heartbeat({:.2f}s) "
- "exceeded limit ({}s).".format(time_since_last_heartbeat, heartbeat_time_limit)
+ f"Time since last heartbeat({time_since_last_heartbeat:.2f}s) exceeded limit "
+ f"({heartbeat_time_limit}s)."
)
finally:
self.on_kill()
diff --git a/airflow/logging_config.py b/airflow/logging_config.py
index e6824c7..3b318c5 100644
--- a/airflow/logging_config.py
+++ b/airflow/logging_config.py
@@ -92,20 +92,16 @@ def validate_logging_config(logging_config):
# Check for pre 1.10 setting that might be in deployed airflow.cfg files
if task_log_reader == "file.task" and _get_handler("task"):
warnings.warn(
- "task_log_reader setting in [logging] has a deprecated value of "
- "{!r}, but no handler with this name was found. Please update "
- "your config to use {!r}. Running config has been adjusted to "
- "match".format(
- task_log_reader,
- "task",
- ),
+ f"task_log_reader setting in [logging] has a deprecated value of {task_log_reader!r}, "
+ "but no handler with this name was found. Please update your config to use task. "
+ "Running config has been adjusted to match",
DeprecationWarning,
)
conf.set('logging', 'task_log_reader', 'task')
else:
raise AirflowConfigException(
- "Configured task_log_reader {!r} was not a handler of the 'airflow.task' "
- "logger.".format(task_log_reader)
+ f"Configured task_log_reader {task_log_reader!r} was not a handler of "
+ f"the 'airflow.task' logger."
)
diff --git a/airflow/migrations/versions/74effc47d867_change_datetime_to_datetime2_6_on_mssql_.py b/airflow/migrations/versions/74effc47d867_change_datetime_to_datetime2_6_on_mssql_.py
index 5505acd..7fa4ab9 100644
--- a/airflow/migrations/versions/74effc47d867_change_datetime_to_datetime2_6_on_mssql_.py
+++ b/airflow/migrations/versions/74effc47d867_change_datetime_to_datetime2_6_on_mssql_.py
@@ -253,14 +253,12 @@ def get_table_constraints(conn, table_name):
:return: a dictionary of ((constraint name, constraint type), column name) of table
:rtype: defaultdict(list)
"""
- query = """SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
+ query = f"""SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
WHERE tc.TABLE_NAME = '{table_name}' AND
(tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE')
- """.format(
- table_name=table_name
- )
+ """
result = conn.execute(query).fetchall()
constraint_dict = defaultdict(list)
for constraint, constraint_type, column in result:
diff --git a/airflow/migrations/versions/7b2661a43ba3_taskinstance_keyed_to_dagrun.py b/airflow/migrations/versions/7b2661a43ba3_taskinstance_keyed_to_dagrun.py
index 059144e..4610a8f 100644
--- a/airflow/migrations/versions/7b2661a43ba3_taskinstance_keyed_to_dagrun.py
+++ b/airflow/migrations/versions/7b2661a43ba3_taskinstance_keyed_to_dagrun.py
@@ -89,14 +89,12 @@ def get_table_constraints(conn, table_name):
:return: a dictionary of ((constraint name, constraint type), column name) of table
:rtype: defaultdict(list)
"""
- query = """SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
+ query = f"""SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
WHERE tc.TABLE_NAME = '{table_name}' AND
(tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE')
- """.format(
- table_name=table_name
- )
+ """
result = conn.execute(query).fetchall()
constraint_dict = defaultdict(lambda: defaultdict(list))
for constraint, constraint_type, col_name in result:
diff --git a/airflow/migrations/versions/83f031fd9f1c_improve_mssql_compatibility.py b/airflow/migrations/versions/83f031fd9f1c_improve_mssql_compatibility.py
index 7458c28..daaf711 100644
--- a/airflow/migrations/versions/83f031fd9f1c_improve_mssql_compatibility.py
+++ b/airflow/migrations/versions/83f031fd9f1c_improve_mssql_compatibility.py
@@ -61,14 +61,12 @@ def get_table_constraints(conn, table_name):
:return: a dictionary of ((constraint name, constraint type), column name) of table
:rtype: defaultdict(list)
"""
- query = """SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
+ query = f"""SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
WHERE tc.TABLE_NAME = '{table_name}' AND
(tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE')
- """.format(
- table_name=table_name
- )
+ """
result = conn.execute(query).fetchall()
constraint_dict = defaultdict(list)
for constraint, constraint_type, column in result:
diff --git a/airflow/migrations/versions/98271e7606e2_add_scheduling_decision_to_dagrun_and_.py b/airflow/migrations/versions/98271e7606e2_add_scheduling_decision_to_dagrun_and_.py
index e4b8510..9d6ca57 100644
--- a/airflow/migrations/versions/98271e7606e2_add_scheduling_decision_to_dagrun_and_.py
+++ b/airflow/migrations/versions/98271e7606e2_add_scheduling_decision_to_dagrun_and_.py
@@ -90,9 +90,12 @@ def upgrade():
# DagParser it will get set to correct value.
op.execute(
- "UPDATE dag SET concurrency={}, has_task_concurrency_limits={} where concurrency IS NULL".format(
- concurrency, 1 if is_sqlite or is_mssql else sa.true()
- )
+ f"""
+ UPDATE dag SET
+ concurrency={concurrency},
+ has_task_concurrency_limits={1 if is_sqlite or is_mssql else sa.true()}
+ where concurrency IS NULL
+ """
)
with op.batch_alter_table('dag', schema=None) as batch_op:
diff --git a/airflow/migrations/versions/bbf4a7ad0465_remove_id_column_from_xcom.py b/airflow/migrations/versions/bbf4a7ad0465_remove_id_column_from_xcom.py
index 045f5a6..1c52da9 100644
--- a/airflow/migrations/versions/bbf4a7ad0465_remove_id_column_from_xcom.py
+++ b/airflow/migrations/versions/bbf4a7ad0465_remove_id_column_from_xcom.py
@@ -50,14 +50,12 @@ def get_table_constraints(conn, table_name):
:return: a dictionary of ((constraint name, constraint type), column name) of table
:rtype: defaultdict(list)
"""
- query = """SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
+ query = f"""SELECT tc.CONSTRAINT_NAME , tc.CONSTRAINT_TYPE, ccu.COLUMN_NAME
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE AS ccu ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
WHERE tc.TABLE_NAME = '{table_name}' AND
(tc.CONSTRAINT_TYPE = 'PRIMARY KEY' or UPPER(tc.CONSTRAINT_TYPE) = 'UNIQUE')
- """.format(
- table_name=table_name
- )
+ """
result = conn.execute(query).fetchall()
constraint_dict = defaultdict(list)
for constraint, constraint_type, column in result:
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
index 04aebee..281e8de 100644
--- a/airflow/models/baseoperator.py
+++ b/airflow/models/baseoperator.py
@@ -533,14 +533,13 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
if kwargs:
if not conf.getboolean('operators', 'ALLOW_ILLEGAL_ARGUMENTS'):
raise AirflowException(
- "Invalid arguments were passed to {c} (task_id: {t}). Invalid "
- "arguments were:\n**kwargs: {k}".format(c=self.__class__.__name__, k=kwargs, t=task_id),
+ f"Invalid arguments were passed to {self.__class__.__name__} (task_id: {task_id}). "
+ f"Invalid arguments were:\n**kwargs: {kwargs}",
)
warnings.warn(
- 'Invalid arguments were passed to {c} (task_id: {t}). '
- 'Support for passing such arguments will be dropped in '
- 'future. Invalid arguments were:'
- '\n**kwargs: {k}'.format(c=self.__class__.__name__, k=kwargs, t=task_id),
+ f'Invalid arguments were passed to {self.__class__.__name__} (task_id: {task_id}). '
+ 'Support for passing such arguments will be dropped in future. '
+ f'Invalid arguments were:\n**kwargs: {kwargs}',
category=PendingDeprecationWarning,
stacklevel=3,
)
@@ -585,13 +584,8 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
if not TriggerRule.is_valid(trigger_rule):
raise AirflowException(
- "The trigger_rule must be one of {all_triggers},"
- "'{d}.{t}'; received '{tr}'.".format(
- all_triggers=TriggerRule.all_triggers(),
- d=dag.dag_id if dag else "",
- t=task_id,
- tr=trigger_rule,
- )
+ f"The trigger_rule must be one of {TriggerRule.all_triggers()},"
+ f"'{dag.dag_id if dag else ''}.{task_id}'; received '{trigger_rule}'."
)
self.trigger_rule = trigger_rule
@@ -646,13 +640,9 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
self.priority_weight = priority_weight
if not WeightRule.is_valid(weight_rule):
raise AirflowException(
- "The weight_rule must be one of {all_weight_rules},"
- "'{d}.{t}'; received '{tr}'.".format(
- all_weight_rules=WeightRule.all_weight_rules,
- d=dag.dag_id if dag else "",
- t=task_id,
- tr=weight_rule,
- )
+ f"The weight_rule must be one of "
+ f"{WeightRule.all_weight_rules},'{dag.dag_id if dag else ''}.{task_id}'; "
+ f"received '{weight_rule}'."
)
self.weight_rule = weight_rule
self.resources: Optional[Resources] = Resources(**resources) if resources else None
@@ -1416,8 +1406,7 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
for task in task_list:
if not isinstance(task, BaseOperator):
raise AirflowException(
- "Relationships can only be set between "
- "Operators; received {}".format(task.__class__.__name__)
+ f"Relationships can only be set between Operators; received {task.__class__.__name__}"
)
# relationships can only be set if the tasks share a single DAG. Tasks
@@ -1434,9 +1423,8 @@ class BaseOperator(Operator, LoggingMixin, TaskMixin, metaclass=BaseOperatorMeta
dag = dags.popitem()[1]
else:
raise AirflowException(
- "Tried to create relationships between tasks that don't have "
- "DAGs yet. Set the DAG for at least one "
- "task and try again: {}".format([self] + task_list)
+ f"Tried to create relationships between tasks that don't have DAGs yet. "
+ f"Set the DAG for at least one task and try again: {[self] + task_list}"
)
if dag and not self.has_dag():
@@ -1779,11 +1767,7 @@ def chain(*tasks: Union[Chainable, Sequence[Chainable]]) -> None:
down_task.set_upstream(up_task)
continue
if not isinstance(up_task, Sequence) or not isinstance(down_task, Sequence):
- raise TypeError(
- 'Chain not supported between instances of {up_type} and {down_type}'.format(
- up_type=type(up_task), down_type=type(down_task)
- )
- )
+ raise TypeError(f'Chain not supported between instances of {type(up_task)} and {type(down_task)}')
up_task_list = up_task
down_task_list = down_task
if len(up_task_list) != len(down_task_list):
diff --git a/airflow/models/connection.py b/airflow/models/connection.py
index 027e59a..3f66d14 100644
--- a/airflow/models/connection.py
+++ b/airflow/models/connection.py
@@ -230,10 +230,8 @@ class Connection(Base, LoggingMixin):
fernet = get_fernet()
if not fernet.is_encrypted:
raise AirflowException(
- "Can't decrypt encrypted password for login={}, \
- FERNET_KEY configuration is missing".format(
- self.login
- )
+ f"Can't decrypt encrypted password for login={self.login} "
+ f"FERNET_KEY configuration is missing"
)
return fernet.decrypt(bytes(self._password, 'utf-8')).decode()
else:
@@ -257,10 +255,8 @@ class Connection(Base, LoggingMixin):
fernet = get_fernet()
if not fernet.is_encrypted:
raise AirflowException(
- "Can't decrypt `extra` params for login={},\
- FERNET_KEY configuration is missing".format(
- self.login
- )
+ f"Can't decrypt `extra` params for login={self.login}, "
+ f"FERNET_KEY configuration is missing"
)
return fernet.decrypt(bytes(self._extra, 'utf-8')).decode()
else:
@@ -324,14 +320,10 @@ class Connection(Base, LoggingMixin):
DeprecationWarning,
stacklevel=2,
)
- return "id: {}. Host: {}, Port: {}, Schema: {}, Login: {}, Password: {}, extra: {}".format(
- self.conn_id,
- self.host,
- self.port,
- self.schema,
- self.login,
- "XXXXXXXX" if self.password else None,
- "XXXXXXXX" if self.extra_dejson else None,
+ return (
+ f"id: {self.conn_id}. Host: {self.host}, Port: {self.port}, Schema: {self.schema}, "
+ f"Login: {self.login}, Password: {'XXXXXXXX' if self.password else None}, "
+ f"extra: {'XXXXXXXX' if self.extra_dejson else None}"
)
def debug_info(self):
@@ -345,14 +337,10 @@ class Connection(Base, LoggingMixin):
DeprecationWarning,
stacklevel=2,
)
- return "id: {}. Host: {}, Port: {}, Schema: {}, Login: {}, Password: {}, extra: {}".format(
- self.conn_id,
- self.host,
- self.port,
- self.schema,
- self.login,
- "XXXXXXXX" if self.password else None,
- self.extra_dejson,
+ return (
+ f"id: {self.conn_id}. Host: {self.host}, Port: {self.port}, Schema: {self.schema}, "
+ f"Login: {self.login}, Password: {'XXXXXXXX' if self.password else None}, "
+ f"extra: {self.extra_dejson}"
)
def test_connection(self):
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 129a30a..bbb1fb7 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -1519,11 +1519,9 @@ class DAG(LoggingMixin):
if recursion_depth + 1 > max_recursion_depth:
# Prevent cycles or accidents.
raise AirflowException(
- "Maximum recursion depth {} reached for {} {}. "
- "Attempted to clear too many tasks "
- "or there may be a cyclic dependency.".format(
- max_recursion_depth, ExternalTaskMarker.__name__, ti.task_id
- )
+ f"Maximum recursion depth {max_recursion_depth} reached for "
+ f"{ExternalTaskMarker.__name__} {ti.task_id}. "
+ f"Attempted to clear too many tasks or there may be a cyclic dependency."
)
ti.render_templates()
external_tis = (
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 2563109..c3f3004 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1078,9 +1078,7 @@ class TaskInstance(Base, LoggingMixin):
# deterministic per task instance
ti_hash = int(
hashlib.sha1(
- "{}#{}#{}#{}".format(
- self.dag_id, self.task_id, self.execution_date, self.try_number
- ).encode('utf-8')
+ f"{self.dag_id}#{self.task_id}#{self.execution_date}#{self.try_number}".encode()
).hexdigest(),
16,
)
@@ -2225,8 +2223,8 @@ class TaskInstance(Base, LoggingMixin):
self_execution_date = self.get_dagrun(session).execution_date
if execution_date and execution_date < self_execution_date:
raise ValueError(
- 'execution_date can not be in the past (current '
- 'execution_date is {}; received {})'.format(self_execution_date, execution_date)
+ f'execution_date can not be in the past (current execution_date is '
+ f'{self_execution_date}; received {execution_date})'
)
XCom.set(
diff --git a/airflow/operators/subdag.py b/airflow/operators/subdag.py
index cc2bdad..35223cb 100644
--- a/airflow/operators/subdag.py
+++ b/airflow/operators/subdag.py
@@ -101,10 +101,8 @@ class SubDagOperator(BaseSensorOperator):
if dag.dag_id + '.' + kwargs['task_id'] != self.subdag.dag_id:
raise AirflowException(
- "The subdag's dag_id should have the form '{{parent_dag_id}}.{{this_task_id}}'. "
- "Expected '{d}.{t}'; received '{rcvd}'.".format(
- d=dag.dag_id, t=kwargs['task_id'], rcvd=self.subdag.dag_id
- )
+ f"The subdag's dag_id should have the form '{{parent_dag_id}}.{{this_task_id}}'. "
+ f"Expected '{dag.dag_id}.{kwargs['task_id']}'; received '{self.subdag.dag_id}'."
)
def _validate_pool(self, session):
diff --git a/airflow/operators/trigger_dagrun.py b/airflow/operators/trigger_dagrun.py
index 1e6cb7f..2a32a33 100644
--- a/airflow/operators/trigger_dagrun.py
+++ b/airflow/operators/trigger_dagrun.py
@@ -107,8 +107,7 @@ class TriggerDagRunOperator(BaseOperator):
if not isinstance(execution_date, (str, datetime.datetime, type(None))):
raise TypeError(
- "Expected str or datetime.datetime type for execution_date."
- "Got {}".format(type(execution_date))
+ f"Expected str or datetime.datetime type for execution_date.Got {type(execution_date)}"
)
self.execution_date: Optional[datetime.datetime] = execution_date # type: ignore
diff --git a/airflow/providers/amazon/aws/hooks/batch_client.py b/airflow/providers/amazon/aws/hooks/batch_client.py
index 3c92406..da78ecd 100644
--- a/airflow/providers/amazon/aws/hooks/batch_client.py
+++ b/airflow/providers/amazon/aws/hooks/batch_client.py
@@ -390,8 +390,8 @@ class AwsBatchClientHook(AwsBaseHook):
retries += 1
if retries >= self.status_retries:
raise AirflowException(
- "AWS Batch job ({}) description error: exceeded "
- "status_retries ({})".format(job_id, self.status_retries)
+ f"AWS Batch job ({job_id}) description error: exceeded status_retries "
+ f"({self.status_retries})"
)
pause = self.exponential_delay(retries)
diff --git a/airflow/providers/amazon/aws/hooks/s3.py b/airflow/providers/amazon/aws/hooks/s3.py
index 77c2638..84f3794 100644
--- a/airflow/providers/amazon/aws/hooks/s3.py
+++ b/airflow/providers/amazon/aws/hooks/s3.py
@@ -560,9 +560,8 @@ class S3Hook(AwsBaseHook):
available_compressions = ['gzip']
if compression is not None and compression not in available_compressions:
raise NotImplementedError(
- "Received {} compression type. String "
- "can currently be compressed in {} "
- "only.".format(compression, available_compressions)
+ f"Received {compression} compression type. "
+ f"String can currently be compressed in {available_compressions} only."
)
if compression == 'gzip':
bytes_data = gz.compress(bytes_data)
diff --git a/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py b/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
index 0b0a103..634c4e0 100644
--- a/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
+++ b/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
@@ -101,9 +101,8 @@ class CloudwatchTaskHandler(FileTaskHandler, LoggingMixin):
def _read(self, task_instance, try_number, metadata=None):
stream_name = self._render_filename(task_instance, try_number)
return (
- '*** Reading remote log from Cloudwatch log_group: {} log_stream: {}.\n{}\n'.format(
- self.log_group, stream_name, self.get_cloudwatch_logs(stream_name=stream_name)
- ),
+ f'*** Reading remote log from Cloudwatch log_group: {self.log_group} '
+ f'log_stream: {stream_name}.\n{self.get_cloudwatch_logs(stream_name=stream_name)}\n',
{'end_of_log': True},
)
@@ -123,9 +122,7 @@ class CloudwatchTaskHandler(FileTaskHandler, LoggingMixin):
return '\n'.join(self._event_to_str(event) for event in events)
except Exception:
- msg = 'Could not read remote logs from log_group: {} log_stream: {}.'.format(
- self.log_group, stream_name
- )
+ msg = f'Could not read remote logs from log_group: {self.log_group} log_stream: {stream_name}.'
self.log.exception(msg)
return msg
diff --git a/airflow/providers/amazon/aws/operators/athena.py b/airflow/providers/amazon/aws/operators/athena.py
index 1883364..a6e0596 100644
--- a/airflow/providers/amazon/aws/operators/athena.py
+++ b/airflow/providers/amazon/aws/operators/athena.py
@@ -112,16 +112,13 @@ class AWSAthenaOperator(BaseOperator):
if query_status in AWSAthenaHook.FAILURE_STATES:
error_message = self.hook.get_state_change_reason(self.query_execution_id)
raise Exception(
- 'Final state of Athena job is {}, query_execution_id is {}. Error: {}'.format(
- query_status, self.query_execution_id, error_message
- )
+ f'Final state of Athena job is {query_status}, query_execution_id is '
+ f'{self.query_execution_id}. Error: {error_message}'
)
elif not query_status or query_status in AWSAthenaHook.INTERMEDIATE_STATES:
raise Exception(
- 'Final state of Athena job is {}. '
- 'Max tries of poll status exceeded, query_execution_id is {}.'.format(
- query_status, self.query_execution_id
- )
+ f'Final state of Athena job is {query_status}. Max tries of poll status exceeded, '
+ f'query_execution_id is {self.query_execution_id}.'
)
return self.query_execution_id
diff --git a/airflow/providers/amazon/aws/operators/datasync.py b/airflow/providers/amazon/aws/operators/datasync.py
index 8fe4cc1..2d7376c 100644
--- a/airflow/providers/amazon/aws/operators/datasync.py
+++ b/airflow/providers/amazon/aws/operators/datasync.py
@@ -179,10 +179,9 @@ class AWSDataSyncOperator(BaseOperator):
valid = True
if not valid:
raise AirflowException(
- "Either specify task_arn or both source_location_uri and destination_location_uri. "
- "task_arn={} source_location_uri={} destination_location_uri={}".format(
- task_arn, source_location_uri, destination_location_uri
- )
+ f"Either specify task_arn or both source_location_uri and destination_location_uri. "
+ f"task_arn={task_arn!r}, source_location_uri={source_location_uri!r}, "
+ f"destination_location_uri={destination_location_uri!r}"
)
# Others
diff --git a/airflow/providers/amazon/aws/sensors/s3_keys_unchanged.py b/airflow/providers/amazon/aws/sensors/s3_keys_unchanged.py
index c74eff5..374c163 100644
--- a/airflow/providers/amazon/aws/sensors/s3_keys_unchanged.py
+++ b/airflow/providers/amazon/aws/sensors/s3_keys_unchanged.py
@@ -143,8 +143,8 @@ class S3KeysUnchangedSensor(BaseSensorOperator):
return False
raise AirflowException(
- "Illegal behavior: objects were deleted in %s between pokes."
- % os.path.join(self.bucket_name, self.prefix)
+ f"Illegal behavior: objects were deleted in"
+ f" {os.path.join(self.bucket_name, self.prefix)} between pokes."
)
if self.last_activity_time:
diff --git a/airflow/providers/apache/hive/hooks/hive.py b/airflow/providers/apache/hive/hooks/hive.py
index 71c2d31..ce2e4d9 100644
--- a/airflow/providers/apache/hive/hooks/hive.py
+++ b/airflow/providers/apache/hive/hooks/hive.py
@@ -107,8 +107,7 @@ class HiveCliHook(BaseHook):
mapred_queue_priority = mapred_queue_priority.upper()
if mapred_queue_priority not in HIVE_QUEUE_PRIORITIES:
raise AirflowException(
- "Invalid Mapred Queue Priority. Valid values are: "
- "{}".format(', '.join(HIVE_QUEUE_PRIORITIES))
+ f"Invalid Mapred Queue Priority. Valid values are: {', '.join(HIVE_QUEUE_PRIORITIES)}"
)
self.mapred_queue = mapred_queue or conf.get('hive', 'default_hive_mapred_queue')
@@ -697,10 +696,8 @@ class HiveMetastoreHook(BaseHook):
is_subset = set(filter_map.keys()).issubset(set(part_specs[0].keys()))
if filter_map and not is_subset:
raise AirflowException(
- "Keys in provided filter_map {} "
- "are not subset of part_spec keys: {}".format(
- ', '.join(filter_map.keys()), ', '.join(part_specs[0].keys())
- )
+ f"Keys in provided filter_map {', '.join(filter_map.keys())} "
+ f"are not subset of part_spec keys: {', '.join(part_specs[0].keys())}"
)
candidates = [
@@ -912,11 +909,9 @@ class HiveServer2Hook(DbApiHook):
):
description = cur.description
if previous_description and previous_description != description:
- message = '''The statements are producing different descriptions:
- Current: {}
- Previous: {}'''.format(
- repr(description), repr(previous_description)
- )
+ message = f'''The statements are producing different descriptions:
+ Current: {repr(description)}
+ Previous: {repr(previous_description)}'''
raise ValueError(message)
elif not previous_description:
previous_description = description
diff --git a/airflow/providers/apache/kylin/operators/kylin_cube.py b/airflow/providers/apache/kylin/operators/kylin_cube.py
index 76b5708..8048e69 100644
--- a/airflow/providers/apache/kylin/operators/kylin_cube.py
+++ b/airflow/providers/apache/kylin/operators/kylin_cube.py
@@ -151,9 +151,7 @@ class KylinCubeOperator(BaseOperator):
_support_invoke_command = kylinpy.CubeSource.support_invoke_command
if self.command.lower() not in _support_invoke_command:
raise AirflowException(
- 'Kylin:Command {} can not match kylin command list {}'.format(
- self.command, _support_invoke_command
- )
+ f'Kylin:Command {self.command} can not match kylin command list {_support_invoke_command}'
)
kylinpy_params = {
diff --git a/airflow/providers/apache/livy/hooks/livy.py b/airflow/providers/apache/livy/hooks/livy.py
index d4fe598..e05ed16 100644
--- a/airflow/providers/apache/livy/hooks/livy.py
+++ b/airflow/providers/apache/livy/hooks/livy.py
@@ -154,9 +154,8 @@ class LivyHook(HttpHook, LoggingMixin):
response.raise_for_status()
except requests.exceptions.HTTPError as err:
raise AirflowException(
- "Could not submit batch. Status code: {}. Message: '{}'".format(
- err.response.status_code, err.response.text
- )
+ "Could not submit batch. "
+ f"Status code: {err.response.status_code}. Message: '{err.response.text}'"
)
batch_id = self._parse_post_response(response.json())
@@ -236,9 +235,7 @@ class LivyHook(HttpHook, LoggingMixin):
except requests.exceptions.HTTPError as err:
self.log.warning("Got status code %d for session %d", err.response.status_code, session_id)
raise AirflowException(
- "Could not kill the batch with session id: {}. Message: {}".format(
- session_id, err.response.text
- )
+ f"Could not kill the batch with session id: {session_id}. Message: {err.response.text}"
)
return response.json()
@@ -264,9 +261,8 @@ class LivyHook(HttpHook, LoggingMixin):
except requests.exceptions.HTTPError as err:
self.log.warning("Got status code %d for session %d", err.response.status_code, session_id)
raise AirflowException(
- "Could not fetch the logs for batch with session id: {}. Message: {}".format(
- session_id, err.response.text
- )
+ f"Could not fetch the logs for batch with session id: {session_id}. "
+ f"Message: {err.response.text}"
)
return response.json()
diff --git a/airflow/providers/apache/spark/hooks/spark_sql.py b/airflow/providers/apache/spark/hooks/spark_sql.py
index 3a91f2f..4ce2e09 100644
--- a/airflow/providers/apache/spark/hooks/spark_sql.py
+++ b/airflow/providers/apache/spark/hooks/spark_sql.py
@@ -191,9 +191,8 @@ class SparkSqlHook(BaseHook):
if returncode:
raise AirflowException(
- "Cannot execute '{}' on {} (additional parameters: '{}'). Process exit code: {}.".format(
- self._sql, self._master, cmd, returncode
- )
+ f"Cannot execute '{self._sql}' on {self._master} (additional parameters: '{cmd}'). "
+ f"Process exit code: {returncode}."
)
def kill(self) -> None:
diff --git a/airflow/providers/apache/spark/hooks/spark_submit.py b/airflow/providers/apache/spark/hooks/spark_submit.py
index 7d00c9d..b17eebe 100644
--- a/airflow/providers/apache/spark/hooks/spark_submit.py
+++ b/airflow/providers/apache/spark/hooks/spark_submit.py
@@ -180,9 +180,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
self._is_kubernetes = 'k8s' in self._connection['master']
if self._is_kubernetes and kube_client is None:
raise RuntimeError(
- "{} specified by kubernetes dependencies are not installed!".format(
- self._connection['master']
- )
+ f"{self._connection['master']} specified by kubernetes dependencies are not installed!"
)
self._should_track_driver_status = self._resolve_should_track_driver_status()
@@ -440,15 +438,12 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
if returncode or (self._is_kubernetes and self._spark_exit_code != 0):
if self._is_kubernetes:
raise AirflowException(
- "Cannot execute: {}. Error code is: {}. Kubernetes spark exit code is: {}".format(
- self._mask_cmd(spark_submit_cmd), returncode, self._spark_exit_code
- )
+ f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}. Error code is: {returncode}. "
+ f"Kubernetes spark exit code is: {self._spark_exit_code}"
)
else:
raise AirflowException(
- "Cannot execute: {}. Error code is: {}.".format(
- self._mask_cmd(spark_submit_cmd), returncode
- )
+ f"Cannot execute: {self._mask_cmd(spark_submit_cmd)}. Error code is: {returncode}."
)
self.log.debug("Should track driver: %s", self._should_track_driver_status)
@@ -468,9 +463,7 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
if self._driver_status != "FINISHED":
raise AirflowException(
- "ERROR : Driver {} badly exited with status {}".format(
- self._driver_id, self._driver_status
- )
+ f"ERROR : Driver {self._driver_id} badly exited with status {self._driver_status}"
)
def _process_spark_submit_log(self, itr: Iterator[Any]) -> None:
@@ -601,9 +594,8 @@ class SparkSubmitHook(BaseHook, LoggingMixin):
missed_job_status_reports += 1
else:
raise AirflowException(
- "Failed to poll for the driver status {} times: returncode = {}".format(
- max_missed_job_status_reports, returncode
- )
+ f"Failed to poll for the driver status {max_missed_job_status_reports} times: "
+ f"returncode = {returncode}"
)
def _build_spark_driver_kill_command(self) -> List[str]:
diff --git a/airflow/providers/dingding/hooks/dingding.py b/airflow/providers/dingding/hooks/dingding.py
index 1a73416..02437f2 100644
--- a/airflow/providers/dingding/hooks/dingding.py
+++ b/airflow/providers/dingding/hooks/dingding.py
@@ -115,8 +115,7 @@ class DingdingHook(HttpHook):
support_type = ['text', 'link', 'markdown', 'actionCard', 'feedCard']
if self.message_type not in support_type:
raise ValueError(
- 'DingdingWebhookHook only support {} '
- 'so far, but receive {}'.format(support_type, self.message_type)
+ f'DingdingWebhookHook only support {support_type} so far, but receive {self.message_type}'
)
data = self._build_message()
diff --git a/airflow/providers/google/cloud/example_dags/example_functions.py b/airflow/providers/google/cloud/example_dags/example_functions.py
index 03749ba..6144c07 100644
--- a/airflow/providers/google/cloud/example_dags/example_functions.py
+++ b/airflow/providers/google/cloud/example_dags/example_functions.py
@@ -52,18 +52,15 @@ from airflow.utils import dates
GCP_PROJECT_ID = os.environ.get('GCP_PROJECT_ID', 'example-project')
GCP_LOCATION = os.environ.get('GCP_LOCATION', 'europe-west1')
-GCF_SHORT_FUNCTION_NAME = os.environ.get('GCF_SHORT_FUNCTION_NAME', 'hello').replace(
- "-", "_"
-) # make sure there are no dashes in function name (!)
-FUNCTION_NAME = 'projects/{}/locations/{}/functions/{}'.format(
- GCP_PROJECT_ID, GCP_LOCATION, GCF_SHORT_FUNCTION_NAME
-)
+# make sure there are no dashes in function name (!)
+GCF_SHORT_FUNCTION_NAME = os.environ.get('GCF_SHORT_FUNCTION_NAME', 'hello').replace("-", "_")
+FUNCTION_NAME = f'projects/{GCP_PROJECT_ID}/locations/{GCP_LOCATION}/functions/{GCF_SHORT_FUNCTION_NAME}'
GCF_SOURCE_ARCHIVE_URL = os.environ.get('GCF_SOURCE_ARCHIVE_URL', '')
GCF_SOURCE_UPLOAD_URL = os.environ.get('GCF_SOURCE_UPLOAD_URL', '')
GCF_SOURCE_REPOSITORY = os.environ.get(
'GCF_SOURCE_REPOSITORY',
- 'https://source.developers.google.com/'
- 'projects/{}/repos/hello-world/moveable-aliases/master'.format(GCP_PROJECT_ID),
+ f'https://source.developers.google.com/projects/{GCP_PROJECT_ID}/'
+ f'repos/hello-world/moveable-aliases/master',
)
GCF_ZIP_PATH = os.environ.get('GCF_ZIP_PATH', '')
GCF_ENTRYPOINT = os.environ.get('GCF_ENTRYPOINT', 'helloWorld')
diff --git a/airflow/providers/google/cloud/hooks/bigquery.py b/airflow/providers/google/cloud/hooks/bigquery.py
index 1aa95ed..853457e 100644
--- a/airflow/providers/google/cloud/hooks/bigquery.py
+++ b/airflow/providers/google/cloud/hooks/bigquery.py
@@ -1002,8 +1002,7 @@ class BigQueryHook(GoogleBaseHook, DbApiHook):
project_id = project_id or self.project_id
if not dataset_id or not isinstance(dataset_id, str):
raise ValueError(
- "dataset_id argument must be provided and has "
- "a type 'str'. You provided: {}".format(dataset_id)
+ f"dataset_id argument must be provided and has a type 'str'. You provided: {dataset_id}"
)
service = self.get_service()
@@ -1802,8 +1801,8 @@ class BigQueryHook(GoogleBaseHook, DbApiHook):
]
if source_format not in allowed_formats:
raise ValueError(
- "{} is not a valid source format. "
- "Please use one of the following types: {}".format(source_format, allowed_formats)
+ f"{source_format} is not a valid source format. "
+ f"Please use one of the following types: {allowed_formats}."
)
# bigquery also allows you to define how you want a table's schema to change
@@ -1813,10 +1812,8 @@ class BigQueryHook(GoogleBaseHook, DbApiHook):
allowed_schema_update_options = ['ALLOW_FIELD_ADDITION', "ALLOW_FIELD_RELAXATION"]
if not set(allowed_schema_update_options).issuperset(set(schema_update_options)):
raise ValueError(
- "{} contains invalid schema update options."
- "Please only use one or more of the following options: {}".format(
- schema_update_options, allowed_schema_update_options
- )
+ f"{schema_update_options} contains invalid schema update options. "
+ f"Please only use one or more of the following options: {allowed_schema_update_options}"
)
destination_project, destination_dataset, destination_table = _split_tablename(
@@ -2228,9 +2225,8 @@ class BigQueryHook(GoogleBaseHook, DbApiHook):
if not set(allowed_schema_update_options).issuperset(set(schema_update_options)):
raise ValueError(
- "{} contains invalid schema update options. "
- "Please only use one or more of the following "
- "options: {}".format(schema_update_options, allowed_schema_update_options)
+ f"{schema_update_options} contains invalid schema update options."
+ f" Please only use one or more of the following options: {allowed_schema_update_options}"
)
if schema_update_options:
@@ -2964,10 +2960,7 @@ def _split_tablename(
return f"Format exception for {var_name}: "
if table_input.count('.') + table_input.count(':') > 3:
- raise Exception(
- '{var}Use either : or . to specify project '
- 'got {input}'.format(var=var_print(var_name), input=table_input)
- )
+ raise Exception(f'{var_print(var_name)}Use either : or . to specify project got {table_input}')
cmpt = table_input.rsplit(':', 1)
project_id = None
rest = table_input
@@ -2980,8 +2973,7 @@ def _split_tablename(
rest = cmpt[1]
else:
raise Exception(
- '{var}Expect format of (<project:)<dataset>.<table>, '
- 'got {input}'.format(var=var_print(var_name), input=table_input)
+ f'{var_print(var_name)}Expect format of (<project:)<dataset>.<table>, got {table_input}'
)
cmpt = rest.split('.')
@@ -2997,8 +2989,7 @@ def _split_tablename(
table_id = cmpt[1]
else:
raise Exception(
- '{var}Expect format of (<project.|<project:)<dataset>.<table>, '
- 'got {input}'.format(var=var_print(var_name), input=table_input)
+ f'{var_print(var_name)}Expect format of (<project.|<project:)<dataset>.<table>, got {table_input}'
)
if project_id is None:
diff --git a/airflow/providers/google/cloud/hooks/cloud_sql.py b/airflow/providers/google/cloud/hooks/cloud_sql.py
index ce77671..b157ab1 100644
--- a/airflow/providers/google/cloud/hooks/cloud_sql.py
+++ b/airflow/providers/google/cloud/hooks/cloud_sql.py
@@ -505,8 +505,8 @@ class CloudSqlProxyRunner(LoggingMixin):
file.write(response.content)
if response.status_code != 200:
raise AirflowException(
- "The cloud-sql-proxy could not be downloaded. Status code = {}. "
- "Reason = {}".format(response.status_code, response.reason)
+ "The cloud-sql-proxy could not be downloaded. "
+ f"Status code = {response.status_code}. Reason = {response.reason}"
)
self.log.info("Moving sql_proxy binary from %s to %s", proxy_path_tmp, self.sql_proxy_path)
@@ -788,9 +788,8 @@ class CloudSQLDatabaseHook(BaseHook):
raise AirflowException("The required extra 'instance' is empty or None")
if self.database_type not in CLOUD_SQL_VALID_DATABASE_TYPES:
raise AirflowException(
- "Invalid database type '{}'. Must be one of {}".format(
- self.database_type, CLOUD_SQL_VALID_DATABASE_TYPES
- )
+ f"Invalid database type '{self.database_type}'. "
+ f"Must be one of {CLOUD_SQL_VALID_DATABASE_TYPES}"
)
if self.use_proxy and self.use_ssl:
raise AirflowException(
@@ -821,17 +820,15 @@ class CloudSQLDatabaseHook(BaseHook):
suffix = "/.s.PGSQL.5432"
else:
suffix = ""
- expected_path = "{}/{}:{}:{}{}".format(
- self._generate_unique_path(), self.project_id, self.instance, self.database, suffix
+ expected_path = (
+ f"{self._generate_unique_path()}/{self.project_id}:{self.instance}:{self.database}{suffix}"
)
if len(expected_path) > UNIX_PATH_MAX:
self.log.info("Too long (%s) path: %s", len(expected_path), expected_path)
raise AirflowException(
- "The UNIX socket path length cannot exceed {} characters "
- "on Linux system. Either use shorter instance/database "
- "name or switch to TCP connection. "
- "The socket path for Cloud SQL proxy is now:"
- "{}".format(UNIX_PATH_MAX, expected_path)
+ f"The UNIX socket path length cannot exceed {UNIX_PATH_MAX} characters on Linux system. "
+ "Either use shorter instance/database name or switch to TCP connection. "
+ f"The socket path for Cloud SQL proxy is now:{expected_path}"
)
@staticmethod
@@ -874,10 +871,7 @@ class CloudSQLDatabaseHook(BaseHook):
format_string = proxy_uris['tcp']
else:
format_string = proxy_uris['socket']
- socket_path = "{sql_proxy_socket_path}/{instance_socket_name}".format(
- sql_proxy_socket_path=self.sql_proxy_unique_path,
- instance_socket_name=self._get_instance_socket_name(),
- )
+ socket_path = f"{self.sql_proxy_unique_path}/{self._get_instance_socket_name()}"
else:
public_uris = database_uris['public'] # type: Dict[str, str]
if self.use_ssl:
diff --git a/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py b/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py
index 328a289..dfb6fe6 100644
--- a/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py
+++ b/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py
@@ -476,10 +476,8 @@ class CloudDataTransferServiceHook(GoogleBaseHook):
body[target_key] = body.get(target_key, self.project_id)
if not body.get(target_key):
raise AirflowException(
- "The project id must be passed either as `{}` key in `{}` parameter or as project_id "
- "extra in Google Cloud connection definition. Both are not set!".format(
- target_key, param_name
- )
+ f"The project id must be passed either as `{target_key}` key in `{param_name}` "
+ f"parameter or as project_id extra in Google Cloud connection definition. Both are not set!"
)
return body
diff --git a/airflow/providers/google/cloud/hooks/dataflow.py b/airflow/providers/google/cloud/hooks/dataflow.py
index 5562e8a..751e656 100644
--- a/airflow/providers/google/cloud/hooks/dataflow.py
+++ b/airflow/providers/google/cloud/hooks/dataflow.py
@@ -498,8 +498,8 @@ class _DataflowJobsController(LoggingMixin):
)
batch.execute()
if self._cancel_timeout and isinstance(self._cancel_timeout, int):
- timeout_error_message = "Canceling jobs failed due to timeout ({}s): {}".format(
- self._cancel_timeout, ", ".join(job_ids)
+ timeout_error_message = (
+ f"Canceling jobs failed due to timeout ({self._cancel_timeout}s): {', '.join(job_ids)}"
)
with timeout(seconds=self._cancel_timeout, error_message=timeout_error_message):
self._wait_for_states({DataflowJobStatus.JOB_STATE_CANCELLED})
@@ -896,9 +896,8 @@ class DataflowHook(GoogleBaseHook):
if not re.match(r"^[a-z]([-a-z0-9]*[a-z0-9])?$", base_job_name):
raise ValueError(
- "Invalid job_name ({}); the name must consist of"
- "only the characters [-a-z0-9], starting with a "
- "letter and ending with a letter or number ".format(base_job_name)
+ f"Invalid job_name ({base_job_name}); the name must consist ofonly the characters "
+ f"[-a-z0-9], starting with a letter and ending with a letter or number "
)
if append_job_name:
diff --git a/airflow/providers/google/cloud/hooks/dlp.py b/airflow/providers/google/cloud/hooks/dlp.py
index a03f2f1..2b43006 100644
--- a/airflow/providers/google/cloud/hooks/dlp.py
+++ b/airflow/providers/google/cloud/hooks/dlp.py
@@ -276,9 +276,8 @@ class CloudDLPHook(GoogleBaseHook):
time.sleep(time_to_sleep_in_seconds)
else:
raise AirflowException(
- "Stopped polling DLP job state. DLP job {} state: {}.".format(
- job.name, DlpJob.JobState.Name(job.state)
- )
+ "Stopped polling DLP job state. "
+ f"DLP job {job.name} state: {DlpJob.JobState.Name(job.state)}."
)
return job
diff --git a/airflow/providers/google/cloud/hooks/gcs.py b/airflow/providers/google/cloud/hooks/gcs.py
index 5aba40f..68cfe11 100644
--- a/airflow/providers/google/cloud/hooks/gcs.py
+++ b/airflow/providers/google/cloud/hooks/gcs.py
@@ -183,8 +183,8 @@ class GCSHook(GoogleBaseHook):
if source_bucket == destination_bucket and source_object == destination_object:
raise ValueError(
- 'Either source/destination bucket or source/destination object '
- 'must be different, not both the same: bucket=%s, object=%s' % (source_bucket, source_object)
+ f'Either source/destination bucket or source/destination object must be different, '
+ f'not both the same: bucket={source_bucket}, object={source_object}'
)
if not source_bucket or not source_object:
raise ValueError('source_bucket and source_object cannot be empty.')
@@ -232,8 +232,8 @@ class GCSHook(GoogleBaseHook):
destination_object = destination_object or source_object
if source_bucket == destination_bucket and source_object == destination_object:
raise ValueError(
- 'Either source/destination bucket or source/destination object '
- 'must be different, not both the same: bucket=%s, object=%s' % (source_bucket, source_object)
+ f'Either source/destination bucket or source/destination object must be different, '
+ f'not both the same: bucket={source_bucket}, object={source_object}'
)
if not source_bucket or not source_object:
raise ValueError('source_bucket and source_object cannot be empty.')
diff --git a/airflow/providers/google/cloud/hooks/pubsub.py b/airflow/providers/google/cloud/hooks/pubsub.py
index 8d33578..4ea0b9a 100644
--- a/airflow/providers/google/cloud/hooks/pubsub.py
+++ b/airflow/providers/google/cloud/hooks/pubsub.py
@@ -610,9 +610,7 @@ class PubSubHook(GoogleBaseHook):
)
except (HttpError, GoogleAPICallError) as e:
raise PubSubException(
- 'Error acknowledging {} messages pulled from subscription {}'.format(
- len(ack_ids), subscription_path
- ),
+ f'Error acknowledging {len(ack_ids)} messages pulled from subscription {subscription_path}',
e,
)
diff --git a/airflow/providers/google/cloud/operators/bigtable.py b/airflow/providers/google/cloud/operators/bigtable.py
index 90b9d20..a69b7d9 100644
--- a/airflow/providers/google/cloud/operators/bigtable.py
+++ b/airflow/providers/google/cloud/operators/bigtable.py
@@ -443,9 +443,7 @@ class BigtableCreateTableOperator(BaseOperator, BigtableValidationMixin):
instance = hook.get_instance(project_id=self.project_id, instance_id=self.instance_id)
if not instance:
raise AirflowException(
- "Dependency: instance '{}' does not exist in project '{}'.".format(
- self.instance_id, self.project_id
- )
+ f"Dependency: instance '{self.instance_id}' does not exist in project '{self.project_id}'."
)
try:
hook.create_table(
@@ -621,9 +619,7 @@ class BigtableUpdateClusterOperator(BaseOperator, BigtableValidationMixin):
hook.update_cluster(instance=instance, cluster_id=self.cluster_id, nodes=self.nodes)
except google.api_core.exceptions.NotFound:
raise AirflowException(
- "Dependency: cluster '{}' does not exist for instance '{}'.".format(
- self.cluster_id, self.instance_id
- )
+ f"Dependency: cluster '{self.cluster_id}' does not exist for instance '{self.instance_id}'."
)
except google.api_core.exceptions.GoogleAPICallError as e:
self.log.error('An error occurred. Exiting.')
diff --git a/airflow/providers/google/cloud/operators/cloud_sql.py b/airflow/providers/google/cloud/operators/cloud_sql.py
index 91086b3..8f2cd1b 100644
--- a/airflow/providers/google/cloud/operators/cloud_sql.py
+++ b/airflow/providers/google/cloud/operators/cloud_sql.py
@@ -442,8 +442,8 @@ class CloudSQLInstancePatchOperator(CloudSQLBaseOperator):
)
if not self._check_if_instance_exists(self.instance, hook):
raise AirflowException(
- 'Cloud SQL instance with ID {} does not exist. '
- 'Please specify another instance to patch.'.format(self.instance)
+ f'Cloud SQL instance with ID {self.instance} does not exist. '
+ 'Please specify another instance to patch.'
)
else:
return hook.patch_instance(project_id=self.project_id, body=self.body, instance=self.instance)
@@ -699,11 +699,8 @@ class CloudSQLPatchInstanceDatabaseOperator(CloudSQLBaseOperator):
)
if not self._check_if_db_exists(self.database, hook):
raise AirflowException(
- "Cloud SQL instance with ID {instance} does not contain "
- "database '{database}'. "
- "Please specify another database to patch.".format(
- instance=self.instance, database=self.database
- )
+ f"Cloud SQL instance with ID {self.instance} does not contain database '{self.database}'. "
+ "Please specify another database to patch."
)
else:
return hook.patch_database(
@@ -786,8 +783,8 @@ class CloudSQLDeleteInstanceDatabaseOperator(CloudSQLBaseOperator):
)
if not self._check_if_db_exists(self.database, hook):
print(
- "Cloud SQL instance with ID {} does not contain database '{}'. "
- "Aborting database delete.".format(self.instance, self.database)
+ f"Cloud SQL instance with ID {self.instance!r} does not contain database {self.database!r}. "
+ f"Aborting database delete."
)
return True
else:
diff --git a/airflow/providers/google/cloud/operators/compute.py b/airflow/providers/google/cloud/operators/compute.py
index 694e5d1..b35f06b 100644
--- a/airflow/providers/google/cloud/operators/compute.py
+++ b/airflow/providers/google/cloud/operators/compute.py
@@ -408,8 +408,8 @@ class ComputeEngineCopyInstanceTemplateOperator(ComputeEngineBaseOperator):
self._field_validator = None # Optional[GcpBodyFieldValidator]
if 'name' not in self.body_patch:
raise AirflowException(
- "The body '{}' should contain at least "
- "name for the new operator in the 'name' field".format(body_patch)
+ f"The body '{body_patch}' should contain at least name for the new operator "
+ f"in the 'name' field"
)
if validate_body:
self._field_validator = GcpBodyFieldValidator(
diff --git a/airflow/providers/google/cloud/operators/dataflow.py b/airflow/providers/google/cloud/operators/dataflow.py
index fbe7aaa..9056b70 100644
--- a/airflow/providers/google/cloud/operators/dataflow.py
+++ b/airflow/providers/google/cloud/operators/dataflow.py
@@ -368,9 +368,8 @@ class DataflowCreateJavaJobOperator(BaseOperator):
) -> None:
# TODO: Remove one day
warnings.warn(
- "The `{cls}` operator is deprecated, please use "
- "`providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator` instead."
- "".format(cls=self.__class__.__name__),
+ f"The `{self.__class__.__name__}` operator is deprecated, "
+ f"please use `providers.apache.beam.operators.beam.BeamRunJavaPipelineOperator` instead.",
DeprecationWarning,
stacklevel=2,
)
@@ -1075,9 +1074,8 @@ class DataflowCreatePythonJobOperator(BaseOperator):
) -> None:
# TODO: Remove one day
warnings.warn(
- "The `{cls}` operator is deprecated, please use "
- "`providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator` instead."
- "".format(cls=self.__class__.__name__),
+ f"The `{self.__class__.__name__}` operator is deprecated, "
+ "please use `providers.apache.beam.operators.beam.BeamRunPythonPipelineOperator` instead.",
DeprecationWarning,
stacklevel=2,
)
diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py
index fb4d8ed..7835d5b 100644
--- a/airflow/providers/google/cloud/operators/dataproc.py
+++ b/airflow/providers/google/cloud/operators/dataproc.py
@@ -293,9 +293,7 @@ class ClusterGenerator:
def _build_gce_cluster_config(self, cluster_data):
if self.zone:
- zone_uri = 'https://www.googleapis.com/compute/v1/projects/{}/zones/{}'.format(
- self.project_id, self.zone
- )
+ zone_uri = f'https://www.googleapis.com/compute/v1/projects/{self.project_id}/zones/{self.zone}'
cluster_data['gce_cluster_config']['zone_uri'] = zone_uri
if self.metadata:
@@ -393,8 +391,8 @@ class ClusterGenerator:
elif self.custom_image:
project_id = self.custom_image_project_id or self.project_id
custom_image_url = (
- 'https://www.googleapis.com/compute/beta/projects/'
- '{}/global/images/{}'.format(project_id, self.custom_image)
+ f'https://www.googleapis.com/compute/beta/projects/{project_id}'
+ f'/global/images/{self.custom_image}'
)
cluster_data['master_config']['image_uri'] = custom_image_url
if not self.single_node:
@@ -550,10 +548,10 @@ class DataprocCreateClusterOperator(BaseOperator):
# TODO: remove one day
if cluster_config is None:
warnings.warn(
- "Passing cluster parameters by keywords to `{}` "
- "will be deprecated. Please provide cluster_config object using `cluster_config` parameter. "
- "You can use `airflow.dataproc.ClusterGenerator.generate_cluster` method to "
- "obtain cluster object.".format(type(self).__name__),
+ f"Passing cluster parameters by keywords to `{type(self).__name__}` will be deprecated. "
+ "Please provide cluster_config object using `cluster_config` parameter. "
+ "You can use `airflow.dataproc.ClusterGenerator.generate_cluster` "
+ "method to obtain cluster object.",
DeprecationWarning,
stacklevel=1,
)
@@ -773,9 +771,8 @@ class DataprocScaleClusterOperator(BaseOperator):
# TODO: Remove one day
warnings.warn(
- "The `{cls}` operator is deprecated, please use `DataprocUpdateClusterOperator` instead.".format(
- cls=type(self).__name__
- ),
+ f"The `{type(self).__name__}` operator is deprecated, "
+ "please use `DataprocUpdateClusterOperator` instead.",
DeprecationWarning,
stacklevel=1,
)
diff --git a/airflow/providers/google/cloud/operators/functions.py b/airflow/providers/google/cloud/operators/functions.py
index 4403431..a9bd568 100644
--- a/airflow/providers/google/cloud/operators/functions.py
+++ b/airflow/providers/google/cloud/operators/functions.py
@@ -286,15 +286,15 @@ class ZipPathPreprocessor:
self.upload_function = True
else:
raise AirflowException(
- "Only one of '{}' in body or '{}' argument "
- "allowed. Found both.".format(GCF_SOURCE_UPLOAD_URL, GCF_ZIP_PATH)
+ f"Only one of '{GCF_SOURCE_UPLOAD_URL}' in body or '{GCF_ZIP_PATH}' argument allowed. "
+ f"Found both."
)
def _verify_archive_url_and_zip_path(self) -> None:
if GCF_SOURCE_ARCHIVE_URL in self.body and self.zip_path:
raise AirflowException(
- "Only one of '{}' in body or '{}' argument "
- "allowed. Found both.".format(GCF_SOURCE_ARCHIVE_URL, GCF_ZIP_PATH)
+ f"Only one of '{GCF_SOURCE_ARCHIVE_URL}' in body or '{GCF_ZIP_PATH}' argument allowed. "
+ f"Found both."
)
def should_upload_function(self) -> bool:
@@ -304,7 +304,7 @@ class ZipPathPreprocessor:
:rtype: bool
"""
if self.upload_function is None:
- raise AirflowException('validate() method has to be invoked before ' 'should_upload_function')
+ raise AirflowException('validate() method has to be invoked before should_upload_function')
return self.upload_function
def preprocess_body(self) -> None:
diff --git a/airflow/providers/google/cloud/operators/mlengine.py b/airflow/providers/google/cloud/operators/mlengine.py
index c227f8f..d511bfb 100644
--- a/airflow/providers/google/cloud/operators/mlengine.py
+++ b/airflow/providers/google/cloud/operators/mlengine.py
@@ -258,8 +258,8 @@ class MLEngineStartBatchPredictionJobOperator(BaseOperator):
if not self._version_name:
prediction_request['predictionInput']['modelName'] = origin_name
else:
- prediction_request['predictionInput']['versionName'] = origin_name + '/versions/{}'.format(
- self._version_name
+ prediction_request['predictionInput']['versionName'] = (
+ origin_name + f'/versions/{self._version_name}'
)
if self._max_worker_count:
diff --git a/airflow/providers/google/cloud/operators/spanner.py b/airflow/providers/google/cloud/operators/spanner.py
index 19d904c..fc84994 100644
--- a/airflow/providers/google/cloud/operators/spanner.py
+++ b/airflow/providers/google/cloud/operators/spanner.py
@@ -491,9 +491,9 @@ class SpannerUpdateDatabaseInstanceOperator(BaseOperator):
project_id=self.project_id, instance_id=self.instance_id, database_id=self.database_id
):
raise AirflowException(
- "The Cloud Spanner database '{}' in project '{}' and "
- "instance '{}' is missing. Create the database first "
- "before you can update it.".format(self.database_id, self.project_id, self.instance_id)
+ f"The Cloud Spanner database '{self.database_id}' in project '{self.project_id}' "
+ f"and instance '{self.instance_id}' is missing. "
+ f"Create the database first before you can update it."
)
else:
return hook.update_database(
diff --git a/airflow/providers/google/cloud/sensors/gcs.py b/airflow/providers/google/cloud/sensors/gcs.py
index 4b30f0f..51ffa5a 100644
--- a/airflow/providers/google/cloud/sensors/gcs.py
+++ b/airflow/providers/google/cloud/sensors/gcs.py
@@ -18,6 +18,7 @@
"""This module contains Google Cloud Storage sensors."""
import os
+import textwrap
import warnings
from datetime import datetime
from typing import Callable, List, Optional, Sequence, Set, Union
@@ -385,22 +386,21 @@ class GCSUploadSessionCompleteSensor(BaseSensorOperator):
self.previous_objects = current_objects
self.last_activity_time = get_time()
self.log.warning(
- """
+ textwrap.dedent(
+ """\
Objects were deleted during the last
poke interval. Updating the file counter and
resetting last_activity_time.
- %s
- """,
+ %s\
+ """
+ ),
self.previous_objects - current_objects,
)
return False
raise AirflowException(
- """
- Illegal behavior: objects were deleted in {} between pokes.
- """.format(
- os.path.join(self.bucket, self.prefix)
- )
+ "Illegal behavior: objects were deleted in "
+ f"{os.path.join(self.bucket, self.prefix)} between pokes."
)
if self.last_activity_time:
@@ -415,10 +415,13 @@ class GCSUploadSessionCompleteSensor(BaseSensorOperator):
if current_num_objects >= self.min_objects:
self.log.info(
- """SUCCESS:
- Sensor found %s objects at %s.
- Waited at least %s seconds, with no new objects dropped.
- """,
+ textwrap.dedent(
+ """\
+ SUCCESS:
+ Sensor found %s objects at %s.
+ Waited at least %s seconds, with no new objects dropped.
+ """
+ ),
current_num_objects,
path,
self.inactivity_period,
diff --git a/airflow/providers/google/cloud/transfers/gcs_to_gcs.py b/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
index 7656ad8..43060ae 100644
--- a/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
+++ b/airflow/providers/google/cloud/transfers/gcs_to_gcs.py
@@ -238,9 +238,8 @@ class GCSToGCSOperator(BaseOperator):
)
if self.source_objects and self.source_object:
error_msg = (
- "You can either set source_object parameter or source_objects "
- "parameter but not both. Found source_object={} and"
- " source_objects={}".format(self.source_object, self.source_objects)
+ f"You can either set source_object parameter or source_objects parameter but not both. "
+ f"Found source_object={self.source_object} and source_objects={self.source_objects}"
)
raise AirflowException(error_msg)
@@ -279,8 +278,6 @@ class GCSToGCSOperator(BaseOperator):
def _copy_source_without_wildcard(self, hook, prefix):
"""
-
-
For source_objects with no wildcard, this operator would first list
all files in source_objects, using provided delimiter if any. Then copy
files from source_objects to destination_object and rename each source
@@ -342,7 +339,7 @@ class GCSToGCSOperator(BaseOperator):
if total_wildcards > 1:
error_msg = (
"Only one wildcard '*' is allowed in source_object parameter. "
- "Found {} in {}.".format(total_wildcards, prefix)
+ f"Found {total_wildcards} in {prefix}."
)
raise AirflowException(error_msg)
diff --git a/airflow/providers/google/cloud/transfers/gcs_to_sftp.py b/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
index 3504a89..41bcc59 100644
--- a/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
+++ b/airflow/providers/google/cloud/transfers/gcs_to_sftp.py
@@ -150,7 +150,7 @@ class GCSToSFTPOperator(BaseOperator):
if total_wildcards > 1:
raise AirflowException(
"Only one wildcard '*' is allowed in source_object parameter. "
- "Found {} in {}.".format(total_wildcards, self.source_object)
+ f"Found {total_wildcards} in {self.source_object}."
)
prefix, delimiter = self.source_object.split(WILDCARD, 1)
diff --git a/airflow/providers/google/cloud/transfers/sftp_to_gcs.py b/airflow/providers/google/cloud/transfers/sftp_to_gcs.py
index 35d0741..e624ad0 100644
--- a/airflow/providers/google/cloud/transfers/sftp_to_gcs.py
+++ b/airflow/providers/google/cloud/transfers/sftp_to_gcs.py
@@ -127,7 +127,7 @@ class SFTPToGCSOperator(BaseOperator):
if total_wildcards > 1:
raise AirflowException(
"Only one wildcard '*' is allowed in source_path parameter. "
- "Found {} in {}.".format(total_wildcards, self.source_path)
+ f"Found {total_wildcards} in {self.source_path}."
)
prefix, delimiter = self.source_path.split(WILDCARD, 1)
diff --git a/airflow/providers/google/cloud/utils/field_validator.py b/airflow/providers/google/cloud/utils/field_validator.py
index 876edc4..a02dcc4 100644
--- a/airflow/providers/google/cloud/utils/field_validator.py
+++ b/airflow/providers/google/cloud/utils/field_validator.py
@@ -216,27 +216,25 @@ class GcpBodyFieldValidator(LoggingMixin):
)
if regexp and field_type:
raise GcpValidationSpecificationException(
- "The validation specification entry '{}' has both type and regexp. "
- "The regexp is only allowed without type (i.e. assume type is 'str' "
- "that can be validated with regexp)".format(full_field_path)
+ f"The validation specification entry '{full_field_path}' has both type and regexp. "
+ "The regexp is only allowed without type (i.e. assume type is 'str' that can be "
+ "validated with regexp)"
)
if allow_empty is not None and field_type:
raise GcpValidationSpecificationException(
- "The validation specification entry '{}' has both type and allow_empty. "
- "The allow_empty is only allowed without type (i.e. assume type is 'str' "
- "that can be validated with allow_empty)".format(full_field_path)
+ f"The validation specification entry '{full_field_path}' has both type and allow_empty. "
+ "The allow_empty is only allowed without type (i.e. assume type is 'str' that can "
+ "be validated with allow_empty)"
)
if children_validation_specs and field_type not in COMPOSITE_FIELD_TYPES:
raise GcpValidationSpecificationException(
- "Nested fields are specified in field '{}' of type '{}'. "
- "Nested fields are only allowed for fields of those types: ('{}').".format(
- full_field_path, field_type, COMPOSITE_FIELD_TYPES
- )
+ f"Nested fields are specified in field '{full_field_path}' of type '{field_type}'. "
+ f"Nested fields are only allowed for fields of those types: ('{COMPOSITE_FIELD_TYPES}')."
)
if custom_validation and field_type:
raise GcpValidationSpecificationException(
- "The validation specification field '{}' has both type and "
- "custom_validation. Custom validation is only allowed without type.".format(full_field_path)
+ f"The validation specification field '{full_field_path}' has both type and "
+ f"custom_validation. Custom validation is only allowed without type."
)
@staticmethod
@@ -244,8 +242,8 @@ class GcpBodyFieldValidator(LoggingMixin):
if not re.match(regexp, value):
# Note matching of only the beginning as we assume the regexps all-or-nothing
raise GcpFieldValidationException(
- "The body field '{}' of value '{}' does not match the field "
- "specification regexp: '{}'.".format(full_field_path, value, regexp)
+ f"The body field '{full_field_path}' of value '{value}' does not match the field "
+ f"specification regexp: '{regexp}'."
)
@staticmethod
@@ -291,10 +289,8 @@ class GcpBodyFieldValidator(LoggingMixin):
field_name = child_validation_spec['name']
if new_field_found and field_found:
raise GcpFieldValidationException(
- "The mutually exclusive fields '{}' and '{}' belonging to the "
- "union '{}' are both present. Please remove one".format(
- field_name, found_field_name, full_field_path
- )
+ f"The mutually exclusive fields '{field_name}' and '{found_field_name}' belonging to "
+ f"the union '{full_field_path}' are both present. Please remove one"
)
if new_field_found:
field_found = True
@@ -372,8 +368,8 @@ class GcpBodyFieldValidator(LoggingMixin):
elif field_type == 'dict':
if not isinstance(value, dict):
raise GcpFieldValidationException(
- "The field '{}' should be of dictionary type according to the "
- "specification '{}' but it is '{}'".format(full_field_path, validation_spec, value)
+ f"The field '{full_field_path}' should be of dictionary type according to "
+ f"the specification '{validation_spec}' but it is '{value}'"
)
if children_validation_specs is None:
self.log.debug(
@@ -388,25 +384,24 @@ class GcpBodyFieldValidator(LoggingMixin):
elif field_type == 'union':
if not children_validation_specs:
raise GcpValidationSpecificationException(
- "The union field '{}' has no nested fields "
- "defined in specification '{}'. Unions should have at least one "
- "nested field defined.".format(full_field_path, validation_spec)
+ f"The union field '{full_field_path}' has no nested fields defined in "
+ f"specification '{validation_spec}'. "
+ "Unions should have at least one nested field defined."
)
self._validate_union(children_validation_specs, full_field_path, dictionary_to_validate)
elif field_type == 'list':
if not isinstance(value, list):
raise GcpFieldValidationException(
- "The field '{}' should be of list type according to the "
- "specification '{}' but it is '{}'".format(full_field_path, validation_spec, value)
+ f"The field '{full_field_path}' should be of list type according to "
+ f"the specification '{validation_spec}' but it is '{value}'"
)
elif custom_validation:
try:
custom_validation(value)
except Exception as e:
raise GcpFieldValidationException(
- "Error while validating custom field '{}' specified by '{}': '{}'".format(
- full_field_path, validation_spec, e
- )
+ f"Error while validating custom field '{full_field_path}' "
+ f"specified by '{validation_spec}': '{e}'"
)
elif field_type is None:
self.log.debug(
@@ -416,8 +411,8 @@ class GcpBodyFieldValidator(LoggingMixin):
)
else:
raise GcpValidationSpecificationException(
- "The field '{}' is of type '{}' in specification '{}'."
- "This type is unknown to validation!".format(full_field_path, field_type, validation_spec)
+ f"The field '{full_field_path}' is of type '{field_type}' in "
+ f"specification '{validation_spec}'.This type is unknown to validation!"
)
return True
diff --git a/airflow/providers/google/suite/hooks/sheets.py b/airflow/providers/google/suite/hooks/sheets.py
index ce4218c..5a3e0d5 100644
--- a/airflow/providers/google/suite/hooks/sheets.py
+++ b/airflow/providers/google/suite/hooks/sheets.py
@@ -271,11 +271,8 @@ class GSheetsHook(GoogleBaseHook):
"""
if len(ranges) != len(values):
raise AirflowException(
- "'Ranges' and 'Lists' must be of equal length. \n \
- 'Ranges' is of length: {} and \n \
- 'Values' is of length: {}.".format(
- str(len(ranges)), str(len(values))
- )
+ f"'Ranges' and 'Lists' must be of equal length. "
+ f"'Ranges' is of length: {len(ranges)} and 'Values' is of length: {len(values)}."
)
service = self.get_conn()
data = []
diff --git a/airflow/providers/google/suite/transfers/gcs_to_gdrive.py b/airflow/providers/google/suite/transfers/gcs_to_gdrive.py
index b75611d..b35a71d 100644
--- a/airflow/providers/google/suite/transfers/gcs_to_gdrive.py
+++ b/airflow/providers/google/suite/transfers/gcs_to_gdrive.py
@@ -130,7 +130,7 @@ class GCSToGoogleDriveOperator(BaseOperator):
if total_wildcards > 1:
error_msg = (
"Only one wildcard '*' is allowed in source_object parameter. "
- "Found {} in {}.".format(total_wildcards, self.source_object)
+ f"Found {total_wildcards} in {self.source_object}."
)
raise AirflowException(error_msg)
diff --git a/airflow/providers/grpc/hooks/grpc.py b/airflow/providers/grpc/hooks/grpc.py
index e58fdc4..e7e0bc7 100644
--- a/airflow/providers/grpc/hooks/grpc.py
+++ b/airflow/providers/grpc/hooks/grpc.py
@@ -116,9 +116,8 @@ class GrpcHook(BaseHook):
channel = self.custom_connection_func(self.conn)
else:
raise AirflowConfigException(
- "auth_type not supported or not provided, channel cannot be established,\
- given value: %s"
- % str(auth_type)
+ "auth_type not supported or not provided, channel cannot be established, "
+ f"given value: {str(auth_type)}"
)
if self.interceptors:
diff --git a/airflow/providers/jenkins/operators/jenkins_job_trigger.py b/airflow/providers/jenkins/operators/jenkins_job_trigger.py
index 155070b..b514c2c 100644
--- a/airflow/providers/jenkins/operators/jenkins_job_trigger.py
+++ b/airflow/providers/jenkins/operators/jenkins_job_trigger.py
@@ -220,9 +220,8 @@ class JenkinsJobTriggerOperator(BaseOperator):
# Check if job ended with not allowed state.
if build_info['result'] not in self.allowed_jenkins_states:
raise AirflowException(
- 'Jenkins job failed, final state : %s.'
- 'Find more information on job url : %s'
- % (build_info['result'], build_info['url'])
+ f"Jenkins job failed, final state : {build_info['result']}. "
+ f"Find more information on job url : {build_info['url']}"
)
else:
self.log.info('Waiting for job to complete : %s , build %s', self.job_name, build_number)
diff --git a/airflow/providers/microsoft/azure/operators/azure_batch.py b/airflow/providers/microsoft/azure/operators/azure_batch.py
index 78b9c46..d58513b 100644
--- a/airflow/providers/microsoft/azure/operators/azure_batch.py
+++ b/airflow/providers/microsoft/azure/operators/azure_batch.py
@@ -220,10 +220,9 @@ class AzureBatchOperator(BaseOperator):
if self.use_latest_image:
if not all(elem for elem in [self.vm_publisher, self.vm_offer]):
raise AirflowException(
- "If use_latest_image_and_sku is"
- " set to True then the parameters vm_publisher, vm_offer, "
- "must all be set. Found "
- "vm_publisher={}, vm_offer={}".format(self.vm_publisher, self.vm_offer)
+ f"If use_latest_image_and_sku is set to True then the parameters vm_publisher, "
+ f"vm_offer, must all be set. "
+ f"Found vm_publisher={self.vm_publisher}, vm_offer={self.vm_offer}"
)
if self.vm_publisher:
if not all([self.vm_sku, self.vm_offer, self.vm_node_agent_sku_id]):
@@ -242,12 +241,10 @@ class AzureBatchOperator(BaseOperator):
if self.enable_auto_scale:
if self.target_dedicated_nodes or self.target_low_priority_nodes:
raise AirflowException(
- "If enable_auto_scale is set, then the parameters "
- "target_dedicated_nodes and target_low_priority_nodes must not "
- "be set. Found target_dedicated_nodes={},"
- " target_low_priority_nodes={}".format(
- self.target_dedicated_nodes, self.target_low_priority_nodes
- )
+ f"If enable_auto_scale is set, then the parameters target_dedicated_nodes and "
+ f"target_low_priority_nodes must not be set. Found "
+ f"target_dedicated_nodes={self.target_dedicated_nodes}, "
+ f"target_low_priority_nodes={self.target_low_priority_nodes}"
)
if not self.auto_scale_formula:
raise AirflowException("The auto_scale_formula is required when enable_auto_scale is set")
diff --git a/airflow/providers/microsoft/winrm/operators/winrm.py b/airflow/providers/microsoft/winrm/operators/winrm.py
index caf0470..8abd5d5 100644
--- a/airflow/providers/microsoft/winrm/operators/winrm.py
+++ b/airflow/providers/microsoft/winrm/operators/winrm.py
@@ -153,7 +153,8 @@ class WinRMOperator(BaseOperator):
else:
return b64encode(b''.join(stdout_buffer)).decode(self.output_encoding)
else:
- error_msg = "Error running cmd: {}, return code: {}, error: {}".format(
- self.command, return_code, b''.join(stderr_buffer).decode(self.output_encoding)
+ stderr_output = b''.join(stderr_buffer).decode(self.output_encoding)
+ error_msg = (
+ f"Error running cmd: {self.command}, return code: {return_code}, error: {stderr_output}"
)
raise AirflowException(error_msg)
diff --git a/airflow/providers/mysql/transfers/vertica_to_mysql.py b/airflow/providers/mysql/transfers/vertica_to_mysql.py
index c1eecab..c1e7df9 100644
--- a/airflow/providers/mysql/transfers/vertica_to_mysql.py
+++ b/airflow/providers/mysql/transfers/vertica_to_mysql.py
@@ -132,9 +132,9 @@ class VerticaToMySqlOperator(BaseOperator):
with closing(mysql.get_conn()) as conn:
with closing(conn.cursor()) as cursor:
cursor.execute(
- "LOAD DATA LOCAL INFILE '%s' INTO "
- "TABLE %s LINES TERMINATED BY '\r\n' (%s)"
- % (tmpfile.name, self.mysql_table, ", ".join(selected_columns))
+ f"LOAD DATA LOCAL INFILE '{tmpfile.name}' "
+ f"INTO TABLE {self.mysql_table} "
+ f"LINES TERMINATED BY '\r\n' ({', '.join(selected_columns)})"
)
conn.commit()
tmpfile.close()
diff --git a/airflow/providers/pagerduty/hooks/pagerduty_events.py b/airflow/providers/pagerduty/hooks/pagerduty_events.py
index f133df2..d66755f 100644
--- a/airflow/providers/pagerduty/hooks/pagerduty_events.py
+++ b/airflow/providers/pagerduty/hooks/pagerduty_events.py
@@ -148,9 +148,8 @@ class PagerdutyEventsHook(BaseHook):
data["dedup_key"] = dedup_key
elif action != 'trigger':
raise ValueError(
- "The dedup_key property is required for event_action=%s events, and it must \
- be a string."
- % action
+ f"The dedup_key property is required for event_action={action} events, "
+ f"and it must be a string."
)
if images is not None:
data["images"] = images
diff --git a/airflow/providers/plexus/hooks/plexus.py b/airflow/providers/plexus/hooks/plexus.py
index 21e8069..6c1d7bc 100644
--- a/airflow/providers/plexus/hooks/plexus.py
+++ b/airflow/providers/plexus/hooks/plexus.py
@@ -53,8 +53,8 @@ class PlexusHook(BaseHook):
response = requests.post(token_endpoint, data={"email": login, "password": pwd}, timeout=5)
if not response.ok:
raise AirflowException(
- "Could not retrieve JWT Token. Status Code: [{}]. "
- "Reason: {} - {}".format(response.status_code, response.reason, response.text)
+ "Could not retrieve JWT Token. "
+ f"Status Code: [{response.status_code}]. Reason: {response.reason} - {response.text}"
)
token = response.json()["access"]
payload = jwt.decode(token, verify=False)
diff --git a/airflow/providers/plexus/operators/job.py b/airflow/providers/plexus/operators/job.py
index e20d64a..1a42837 100644
--- a/airflow/providers/plexus/operators/job.py
+++ b/airflow/providers/plexus/operators/job.py
@@ -87,8 +87,8 @@ class PlexusJobOperator(BaseOperator):
get_job = requests.get(jid_endpoint, headers=headers, timeout=5)
if not get_job.ok:
raise AirflowException(
- "Could not retrieve job status. Status Code: [{}]. "
- "Reason: {} - {}".format(get_job.status_code, get_job.reason, get_job.text)
+ "Could not retrieve job status. "
+ f"Status Code: [{get_job.status_code}]. Reason: {get_job.reason} - {get_job.text}"
)
new_state = get_job.json()["last_state"]
if new_state in ("Cancelled", "Failed"):
@@ -98,8 +98,8 @@ class PlexusJobOperator(BaseOperator):
state = new_state
else:
raise AirflowException(
- "Could not start job. Status Code: [{}]. "
- "Reason: {} - {}".format(create_job.status_code, create_job.reason, create_job.text)
+ "Could not start job. "
+ f"Status Code: [{create_job.status_code}]. Reason: {create_job.reason} - {create_job.text}"
)
def _api_lookup(self, param: str, hook):
diff --git a/airflow/providers/postgres/hooks/postgres.py b/airflow/providers/postgres/hooks/postgres.py
index 67cc8b3..6e4ab68 100644
--- a/airflow/providers/postgres/hooks/postgres.py
+++ b/airflow/providers/postgres/hooks/postgres.py
@@ -269,8 +269,5 @@ class PostgresHook(DbApiHook):
replace_target = [
"{0} = excluded.{0}".format(col) for col in target_fields if col not in replace_index_set
]
- sql += " ON CONFLICT ({}) DO UPDATE SET {}".format(
- ", ".join(replace_index),
- ", ".join(replace_target),
- )
+ sql += f" ON CONFLICT ({', '.join(replace_index)}) DO UPDATE SET {', '.join(replace_target)}"
return sql
diff --git a/airflow/providers/qubole/operators/qubole_check.py b/airflow/providers/qubole/operators/qubole_check.py
index 329cbc8..b91515e 100644
--- a/airflow/providers/qubole/operators/qubole_check.py
+++ b/airflow/providers/qubole/operators/qubole_check.py
@@ -197,11 +197,7 @@ def handle_airflow_exception(airflow_exception, hook: QuboleCheckHook):
qubole_command_results = hook.get_query_results()
qubole_command_id = cmd.id
exception_message = (
- '\nQubole Command Id: {qubole_command_id}'
- '\nQubole Command Results:'
- '\n{qubole_command_results}'.format(
- qubole_command_id=qubole_command_id, qubole_command_results=qubole_command_results
- )
+ f'\nQubole Command Id: {qubole_command_id}\nQubole Command Results:\n{qubole_command_results}'
)
raise AirflowException(str(airflow_exception) + exception_message)
raise AirflowException(str(airflow_exception))
diff --git a/airflow/providers/qubole/sensors/qubole.py b/airflow/providers/qubole/sensors/qubole.py
index 8936fe8..ba42106 100644
--- a/airflow/providers/qubole/sensors/qubole.py
+++ b/airflow/providers/qubole/sensors/qubole.py
@@ -37,8 +37,8 @@ class QuboleSensor(BaseSensorOperator):
if 'poke_interval' in kwargs and kwargs['poke_interval'] < 5:
raise AirflowException(
- "Sorry, poke_interval can't be less than 5 sec for "
- "task '{}' in dag '{}'.".format(kwargs['task_id'], kwargs['dag'].dag_id)
+ f"Sorry, poke_interval can't be less than 5 sec for task '{kwargs['task_id']}' "
+ f"in dag '{kwargs['dag'].dag_id}'."
)
super().__init__(**kwargs)
diff --git a/airflow/providers/sftp/operators/sftp.py b/airflow/providers/sftp/operators/sftp.py
index cb2f24f..3a1d023a 100644
--- a/airflow/providers/sftp/operators/sftp.py
+++ b/airflow/providers/sftp/operators/sftp.py
@@ -104,9 +104,8 @@ class SFTPOperator(BaseOperator):
self.create_intermediate_dirs = create_intermediate_dirs
if not (self.operation.lower() == SFTPOperation.GET or self.operation.lower() == SFTPOperation.PUT):
raise TypeError(
- "unsupported operation value {}, expected {} or {}".format(
- self.operation, SFTPOperation.GET, SFTPOperation.PUT
- )
+ f"Unsupported operation value {self.operation}, "
+ f"expected {SFTPOperation.GET} or {SFTPOperation.PUT}."
)
def execute(self, context: Any) -> str:
diff --git a/airflow/sensors/base.py b/airflow/sensors/base.py
index 0019c41..2698516 100644
--- a/airflow/sensors/base.py
+++ b/airflow/sensors/base.py
@@ -122,13 +122,7 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
raise AirflowException("The timeout must be a non-negative number")
if self.mode not in self.valid_modes:
raise AirflowException(
- "The mode must be one of {valid_modes},"
- "'{d}.{t}'; received '{m}'.".format(
- valid_modes=self.valid_modes,
- d=self.dag.dag_id if self.has_dag() else "",
- t=self.task_id,
- m=self.mode,
- )
+ f"The mode must be one of {self.valid_modes},'{self.dag.dag_id if self.has_dag() else ''}.{self.task_id}'; received '{self.mode}'."
)
def poke(self, context: Dict) -> bool:
diff --git a/airflow/sensors/external_task.py b/airflow/sensors/external_task.py
index c451001..ef68c77 100644
--- a/airflow/sensors/external_task.py
+++ b/airflow/sensors/external_task.py
@@ -113,8 +113,8 @@ class ExternalTaskSensor(BaseSensorOperator):
if set(self.failed_states).intersection(set(self.allowed_states)):
raise AirflowException(
- "Duplicate values provided as allowed "
- "`{}` and failed states `{}`".format(self.allowed_states, self.failed_states)
+ f"Duplicate values provided as allowed "
+ f"`{self.allowed_states}` and failed states `{self.failed_states}`"
)
if external_task_id is not None and external_task_ids is not None:
diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
index 3d30ca4..9d550c9 100644
--- a/airflow/serialization/serialized_objects.py
+++ b/airflow/serialization/serialized_objects.py
@@ -737,13 +737,11 @@ class SerializedBaseOperator(BaseOperator, BaseSerialization):
op_link_arguments = cattr.unstructure(operator_extra_link)
if not isinstance(op_link_arguments, dict):
op_link_arguments = {}
- serialize_operator_extra_links.append(
- {
- "{}.{}".format(
- operator_extra_link.__class__.__module__, operator_extra_link.__class__.__name__
- ): op_link_arguments
- }
+
+ module_path = (
+ f"{operator_extra_link.__class__.__module__}.{operator_extra_link.__class__.__name__}"
)
+ serialize_operator_extra_links.append({module_path: op_link_arguments})
return serialize_operator_extra_links
diff --git a/airflow/stats.py b/airflow/stats.py
index 0a7004d..e3b069b 100644
--- a/airflow/stats.py
+++ b/airflow/stats.py
@@ -19,7 +19,6 @@
import logging
import socket
import string
-import textwrap
import time
from functools import wraps
from typing import TYPE_CHECKING, Callable, Optional, TypeVar, cast
@@ -189,24 +188,11 @@ def stat_name_default_handler(stat_name, max_length=250) -> str:
raise InvalidStatsNameException('The stat_name has to be a string')
if len(stat_name) > max_length:
raise InvalidStatsNameException(
- textwrap.dedent(
- """\
- The stat_name ({stat_name}) has to be less than {max_length} characters.
- """.format(
- stat_name=stat_name, max_length=max_length
- )
- )
+ f"The stat_name ({stat_name}) has to be less than {max_length} characters."
)
if not all((c in ALLOWED_CHARACTERS) for c in stat_name):
raise InvalidStatsNameException(
- textwrap.dedent(
- """\
- The stat name ({stat_name}) has to be composed with characters in
- {allowed_characters}.
- """.format(
- stat_name=stat_name, allowed_characters=ALLOWED_CHARACTERS
- )
- )
+ f"The stat name ({stat_name}) has to be composed with characters in {ALLOWED_CHARACTERS}."
)
return stat_name
diff --git a/airflow/task/task_runner/cgroup_task_runner.py b/airflow/task/task_runner/cgroup_task_runner.py
index 04b3a23..15075de 100644
--- a/airflow/task/task_runner/cgroup_task_runner.py
+++ b/airflow/task/task_runner/cgroup_task_runner.py
@@ -133,9 +133,7 @@ class CgroupTaskRunner(BaseTaskRunner):
return
# Create a unique cgroup name
- cgroup_name = "airflow/{}/{}".format(
- datetime.datetime.utcnow().strftime("%Y-%m-%d"), str(uuid.uuid4())
- )
+ cgroup_name = f"airflow/{datetime.datetime.utcnow().strftime('%Y-%m-%d')}/{str(uuid.uuid4())}"
self.mem_cgroup_name = f"memory/{cgroup_name}"
self.cpu_cgroup_name = f"cpu/{cgroup_name}"
diff --git a/airflow/ti_deps/deps/dag_ti_slots_available_dep.py b/airflow/ti_deps/deps/dag_ti_slots_available_dep.py
index ba75c4d..e9dadd1 100644
--- a/airflow/ti_deps/deps/dag_ti_slots_available_dep.py
+++ b/airflow/ti_deps/deps/dag_ti_slots_available_dep.py
@@ -30,6 +30,8 @@ class DagTISlotsAvailableDep(BaseTIDep):
def _get_dep_statuses(self, ti, session, dep_context):
if ti.task.dag.get_concurrency_reached(session):
yield self._failing_status(
- reason="The maximum number of running tasks ({}) for this task's DAG "
- "'{}' has been reached.".format(ti.task.dag.max_active_tasks, ti.dag_id)
+ reason=(
+ f"The maximum number of running tasks ({ti.task.dag.max_active_tasks}) for "
+ f"this task's DAG '{ti.dag_id}' has been reached."
+ )
)
diff --git a/airflow/ti_deps/deps/dagrun_exists_dep.py b/airflow/ti_deps/deps/dagrun_exists_dep.py
index 0aa21e8..0910aec 100644
--- a/airflow/ti_deps/deps/dagrun_exists_dep.py
+++ b/airflow/ti_deps/deps/dagrun_exists_dep.py
@@ -32,6 +32,5 @@ class DagrunRunningDep(BaseTIDep):
dr = ti.get_dagrun(session)
if dr.state != State.RUNNING:
yield self._failing_status(
- reason="Task instance's dagrun was not in the 'running' state but in "
- "the state '{}'.".format(dr.state)
+ reason=f"Task instance's dagrun was not in the 'running' state but in the state '{dr.state}'."
)
diff --git a/airflow/ti_deps/deps/exec_date_after_start_date_dep.py b/airflow/ti_deps/deps/exec_date_after_start_date_dep.py
index 7f9233b..a8b3496 100644
--- a/airflow/ti_deps/deps/exec_date_after_start_date_dep.py
+++ b/airflow/ti_deps/deps/exec_date_after_start_date_dep.py
@@ -30,14 +30,16 @@ class ExecDateAfterStartDateDep(BaseTIDep):
def _get_dep_statuses(self, ti, session, dep_context):
if ti.task.start_date and ti.execution_date < ti.task.start_date:
yield self._failing_status(
- reason="The execution date is {} but this is before the task's start "
- "date {}.".format(ti.execution_date.isoformat(), ti.task.start_date.isoformat())
+ reason=(
+ f"The execution date is {ti.execution_date.isoformat()} but this is before "
+ f"the task's start date {ti.task.start_date.isoformat()}."
+ )
)
if ti.task.dag and ti.task.dag.start_date and ti.execution_date < ti.task.dag.start_date:
yield self._failing_status(
- reason="The execution date is {} but this is before the task's "
- "DAG's start date {}.".format(
- ti.execution_date.isoformat(), ti.task.dag.start_date.isoformat()
+ reason=(
+ f"The execution date is {ti.execution_date.isoformat()} but this is "
+ f"before the task's DAG's start date {ti.task.dag.start_date.isoformat()}."
)
)
diff --git a/airflow/ti_deps/deps/not_in_retry_period_dep.py b/airflow/ti_deps/deps/not_in_retry_period_dep.py
index be52522..17ba4d7 100644
--- a/airflow/ti_deps/deps/not_in_retry_period_dep.py
+++ b/airflow/ti_deps/deps/not_in_retry_period_dep.py
@@ -47,7 +47,9 @@ class NotInRetryPeriodDep(BaseTIDep):
next_task_retry_date = ti.next_retry_datetime()
if ti.is_premature:
yield self._failing_status(
- reason="Task is not ready for retry yet but will be retried "
- "automatically. Current date is {} and task will be retried "
- "at {}.".format(cur_date.isoformat(), next_task_retry_date.isoformat())
+ reason=(
+ f"Task is not ready for retry yet but will be retried automatically. "
+ f"Current date is {cur_date.isoformat()} and task will be retried "
+ f"at {next_task_retry_date.isoformat()}."
+ )
)
diff --git a/airflow/ti_deps/deps/prev_dagrun_dep.py b/airflow/ti_deps/deps/prev_dagrun_dep.py
index 70573bf..eecacae 100644
--- a/airflow/ti_deps/deps/prev_dagrun_dep.py
+++ b/airflow/ti_deps/deps/prev_dagrun_dep.py
@@ -74,14 +74,17 @@ class PrevDagrunDep(BaseTIDep):
if previous_ti.state not in {State.SKIPPED, State.SUCCESS}:
yield self._failing_status(
- reason="depends_on_past is true for this task, but the previous task "
- "instance {} is in the state '{}' which is not a successful "
- "state.".format(previous_ti, previous_ti.state)
+ reason=(
+ f"depends_on_past is true for this task, but the previous task instance {previous_ti} "
+ f"is in the state '{previous_ti.state}' which is not a successful state."
+ )
)
previous_ti.task = ti.task
if ti.task.wait_for_downstream and not previous_ti.are_dependents_done(session=session):
yield self._failing_status(
- reason="The tasks downstream of the previous task instance {} haven't "
- "completed (and wait_for_downstream is True).".format(previous_ti)
+ reason=(
+ f"The tasks downstream of the previous task instance {previous_ti} haven't completed "
+ f"(and wait_for_downstream is True)."
+ )
)
diff --git a/airflow/ti_deps/deps/ready_to_reschedule.py b/airflow/ti_deps/deps/ready_to_reschedule.py
index 9a336c2..e867c78 100644
--- a/airflow/ti_deps/deps/ready_to_reschedule.py
+++ b/airflow/ti_deps/deps/ready_to_reschedule.py
@@ -68,7 +68,9 @@ class ReadyToRescheduleDep(BaseTIDep):
return
yield self._failing_status(
- reason="Task is not ready for reschedule yet but will be rescheduled "
- "automatically. Current date is {} and task will be rescheduled "
- "at {}.".format(now.isoformat(), next_reschedule_date.isoformat())
+ reason=(
+ "Task is not ready for reschedule yet but will be rescheduled automatically. "
+ f"Current date is {now.isoformat()} and task will be "
+ f"rescheduled at {next_reschedule_date.isoformat()}."
+ )
)
diff --git a/airflow/ti_deps/deps/runnable_exec_date_dep.py b/airflow/ti_deps/deps/runnable_exec_date_dep.py
index 0607c11..ab26b9e 100644
--- a/airflow/ti_deps/deps/runnable_exec_date_dep.py
+++ b/airflow/ti_deps/deps/runnable_exec_date_dep.py
@@ -36,18 +36,24 @@ class RunnableExecDateDep(BaseTIDep):
logical_date = ti.get_dagrun(session).execution_date
if logical_date > cur_date and not ti.task.dag.allow_future_exec_dates:
yield self._failing_status(
- reason="Execution date {} is in the future (the current "
- "date is {}).".format(logical_date.isoformat(), cur_date.isoformat())
+ reason=(
+ f"Execution date {logical_date.isoformat()} is in the future "
+ f"(the current date is {cur_date.isoformat()})."
+ )
)
if ti.task.end_date and logical_date > ti.task.end_date:
yield self._failing_status(
- reason="The execution date is {} but this is after the task's end date "
- "{}.".format(logical_date.isoformat(), ti.task.end_date.isoformat())
+ reason=(
+ f"The execution date is {logical_date.isoformat()} but this is "
+ f"after the task's end date {ti.task.end_date.isoformat()}."
+ )
)
if ti.task.dag and ti.task.dag.end_date and logical_date > ti.task.dag.end_date:
yield self._failing_status(
- reason="The execution date is {} but this is after the task's DAG's "
- "end date {}.".format(logical_date.isoformat(), ti.task.dag.end_date.isoformat())
+ reason=(
+ f"The execution date is {logical_date.isoformat()} but this is after "
+ f"the task's DAG's end date {ti.task.dag.end_date.isoformat()}."
+ )
)
diff --git a/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow/ti_deps/deps/trigger_rule_dep.py
index 5d72410..e54fc4c 100644
--- a/airflow/ti_deps/deps/trigger_rule_dep.py
+++ b/airflow/ti_deps/deps/trigger_rule_dep.py
@@ -158,78 +158,83 @@ class TriggerRuleDep(BaseTIDep):
if trigger_rule == TR.ONE_SUCCESS:
if successes <= 0:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires one upstream "
- "task success, but none were found. "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires one upstream task success, "
+ f"but none were found. upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.ONE_FAILED:
if not failed and not upstream_failed:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires one upstream "
- "task failure, but none were found. "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires one upstream task failure, "
+ f"but none were found. upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.ALL_SUCCESS:
num_failures = upstream - successes
if num_failures > 0:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to have succeeded, but found {} non-success(es). "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, num_failures, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to have "
+ f"succeeded, but found {num_failures} non-success(es). "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.ALL_FAILED:
num_successes = upstream - failed - upstream_failed
if num_successes > 0:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to have failed, but found {} non-failure(s). "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, num_successes, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to have failed, "
+ f"but found {num_successes} non-failure(s). "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.ALL_DONE:
if not upstream_done:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to have completed, but found {} task(s) that "
- "were not done. upstream_tasks_state={}, "
- "upstream_task_ids={}".format(
- trigger_rule, upstream_done, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to have "
+ f"completed, but found {upstream_done} task(s) that were not done. "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.NONE_FAILED:
num_failures = upstream - successes - skipped
if num_failures > 0:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to have succeeded or been skipped, but found {} non-success(es). "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, num_failures, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to have "
+ f"succeeded or been skipped, but found {num_failures} non-success(es). "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.NONE_FAILED_MIN_ONE_SUCCESS:
num_failures = upstream - successes - skipped
if num_failures > 0:
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to have succeeded or been skipped, but found {} non-success(es). "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, num_failures, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to have "
+ f"succeeded or been skipped, but found {num_failures} non-success(es). "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
elif trigger_rule == TR.NONE_SKIPPED:
if not upstream_done or (skipped > 0):
yield self._failing_status(
- reason="Task's trigger rule '{}' requires all upstream "
- "tasks to not have been skipped, but found {} task(s) skipped. "
- "upstream_tasks_state={}, upstream_task_ids={}".format(
- trigger_rule, skipped, upstream_tasks_state, task.upstream_task_ids
+ reason=(
+ f"Task's trigger rule '{trigger_rule}' requires all upstream tasks to not have been "
+ f"skipped, but found {skipped} task(s) skipped. "
+ f"upstream_tasks_state={upstream_tasks_state}, "
+ f"upstream_task_ids={task.upstream_task_ids}"
)
)
else:
diff --git a/airflow/ti_deps/deps/valid_state_dep.py b/airflow/ti_deps/deps/valid_state_dep.py
index 5a3e243..5ec1402 100644
--- a/airflow/ti_deps/deps/valid_state_dep.py
+++ b/airflow/ti_deps/deps/valid_state_dep.py
@@ -58,6 +58,8 @@ class ValidStateDep(BaseTIDep):
return
yield self._failing_status(
- reason="Task is in the '{}' state which is not a valid state for "
- "execution. The task must be cleared in order to be run.".format(ti.state)
+ reason=(
+ f"Task is in the '{ti.state}' state which is not a valid state for execution. "
+ f"The task must be cleared in order to be run."
+ )
)
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index 6415676..6284605 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -177,8 +177,7 @@ def get_dag_by_file_location(dag_id: str):
dag_model = DagModel.get_current(dag_id)
if dag_model is None:
raise AirflowException(
- 'dag_id could not be found: {}. Either the dag did not exist or it failed to '
- 'parse.'.format(dag_id)
+ f'dag_id could not be found: {dag_id}. Either the dag did not exist or it failed to parse.'
)
dagbag = DagBag(dag_folder=dag_model.fileloc)
return dagbag.dags[dag_id]
@@ -191,8 +190,7 @@ def get_dag(subdir: Optional[str], dag_id: str) -> "DAG":
dagbag = DagBag(process_subdir(subdir))
if dag_id not in dagbag.dags:
raise AirflowException(
- 'dag_id could not be found: {}. Either the dag did not exist or it failed to '
- 'parse.'.format(dag_id)
+ f'dag_id could not be found: {dag_id}. Either the dag did not exist or it failed to parse.'
)
return dagbag.dags[dag_id]
@@ -207,8 +205,8 @@ def get_dags(subdir: Optional[str], dag_id: str, use_regex: bool = False):
matched_dags = [dag for dag in dagbag.dags.values() if re.search(dag_id, dag.dag_id)]
if not matched_dags:
raise AirflowException(
- 'dag_id could not be found with regex: {}. Either the dag did not exist '
- 'or it failed to parse.'.format(dag_id)
+ f'dag_id could not be found with regex: {dag_id}. Either the dag did not exist or '
+ f'it failed to parse.'
)
return matched_dags
diff --git a/airflow/utils/compression.py b/airflow/utils/compression.py
index aa39b03..7307b68 100644
--- a/airflow/utils/compression.py
+++ b/airflow/utils/compression.py
@@ -26,8 +26,7 @@ def uncompress_file(input_file_name, file_extension, dest_dir):
"""Uncompress gz and bz2 files"""
if file_extension.lower() not in ('.gz', '.bz2'):
raise NotImplementedError(
- "Received {} format. Only gz and bz2 "
- "files can currently be uncompressed.".format(file_extension)
+ f"Received {file_extension} format. Only gz and bz2 files can currently be uncompressed."
)
if file_extension.lower() == '.gz':
fmodule = gzip.GzipFile
diff --git a/airflow/utils/helpers.py b/airflow/utils/helpers.py
index e6ab39a..93b892e 100644
--- a/airflow/utils/helpers.py
+++ b/airflow/utils/helpers.py
@@ -50,8 +50,8 @@ def validate_key(k: str, max_length: int = 250):
raise AirflowException(f"The key has to be less than {max_length} characters")
if not KEY_REGEX.match(k):
raise AirflowException(
- "The key ({k}) has to be made of alphanumeric characters, dashes, "
- "dots and underscores exclusively".format(k=k)
+ f"The key ({k}) has to be made of alphanumeric characters, dashes, "
+ f"dots and underscores exclusively"
)
@@ -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!r}) has to be made of alphanumeric characters, dashes and underscores exclusively"
)
diff --git a/airflow/utils/log/file_task_handler.py b/airflow/utils/log/file_task_handler.py
index 6d88c20..8e5dbca 100644
--- a/airflow/utils/log/file_task_handler.py
+++ b/airflow/utils/log/file_task_handler.py
@@ -143,9 +143,7 @@ class FileTaskHandler(logging.Handler):
if len(matches[0]) > len(ti.hostname):
ti.hostname = matches[0]
- log += '*** Trying to get logs (last 100 lines) from worker pod {} ***\n\n'.format(
- ti.hostname
- )
+ log += f'*** Trying to get logs (last 100 lines) from worker pod {ti.hostname} ***\n\n'
res = kube_client.read_namespaced_pod_log(
name=ti.hostname,
diff --git a/airflow/utils/operator_resources.py b/airflow/utils/operator_resources.py
index 8781021..0f855e4 100644
--- a/airflow/utils/operator_resources.py
+++ b/airflow/utils/operator_resources.py
@@ -44,8 +44,8 @@ class Resource:
def __init__(self, name, units_str, qty):
if qty < 0:
raise AirflowException(
- 'Received resource quantity {} for resource {} but resource quantity '
- 'must be non-negative.'.format(qty, name)
+ f'Received resource quantity {qty} for resource {name}, '
+ f'but resource quantity must be non-negative.'
)
self._name = name
diff --git a/airflow/utils/orm_event_handlers.py b/airflow/utils/orm_event_handlers.py
index 1a5da96..23c99d8 100644
--- a/airflow/utils/orm_event_handlers.py
+++ b/airflow/utils/orm_event_handlers.py
@@ -58,8 +58,8 @@ def setup_event_handlers(engine):
if connection_record.info['pid'] != pid:
connection_record.connection = connection_proxy.connection = None
raise exc.DisconnectionError(
- "Connection record belongs to pid {}, "
- "attempting to check out in pid {}".format(connection_record.info['pid'], pid)
+ f"Connection record belongs to pid {connection_record.info['pid']}, "
+ f"attempting to check out in pid {pid}"
)
if conf.getboolean('debug', 'sqlalchemy_stats', fallback=False):
diff --git a/airflow/www/api/experimental/endpoints.py b/airflow/www/api/experimental/endpoints.py
index 91528e9..56b201a 100644
--- a/airflow/www/api/experimental/endpoints.py
+++ b/airflow/www/api/experimental/endpoints.py
@@ -104,8 +104,8 @@ def trigger_dag(dag_id):
execution_date = timezone.parse(execution_date)
except ValueError:
error_message = (
- 'Given execution date, {}, could not be identified '
- 'as a date. Example date format: 2015-11-16T14:34:15+00:00'.format(execution_date)
+ f'Given execution date, {execution_date}, could not be identified as a date. '
+ f'Example date format: 2015-11-16T14:34:15+00:00'
)
log.error(error_message)
response = jsonify({'error': error_message})
@@ -254,8 +254,8 @@ def task_instance_info(dag_id, execution_date, task_id):
execution_date = timezone.parse(execution_date)
except ValueError:
error_message = (
- 'Given execution date, {}, could not be identified '
- 'as a date. Example date format: 2015-11-16T14:34:15+00:00'.format(execution_date)
+ f'Given execution date, {execution_date}, could not be identified as a date. '
+ f'Example date format: 2015-11-16T14:34:15+00:00'
)
log.error(error_message)
response = jsonify({'error': error_message})
@@ -290,8 +290,8 @@ def dag_run_status(dag_id, execution_date):
execution_date = timezone.parse(execution_date)
except ValueError:
error_message = (
- 'Given execution date, {}, could not be identified '
- 'as a date. Example date format: 2015-11-16T14:34:15+00:00'.format(execution_date)
+ f'Given execution date, {execution_date}, could not be identified as a date. '
+ f'Example date format: 2015-11-16T14:34:15+00:00'
)
log.error(error_message)
response = jsonify({'error': error_message})
@@ -403,8 +403,8 @@ def get_lineage(dag_id: str, execution_date: str):
execution_dt = timezone.parse(execution_date)
except ValueError:
error_message = (
- 'Given execution date, {}, could not be identified '
- 'as a date. Example date format: 2015-11-16T14:34:15+00:00'.format(execution_date)
+ f'Given execution date, {execution_date}, could not be identified as a date. '
+ f'Example date format: 2015-11-16T14:34:15+00:00'
)
log.error(error_message)
response = jsonify({'error': error_message})
diff --git a/airflow/www/fab_security/manager.py b/airflow/www/fab_security/manager.py
index 997d70b..32510a6 100644
--- a/airflow/www/fab_security/manager.py
+++ b/airflow/www/fab_security/manager.py
@@ -842,9 +842,7 @@ class BaseSecurityManager:
# build the filter string for the LDAP search
if self.auth_ldap_search_filter:
- filter_str = "(&{}({}={}))".format(
- self.auth_ldap_search_filter, self.auth_ldap_uid_field, username
- )
+ filter_str = f"(&{self.auth_ldap_search_filter}({self.auth_ldap_uid_field}={username}))"
else:
filter_str = f"({self.auth_ldap_uid_field}={username})"
@@ -859,8 +857,7 @@ class BaseSecurityManager:
# perform the LDAP search
log.debug(
- "LDAP search for '%s' with fields %s in scope '%s'"
- % (filter_str, request_fields, self.auth_ldap_search)
+ f"LDAP search for '{filter_str}' with fields {request_fields} in scope '{self.auth_ldap_search}'"
)
raw_search_result = con.search_s(
self.auth_ldap_search, ldap.SCOPE_SUBTREE, filter_str, request_fields
@@ -875,8 +872,8 @@ class BaseSecurityManager:
# only continue if 0 or 1 results were returned
if len(search_result) > 1:
log.error(
- "LDAP search for '%s' in scope '%a' returned multiple results"
- % (filter_str, self.auth_ldap_search)
+ f"LDAP search for '{filter_str}' in scope "
+ f"'{self.auth_ldap_search!a}' returned multiple results"
)
return None, None
diff --git a/airflow/www/security.py b/airflow/www/security.py
index 362e6b7..61944a0 100644
--- a/airflow/www/security.py
+++ b/airflow/www/security.py
@@ -830,17 +830,17 @@ class AirflowSecurityManager(SecurityManager, LoggingMixin):
role = self.find_role(rolename)
if not role:
raise AirflowException(
- "The access_control mapping for DAG '{}' includes a role "
- "named '{}', but that role does not exist".format(dag_id, rolename)
+ f"The access_control mapping for DAG '{dag_id}' includes a role named "
+ f"'{rolename}', but that role does not exist"
)
action_names = set(action_names)
invalid_action_names = action_names - self.DAG_ACTIONS
if invalid_action_names:
raise AirflowException(
- "The access_control map for DAG '{}' includes the following "
- "invalid permissions: {}; The set of valid permissions "
- "is: {}".format(dag_resource_name, invalid_action_names, self.DAG_ACTIONS)
+ f"The access_control map for DAG '{dag_resource_name}' includes "
+ f"the following invalid permissions: {invalid_action_names}; "
+ f"The set of valid permissions is: {self.DAG_ACTIONS}"
)
for action_name in action_names:
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 71fedd6..694d07f 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1200,8 +1200,8 @@ class Airflow(AirflowBaseView):
execution_date = timezone.parse(execution_date)
except ValueError:
error_message = (
- 'Given execution date, {}, could not be identified '
- 'as a date. Example date format: 2015-11-16T14:34:15+00:00'.format(execution_date)
+ f'Given execution date, {execution_date}, could not be identified as a date. '
+ 'Example date format: 2015-11-16T14:34:15+00:00'
)
response = jsonify({'error': error_message})
response.status_code = 400
@@ -1571,8 +1571,7 @@ class Airflow(AirflowBaseView):
if failed_deps:
failed_deps_str = ", ".join(f"{dep.dep_name}: {dep.reason}" for dep in failed_deps)
flash(
- "Could not queue task instance for execution, dependencies not met: "
- "{}".format(failed_deps_str),
+ f"Could not queue task instance for execution, dependencies not met: {failed_deps_str}",
"error",
)
return redirect(origin)
@@ -4008,10 +4007,7 @@ class DagRunModelView(AirflowPrivilegeVerifierModelView):
current_app.dag_bag.get_dag(dr.dag_id), dr.execution_date, commit=True, session=session
)
altered_ti_count = len(altered_tis)
- flash(
- "{count} dag runs and {altered_ti_count} task instances "
- "were set to failed".format(count=count, altered_ti_count=altered_ti_count)
- )
+ flash(f"{count} dag runs and {altered_ti_count} task instances were set to failed")
except Exception:
flash('Failed to set state', 'error')
return redirect(self.get_default_url())
@@ -4035,10 +4031,7 @@ class DagRunModelView(AirflowPrivilegeVerifierModelView):
current_app.dag_bag.get_dag(dr.dag_id), dr.execution_date, commit=True, session=session
)
altered_ti_count = len(altered_tis)
- flash(
- "{count} dag runs and {altered_ti_count} task instances "
- "were set to success".format(count=count, altered_ti_count=altered_ti_count)
- )
+ flash(f"{count} dag runs and {altered_ti_count} task instances were set to success")
except Exception:
flash('Failed to set state', 'error')
return redirect(self.get_default_url())
diff --git a/dev/airflow-github b/dev/airflow-github
index bf89500..027aa20 100755
--- a/dev/airflow-github
+++ b/dev/airflow-github
@@ -190,9 +190,7 @@ def compare(target_version, github_token, previous_version=None, show_uncherrypi
)
print(
- "Commits on branch: {:d}, {:d} ({}) yet to be cherry-picked".format(
- num_cherrypicked, sum(num_uncherrypicked.values()), dict(num_uncherrypicked)
- )
+ f"Commits on branch: {num_cherrypicked:d}, {sum(num_uncherrypicked.values()):d} ({dict(num_uncherrypicked)}) yet to be cherry-picked"
)
diff --git a/docs/exts/docs_build/run_patched_sphinx.py b/docs/exts/docs_build/run_patched_sphinx.py
index 887b982..d93a84f 100755
--- a/docs/exts/docs_build/run_patched_sphinx.py
+++ b/docs/exts/docs_build/run_patched_sphinx.py
@@ -51,8 +51,7 @@ def run_autoapi(app):
for _dir in normalized_dirs:
if not os.path.exists(_dir):
raise ExtensionError(
- "AutoAPI Directory `{dir}` not found. "
- "Please check your `autoapi_dirs` setting.".format(dir=_dir)
+ f"AutoAPI Directory `{_dir}` not found. Please check your `autoapi_dirs` setting."
)
# Change from app.confdir to app.srcdir.
diff --git a/scripts/ci/pre_commit/pre_commit_yaml_to_cfg.py b/scripts/ci/pre_commit/pre_commit_yaml_to_cfg.py
index 5a66d6c..8a22264 100755
--- a/scripts/ci/pre_commit/pre_commit_yaml_to_cfg.py
+++ b/scripts/ci/pre_commit/pre_commit_yaml_to_cfg.py
@@ -126,8 +126,8 @@ def _write_option(configfile, idx, option):
if option["default"] is not None:
if not isinstance(option["default"], str):
raise Exception(
- "Key \"default\" in element with name=\"{}\" has an invalid type. "
- "Current type: {}".format(option["name"], type(option["default"]))
+ f"Key \"default\" in element with name=\"{option['name']}\" has an invalid type. "
+ f"Current type: {type(option['default'])}"
)
# Remove trailing whitespace on empty string
if option["default"]:
diff --git a/tests/api_connexion/endpoints/test_extra_link_endpoint.py b/tests/api_connexion/endpoints/test_extra_link_endpoint.py
index bec2ed8..4d53119 100644
--- a/tests/api_connexion/endpoints/test_extra_link_endpoint.py
+++ b/tests/api_connexion/endpoints/test_extra_link_endpoint.py
@@ -214,10 +214,9 @@ class TestGetExtraLinks:
operators = [BigQueryExecuteQueryOperator]
def get_link(self, operator, dttm):
- return "https://s3.amazonaws.com/airflow-logs/{dag_id}/{task_id}/{execution_date}".format(
- dag_id=operator.dag_id,
- task_id=operator.task_id,
- execution_date=quote_plus(dttm.isoformat()),
+ return (
+ f"https://s3.amazonaws.com/airflow-logs/{operator.dag_id}/"
+ f"{operator.task_id}/{quote_plus(dttm.isoformat())}"
)
class AirflowTestPlugin(AirflowPlugin):
diff --git a/tests/api_connexion/endpoints/test_log_endpoint.py b/tests/api_connexion/endpoints/test_log_endpoint.py
index 9b7972f..58ee683 100644
--- a/tests/api_connexion/endpoints/test_log_endpoint.py
+++ b/tests/api_connexion/endpoints/test_log_endpoint.py
@@ -121,8 +121,8 @@ class TestGetLog:
headers={'Accept': 'application/json'},
environ_overrides={'REMOTE_USER': "test"},
)
- expected_filename = "{}/{}/{}/{}/1.log".format(
- self.log_dir, self.DAG_ID, self.TASK_ID, self.default_time.replace(":", ".")
+ expected_filename = (
+ f"{self.log_dir}/{self.DAG_ID}/{self.TASK_ID}/{self.default_time.replace(':', '.')}/1.log"
)
assert (
response.json['content']
@@ -143,8 +143,8 @@ class TestGetLog:
headers={'Accept': 'text/plain'},
environ_overrides={'REMOTE_USER': "test"},
)
- expected_filename = "{}/{}/{}/{}/1.log".format(
- self.log_dir, self.DAG_ID, self.TASK_ID, self.default_time.replace(':', '.')
+ expected_filename = (
+ f"{self.log_dir}/{self.DAG_ID}/{self.TASK_ID}/{self.default_time.replace(':', '.')}/1.log"
)
assert 200 == response.status_code
assert (
@@ -169,8 +169,8 @@ class TestGetLog:
headers={'Accept': 'text/plain'},
environ_overrides={'REMOTE_USER': "test"},
)
- expected_filename = "{}/{}/{}/{}/1.log".format(
- self.log_dir, self.DAG_ID, self.TASK_ID, self.default_time.replace(':', '.')
+ expected_filename = (
+ f"{self.log_dir}/{self.DAG_ID}/{self.TASK_ID}/{self.default_time.replace(':', '.')}/1.log"
)
assert 200 == response.status_code
assert (
diff --git a/tests/conftest.py b/tests/conftest.py
index 94f915f..d357899 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -250,9 +250,9 @@ def skip_if_not_marked_with_integration(selected_integrations, item):
if integration_name in selected_integrations or "all" in selected_integrations:
return
pytest.skip(
- "The test is skipped because it does not have the right integration marker. "
- "Only tests marked with pytest.mark.integration(INTEGRATION) are run with INTEGRATION"
- " being one of {integration}. {item}".format(integration=selected_integrations, item=item)
+ f"The test is skipped because it does not have the right integration marker. "
+ f"Only tests marked with pytest.mark.integration(INTEGRATION) are run with INTEGRATION "
+ f"being one of {selected_integrations}. {item}"
)
@@ -262,9 +262,8 @@ def skip_if_not_marked_with_backend(selected_backend, item):
if selected_backend in backend_names:
return
pytest.skip(
- "The test is skipped because it does not have the right backend marker "
- "Only tests marked with pytest.mark.backend('{backend}') are run"
- ": {item}".format(backend=selected_backend, item=item)
+ f"The test is skipped because it does not have the right backend marker "
+ f"Only tests marked with pytest.mark.backend('{selected_backend}') are run: {item}"
)
@@ -274,36 +273,33 @@ def skip_if_not_marked_with_system(selected_systems, item):
if systems_name in selected_systems or "all" in selected_systems:
return
pytest.skip(
- "The test is skipped because it does not have the right system marker. "
- "Only tests marked with pytest.mark.system(SYSTEM) are run with SYSTEM"
- " being one of {systems}. {item}".format(systems=selected_systems, item=item)
+ f"The test is skipped because it does not have the right system marker. "
+ f"Only tests marked with pytest.mark.system(SYSTEM) are run with SYSTEM "
+ f"being one of {selected_systems}. {item}"
)
def skip_system_test(item):
for marker in item.iter_markers(name="system"):
pytest.skip(
- "The test is skipped because it has system marker. "
- "System tests are only run when --system flag "
- "with the right system ({system}) is passed to pytest. {item}".format(
- system=marker.args[0], item=item
- )
+ f"The test is skipped because it has system marker. System tests are only run when "
+ f"--system flag with the right system ({marker.args[0]}) is passed to pytest. {item}"
)
def skip_long_running_test(item):
for _ in item.iter_markers(name="long_running"):
pytest.skip(
- "The test is skipped because it has long_running marker. "
- "And --include-long-running flag is not passed to pytest. {item}".format(item=item)
+ f"The test is skipped because it has long_running marker. "
+ f"And --include-long-running flag is not passed to pytest. {item}"
)
def skip_quarantined_test(item):
for _ in item.iter_markers(name="quarantined"):
pytest.skip(
- "The test is skipped because it has quarantined marker. "
- "And --include-quarantined flag is passed to pytest. {item}".format(item=item)
+ f"The test is skipped because it has quarantined marker. "
+ f"And --include-quarantined flag is passed to pytest. {item}"
)
@@ -331,15 +327,9 @@ def skip_if_wrong_backend(marker, item):
environment_variable_value = os.environ.get(environment_variable_name)
if not environment_variable_value or environment_variable_value not in valid_backend_names:
pytest.skip(
- "The test requires one of {valid_backend_names} backend started and "
- "{name} environment variable to be set to 'true' (it is '{value}')."
- " It can be set by specifying backend at breeze startup"
- ": {item}".format(
- name=environment_variable_name,
- value=environment_variable_value,
- valid_backend_names=valid_backend_names,
- item=item,
- )
+ f"The test requires one of {valid_backend_names} backend started and "
+ f"{environment_variable_name} environment variable to be set to 'true' (it is "
+ f"'{environment_variable_value}'). It can be set by specifying backend at breeze startup: {item}"
)
diff --git a/tests/decorators/test_python.py b/tests/decorators/test_python.py
index 8782999..2355568 100644
--- a/tests/decorators/test_python.py
+++ b/tests/decorators/test_python.py
@@ -312,9 +312,7 @@ class TestAirflowTaskDecorator(TestPythonBase):
Call(
an_int=4,
a_date=date(2019, 1, 1),
- a_templated_string="dag {} ran on {}.".format(
- self.dag.dag_id, DEFAULT_DATE.date().isoformat()
- ),
+ a_templated_string=f"dag {self.dag.dag_id} ran on {DEFAULT_DATE.date().isoformat()}.",
),
)
diff --git a/tests/jobs/test_backfill_job.py b/tests/jobs/test_backfill_job.py
index fd76c8f..9ebce6c 100644
--- a/tests/jobs/test_backfill_job.py
+++ b/tests/jobs/test_backfill_job.py
@@ -847,9 +847,7 @@ class TestBackfillJob:
assert ti.state == State.SUCCESS
# raises backwards
- expected_msg = 'You cannot backfill backwards because one or more tasks depend_on_past: {}'.format(
- 'test_dop_task'
- )
+ expected_msg = 'You cannot backfill backwards because one or more tasks depend_on_past: test_dop_task'
with pytest.raises(AirflowException, match=expected_msg):
executor = MockExecutor()
job = BackfillJob(dag=dag, executor=executor, run_backwards=True, **kwargs)
diff --git a/tests/operators/test_python.py b/tests/operators/test_python.py
index 3adb035..b9ce518 100644
--- a/tests/operators/test_python.py
+++ b/tests/operators/test_python.py
@@ -212,9 +212,7 @@ class TestPythonOperator(TestPythonBase):
Call(
an_int=4,
a_date=date(2019, 1, 1),
- a_templated_string="dag {} ran on {}.".format(
- self.dag.dag_id, DEFAULT_DATE.date().isoformat()
- ),
+ a_templated_string=f"dag {self.dag.dag_id} ran on {DEFAULT_DATE.date().isoformat()}.",
),
)
diff --git a/tests/providers/amazon/aws/log/test_cloudwatch_task_handler.py b/tests/providers/amazon/aws/log/test_cloudwatch_task_handler.py
index db8ddcd..bd59121 100644
--- a/tests/providers/amazon/aws/log/test_cloudwatch_task_handler.py
+++ b/tests/providers/amazon/aws/log/test_cloudwatch_task_handler.py
@@ -65,9 +65,9 @@ class TestCloudwatchTaskHandler(unittest.TestCase):
self.ti.try_number = 1
self.ti.state = State.RUNNING
- self.remote_log_stream = '{}/{}/{}/{}.log'.format(
- dag_id, task_id, date.isoformat(), self.ti.try_number
- ).replace(':', '_')
+ self.remote_log_stream = f'{dag_id}/{task_id}/{date.isoformat()}/{self.ti.try_number}.log'.replace(
+ ':', '_'
+ )
moto.core.moto_api_backend.reset()
self.conn = boto3.client('logs', region_name=self.region_name)
@@ -171,8 +171,9 @@ class TestCloudwatchTaskHandler(unittest.TestCase):
)
msg_template = '*** Reading remote log from Cloudwatch log_group: {} log_stream: {}.\n{}\n'
- error_msg = 'Could not read remote logs from log_group: {} log_stream: {}.'.format(
- self.remote_log_group, self.remote_log_stream
+ error_msg = (
+ 'Could not read remote logs from log_group: '
+ f'{self.remote_log_group} log_stream: {self.remote_log_stream}.'
)
assert self.cloudwatch_task_handler.read(self.ti) == (
[[('', msg_template.format(self.remote_log_group, self.remote_log_stream, error_msg))]],
@@ -192,8 +193,9 @@ class TestCloudwatchTaskHandler(unittest.TestCase):
)
msg_template = '*** Reading remote log from Cloudwatch log_group: {} log_stream: {}.\n{}\n'
- error_msg = 'Could not read remote logs from log_group: {} log_stream: {}.'.format(
- self.remote_log_group, self.remote_log_stream
+ error_msg = (
+ f'Could not read remote logs from log_group: '
+ f'{self.remote_log_group} log_stream: {self.remote_log_stream}.'
)
assert self.cloudwatch_task_handler.read(self.ti) == (
[[('', msg_template.format(self.remote_log_group, self.remote_log_stream, error_msg))]],
diff --git a/tests/providers/apache/hive/hooks/test_hive.py b/tests/providers/apache/hive/hooks/test_hive.py
index 5242719..d0a8026 100644
--- a/tests/providers/apache/hive/hooks/test_hive.py
+++ b/tests/providers/apache/hive/hooks/test_hive.py
@@ -255,9 +255,7 @@ class TestHiveCliHook(unittest.TestCase):
"STORED AS textfile\n;".format(table=table, fields=fields)
)
- load_data = "LOAD DATA LOCAL INPATH '{filepath}' OVERWRITE INTO TABLE {table} ;\n".format(
- filepath=filepath, table=table
- )
+ load_data = f"LOAD DATA LOCAL INPATH '{filepath}' OVERWRITE INTO TABLE {table} ;\n"
calls = [mock.call(create_table), mock.call(load_data)]
mock_run_cli.assert_has_calls(calls, any_order=True)
diff --git a/tests/providers/apache/hive/operators/test_hive.py b/tests/providers/apache/hive/operators/test_hive.py
index 944d194..4b0f01f 100644
--- a/tests/providers/apache/hive/operators/test_hive.py
+++ b/tests/providers/apache/hive/operators/test_hive.py
@@ -92,9 +92,8 @@ class HiveOperatorTest(TestHiveEnvironment):
op.execute(fake_context)
assert (
- "Airflow HiveOperator task for {}.{}.{}.{}".format(
- fake_ti.hostname, self.dag.dag_id, op.task_id, fake_execution_date.isoformat()
- )
+ "Airflow HiveOperator task for "
+ f"{fake_ti.hostname}.{self.dag.dag_id}.{op.task_id}.{fake_execution_date.isoformat()}"
== mock_hook.mapred_job_name
)
diff --git a/tests/providers/apache/spark/hooks/test_spark_sql.py b/tests/providers/apache/spark/hooks/test_spark_sql.py
index 7fb44d6..039af32 100644
--- a/tests/providers/apache/spark/hooks/test_spark_sql.py
+++ b/tests/providers/apache/spark/hooks/test_spark_sql.py
@@ -206,8 +206,7 @@ class TestSparkSqlHook(unittest.TestCase):
hook.run_query(params)
# Then
- assert str(
- ctx.value
- ) == "Cannot execute '{}' on {} (additional parameters: '{}'). Process exit code: {}.".format(
- sql, master, params, status
+ assert str(ctx.value) == (
+ f"Cannot execute '{sql}' on {master} (additional parameters: '{params}'). "
+ f"Process exit code: {status}."
)
diff --git a/tests/providers/elasticsearch/log/test_es_task_handler.py b/tests/providers/elasticsearch/log/test_es_task_handler.py
index 604e5aa..05e53c2 100644
--- a/tests/providers/elasticsearch/log/test_es_task_handler.py
+++ b/tests/providers/elasticsearch/log/test_es_task_handler.py
@@ -132,8 +132,9 @@ class TestElasticsearchTaskHandler:
assert timezone.parse(metadatas[0]['last_log_timestamp']) > ts
def test_read_with_match_phrase_query(self, ti):
- similar_log_id = '{task_id}-{dag_id}-2016-01-01T00:00:00+00:00-1'.format(
- dag_id=TestElasticsearchTaskHandler.DAG_ID, task_id=TestElasticsearchTaskHandler.TASK_ID
+ similar_log_id = (
+ f'{TestElasticsearchTaskHandler.TASK_ID}-'
+ f'{TestElasticsearchTaskHandler.DAG_ID}-2016-01-01T00:00:00+00:00-1'
)
another_test_message = 'another message'
diff --git a/tests/providers/google/cloud/hooks/test_bigquery.py b/tests/providers/google/cloud/hooks/test_bigquery.py
index 66d0003..e0c2341 100644
--- a/tests/providers/google/cloud/hooks/test_bigquery.py
+++ b/tests/providers/google/cloud/hooks/test_bigquery.py
@@ -172,7 +172,7 @@ class TestBigQueryHookMethods(_BigQueryBaseTestClass):
with pytest.raises(
Exception,
match=(
- r"\['THIS IS NOT VALID'\] contains invalid schema update options."
+ r"\['THIS IS NOT VALID'\] contains invalid schema update options. "
r"Please only use one or more of the following options: "
r"\['ALLOW_FIELD_ADDITION', 'ALLOW_FIELD_RELAXATION'\]"
),
diff --git a/tests/providers/google/cloud/hooks/test_cloud_memorystore.py b/tests/providers/google/cloud/hooks/test_cloud_memorystore.py
index c2f9c33..98f3b10 100644
--- a/tests/providers/google/cloud/hooks/test_cloud_memorystore.py
+++ b/tests/providers/google/cloud/hooks/test_cloud_memorystore.py
@@ -50,11 +50,11 @@ TEST_UPDATE_MASK = {"paths": ["memory_size_gb"]} # type: Dict
TEST_UPDATE_MASK_MEMCACHED = {"displayName": "updated name"} # type: Dict
TEST_PARENT = "projects/test-project-id/locations/test-location" # type: str
TEST_NAME = "projects/test-project-id/locations/test-location/instances/test-instance-id" # type: str
-TEST_PARENT_DEFAULT_PROJECT_ID = "projects/{}/locations/test-location".format(
- GCP_PROJECT_ID_HOOK_UNIT_TEST
+TEST_PARENT_DEFAULT_PROJECT_ID = (
+ f"projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/locations/test-location"
) # type: str
-TEST_NAME_DEFAULT_PROJECT_ID = "projects/{}/locations/test-location/instances/test-instance-id".format(
- GCP_PROJECT_ID_HOOK_UNIT_TEST
+TEST_NAME_DEFAULT_PROJECT_ID = (
+ f"projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/locations/test-location/instances/test-instance-id"
) # type: str
diff --git a/tests/providers/google/cloud/hooks/test_cloud_storage_transfer_service.py b/tests/providers/google/cloud/hooks/test_cloud_storage_transfer_service.py
index f2fe788..cf8f2fa 100644
--- a/tests/providers/google/cloud/hooks/test_cloud_storage_transfer_service.py
+++ b/tests/providers/google/cloud/hooks/test_cloud_storage_transfer_service.py
@@ -500,9 +500,7 @@ class TestGCPTransferServiceHookWithPassedProjectId(unittest.TestCase):
with pytest.raises(
AirflowException,
- match="An unexpected operation status was encountered. Expected: {}".format(
- ", ".join(expected_statuses)
- ),
+ match=f"An unexpected operation status was encountered. Expected: {', '.join(expected_statuses)}",
):
CloudDataTransferServiceHook.operations_contain_expected_statuses(
operations, GcpTransferOperationStatus.IN_PROGRESS
diff --git a/tests/providers/google/cloud/hooks/test_dlp.py b/tests/providers/google/cloud/hooks/test_dlp.py
index dedf6da..d11db57 100644
--- a/tests/providers/google/cloud/hooks/test_dlp.py
+++ b/tests/providers/google/cloud/hooks/test_dlp.py
@@ -44,15 +44,9 @@ DLP_JOB_PATH = f"projects/{PROJECT_ID}/dlpJobs/{DLP_JOB_ID}"
TEMPLATE_ID = "template123"
STORED_INFO_TYPE_ID = "type123"
TRIGGER_ID = "trigger123"
-DEIDENTIFY_TEMPLATE_ORGANIZATION_PATH = "organizations/{}/deidentifyTemplates/{}".format(
- ORGANIZATION_ID, TEMPLATE_ID
-)
-INSPECT_TEMPLATE_ORGANIZATION_PATH = "organizations/{}/inspectTemplates/{}".format(
- ORGANIZATION_ID, TEMPLATE_ID
-)
-STORED_INFO_TYPE_ORGANIZATION_PATH = "organizations/{}/storedInfoTypes/{}".format(
- ORGANIZATION_ID, STORED_INFO_TYPE_ID
-)
+DEIDENTIFY_TEMPLATE_ORGANIZATION_PATH = f"organizations/{ORGANIZATION_ID}/deidentifyTemplates/{TEMPLATE_ID}"
+INSPECT_TEMPLATE_ORGANIZATION_PATH = f"organizations/{ORGANIZATION_ID}/inspectTemplates/{TEMPLATE_ID}"
+STORED_INFO_TYPE_ORGANIZATION_PATH = f"organizations/{ORGANIZATION_ID}/storedInfoTypes/{STORED_INFO_TYPE_ID}"
DEIDENTIFY_TEMPLATE_PROJECT_PATH = f"projects/{PROJECT_ID}/deidentifyTemplates/{TEMPLATE_ID}"
INSPECT_TEMPLATE_PROJECT_PATH = f"projects/{PROJECT_ID}/inspectTemplates/{TEMPLATE_ID}"
STORED_INFO_TYPE_PROJECT_PATH = f"projects/{PROJECT_ID}/storedInfoTypes/{STORED_INFO_TYPE_ID}"
diff --git a/tests/providers/google/cloud/hooks/test_functions.py b/tests/providers/google/cloud/hooks/test_functions.py
index 36f36a0..614edbd 100644
--- a/tests/providers/google/cloud/hooks/test_functions.py
+++ b/tests/providers/google/cloud/hooks/test_functions.py
@@ -278,9 +278,7 @@ class TestFunctionHookDefaultProjectId(unittest.TestCase):
function_id = "function1234"
input_data = {'key': 'value'}
- name = "projects/{project_id}/locations/{location}/functions/{function_id}".format(
- project_id=GCP_PROJECT_ID_HOOK_UNIT_TEST, location=GCF_LOCATION, function_id=function_id
- )
+ name = f"projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/locations/{GCF_LOCATION}/functions/{function_id}"
result = self.gcf_function_hook.call_function(
function_id=function_id,
diff --git a/tests/providers/google/cloud/hooks/test_kms.py b/tests/providers/google/cloud/hooks/test_kms.py
index b9f7c1c..3d7f0d4 100644
--- a/tests/providers/google/cloud/hooks/test_kms.py
+++ b/tests/providers/google/cloud/hooks/test_kms.py
@@ -37,8 +37,8 @@ TEST_PROJECT = "test-project"
TEST_LOCATION = "global"
TEST_KEY_RING = "test-key-ring"
TEST_KEY = "test-key"
-TEST_KEY_ID = "projects/{}/locations/{}/keyRings/{}/cryptoKeys/{}".format(
- TEST_PROJECT, TEST_LOCATION, TEST_KEY_RING, TEST_KEY
+TEST_KEY_ID = (
+ f"projects/{TEST_PROJECT}/locations/{TEST_LOCATION}/keyRings/{TEST_KEY_RING}/cryptoKeys/{TEST_KEY}"
)
RESPONSE = Response(PLAINTEXT, PLAINTEXT)
diff --git a/tests/providers/google/cloud/hooks/test_mlengine.py b/tests/providers/google/cloud/hooks/test_mlengine.py
index a50b05a..f06a897 100644
--- a/tests/providers/google/cloud/hooks/test_mlengine.py
+++ b/tests/providers/google/cloud/hooks/test_mlengine.py
@@ -911,9 +911,7 @@ class TestMLEngineHookWithDefaultProjectId(unittest.TestCase):
model_name = 'test-model'
version_name = 'test-version'
operation_path = f'projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/operations/test-operation'
- version_path = 'projects/{}/models/{}/versions/{}'.format(
- GCP_PROJECT_ID_HOOK_UNIT_TEST, model_name, version_name
- )
+ version_path = f'projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/models/{model_name}/versions/{version_name}'
operation_done = {'name': operation_path, 'done': True}
# fmt: off
(
@@ -999,12 +997,8 @@ class TestMLEngineHookWithDefaultProjectId(unittest.TestCase):
model_name = 'test-model'
version_name = 'test-version'
# fmt: off
- operation_path = 'projects/{}/operations/test-operation'.format(
- GCP_PROJECT_ID_HOOK_UNIT_TEST
- )
- version_path = 'projects/{}/models/{}/versions/{}'.format(
- GCP_PROJECT_ID_HOOK_UNIT_TEST, model_name, version_name
- )
+ operation_path = f'projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/operations/test-operation'
+ version_path = f'projects/{GCP_PROJECT_ID_HOOK_UNIT_TEST}/models/{model_name}/versions/{version_name}'
version = {'name': operation_path}
operation_not_done = {'name': operation_path, 'done': False}
operation_done = {'name': operation_path, 'done': True}
diff --git a/tests/providers/google/cloud/hooks/test_pubsub.py b/tests/providers/google/cloud/hooks/test_pubsub.py
index 7304364..027c2b2 100644
--- a/tests/providers/google/cloud/hooks/test_pubsub.py
+++ b/tests/providers/google/cloud/hooks/test_pubsub.py
@@ -201,9 +201,7 @@ class TestPubSubHook(unittest.TestCase):
subscription=TEST_SUBSCRIPTION,
subscription_project_id='a-different-project',
)
- expected_subscription = 'projects/{}/subscriptions/{}'.format(
- 'a-different-project', TEST_SUBSCRIPTION
- )
+ expected_subscription = f'projects/a-different-project/subscriptions/{TEST_SUBSCRIPTION}'
create_method.assert_called_once_with(
request=dict(
name=expected_subscription,
diff --git a/tests/providers/google/cloud/hooks/test_vision.py b/tests/providers/google/cloud/hooks/test_vision.py
index 09edb53..5b68250 100644
--- a/tests/providers/google/cloud/hooks/test_vision.py
+++ b/tests/providers/google/cloud/hooks/test_vision.py
@@ -40,14 +40,10 @@ LOC_ID_TEST = 'loc-id'
LOC_ID_TEST_2 = 'loc-id-2'
PRODUCTSET_ID_TEST = 'ps-id'
PRODUCTSET_ID_TEST_2 = 'ps-id-2'
-PRODUCTSET_NAME_TEST = 'projects/{}/locations/{}/productSets/{}'.format(
- PROJECT_ID_TEST, LOC_ID_TEST, PRODUCTSET_ID_TEST
-)
+PRODUCTSET_NAME_TEST = f'projects/{PROJECT_ID_TEST}/locations/{LOC_ID_TEST}/productSets/{PRODUCTSET_ID_TEST}'
PRODUCT_ID_TEST = 'p-id'
PRODUCT_ID_TEST_2 = 'p-id-2'
-PRODUCT_NAME_TEST = "projects/{}/locations/{}/products/{}".format(
- PROJECT_ID_TEST, LOC_ID_TEST, PRODUCT_ID_TEST
-)
+PRODUCT_NAME_TEST = f"projects/{PROJECT_ID_TEST}/locations/{LOC_ID_TEST}/products/{PRODUCT_ID_TEST}"
PRODUCT_NAME = f"projects/{PROJECT_ID_TEST}/locations/{LOC_ID_TEST}/products/{PRODUCT_ID_TEST}"
REFERENCE_IMAGE_ID_TEST = 'ri-id'
REFERENCE_IMAGE_GEN_ID_TEST = 'ri-id'
@@ -65,8 +61,9 @@ BATCH_ANNOTATE_IMAGE_REQUEST = [
'features': [{'type': enums.Feature.Type.LOGO_DETECTION}],
},
]
-REFERENCE_IMAGE_NAME_TEST = "projects/{}/locations/{}/products/{}/referenceImages/{}".format(
- PROJECT_ID_TEST, LOC_ID_TEST, PRODUCTSET_ID_TEST, REFERENCE_IMAGE_ID_TEST
+REFERENCE_IMAGE_NAME_TEST = (
+ f"projects/{PROJECT_ID_TEST}/locations/{LOC_ID_TEST}/products/"
+ f"{PRODUCTSET_ID_TEST}/referenceImages/{REFERENCE_IMAGE_ID_TEST}"
)
REFERENCE_IMAGE_TEST = ReferenceImage(name=REFERENCE_IMAGE_GEN_ID_TEST)
REFERENCE_IMAGE_WITHOUT_ID_NAME = ReferenceImage()
diff --git a/tests/providers/google/cloud/operators/test_cloud_build_system_helper.py b/tests/providers/google/cloud/operators/test_cloud_build_system_helper.py
index 5a1c849..f389bc5 100755
--- a/tests/providers/google/cloud/operators/test_cloud_build_system_helper.py
+++ b/tests/providers/google/cloud/operators/test_cloud_build_system_helper.py
@@ -75,9 +75,7 @@ class GCPCloudBuildTestHelper(LoggingCommandExecutor):
self.execute_cmd(["git", "add", "quickstart.sh"], cwd=tmp_dir)
self.execute_cmd(["git", "add", "Dockerfile"], cwd=tmp_dir)
self.execute_cmd(["git", "commit", "-m", "Initial commit"], cwd=tmp_dir)
- repo_url = "https://source.developers.google.com/p/{}/r/{}".format(
- GCP_PROJECT_ID, GCP_REPOSITORY_NAME
- )
+ repo_url = f"https://source.developers.google.com/p/{GCP_PROJECT_ID}/r/{GCP_REPOSITORY_NAME}"
self.execute_cmd(["git", "remote", "add", "origin", repo_url], cwd=tmp_dir)
self.execute_cmd(["git", "push", "--force", "origin", "master"], cwd=tmp_dir)
diff --git a/tests/providers/google/cloud/operators/test_cloud_sql.py b/tests/providers/google/cloud/operators/test_cloud_sql.py
index 8eda92d..47352d7 100644
--- a/tests/providers/google/cloud/operators/test_cloud_sql.py
+++ b/tests/providers/google/cloud/operators/test_cloud_sql.py
@@ -739,17 +739,9 @@ class TestCloudSqlQueryValidation(unittest.TestCase):
get_connection,
):
uri = (
- "gcpcloudsql://user:password@127.0.0.1:3200/testdb?"
- "database_type={database_type}&"
- "project_id={project_id}&location={location}&instance={instance_name}&"
- "use_proxy={use_proxy}&use_ssl={use_ssl}".format(
- database_type=database_type,
- project_id=project_id,
- location=location,
- instance_name=instance_name,
- use_proxy=use_proxy,
- use_ssl=use_ssl,
- )
+ f"gcpcloudsql://user:password@127.0.0.1:3200/testdb?"
+ f"database_type={database_type}&project_id={project_id}&location={location}"
+ f"&instance={instance_name}&use_proxy={use_proxy}&use_ssl={use_ssl}"
)
self._setup_connections(get_connection, uri)
with pytest.raises(AirflowException) as ctx:
@@ -763,9 +755,7 @@ class TestCloudSqlQueryValidation(unittest.TestCase):
uri = (
"gcpcloudsql://user:password@127.0.0.1:3200/testdb?database_type=postgres&"
"project_id=example-project&location=europe-west1&"
- "instance="
- "test_db_with_long_name_a_bit_above"
- "_the_limit_of_UNIX_socket_asdadadasadasd&"
+ "instance=test_db_with_long_name_a_bit_above_the_limit_of_UNIX_socket_asdadadasadasd&"
"use_proxy=True&sql_proxy_use_tcp=False"
)
self._setup_connections(get_connection, uri)
diff --git a/tests/providers/google/cloud/transfers/test_gcs_to_gcs.py b/tests/providers/google/cloud/transfers/test_gcs_to_gcs.py
index dc9039d..a23b50d 100644
--- a/tests/providers/google/cloud/transfers/test_gcs_to_gcs.py
+++ b/tests/providers/google/cloud/transfers/test_gcs_to_gcs.py
@@ -379,9 +379,7 @@ class TestGoogleCloudStorageToCloudStorageOperator(unittest.TestCase):
total_wildcards = operator.source_object.count(WILDCARD)
- error_msg = "Only one wildcard '[*]' is allowed in source_object parameter. Found {}".format(
- total_wildcards
- )
+ error_msg = f"Only one wildcard '[*]' is allowed in source_object parameter. Found {total_wildcards}"
with pytest.raises(AirflowException, match=error_msg):
operator.execute(None)
diff --git a/tests/providers/google/cloud/utils/gcp_authenticator.py b/tests/providers/google/cloud/utils/gcp_authenticator.py
index cae98d9..0714710 100644
--- a/tests/providers/google/cloud/utils/gcp_authenticator.py
+++ b/tests/providers/google/cloud/utils/gcp_authenticator.py
@@ -168,9 +168,7 @@ class GcpAuthenticator(LoggingCommandExecutor):
raise AirflowException("The gcp_key is not set!")
if not os.path.isfile(self.full_key_path):
raise AirflowException(
- "The key {} could not be found. Please copy it to the {} path.".format(
- self.gcp_key, self.full_key_path
- )
+ f"The key {self.gcp_key} could not be found. Please copy it to the {self.full_key_path} path."
)
def gcp_authenticate(self):
diff --git a/tests/providers/grpc/hooks/test_grpc.py b/tests/providers/grpc/hooks/test_grpc.py
index 3d9261e..0f1cd4d 100644
--- a/tests/providers/grpc/hooks/test_grpc.py
+++ b/tests/providers/grpc/hooks/test_grpc.py
@@ -27,13 +27,11 @@ from airflow.providers.grpc.hooks.grpc import GrpcHook
def get_airflow_connection(auth_type="NO_AUTH", credential_pem_file=None, scopes=None):
- extra = (
- '{{"extra__grpc__auth_type": "{auth_type}",'
- '"extra__grpc__credential_pem_file": "{credential_pem_file}",'
- '"extra__grpc__scopes": "{scopes}"}}'.format(
- auth_type=auth_type, credential_pem_file=credential_pem_file, scopes=scopes
- )
- )
+ extra = {
+ "extra__grpc__auth_type": f"{auth_type}",
+ "extra__grpc__credential_pem_file": f"{credential_pem_file}",
+ "extra__grpc__scopes": f"{scopes}",
+ }
return Connection(conn_id='grpc_default', conn_type='grpc', host='test:8080', extra=extra)
diff --git a/tests/providers/imap/hooks/test_imap.py b/tests/providers/imap/hooks/test_imap.py
index 56654ea..84d00d8 100644
--- a/tests/providers/imap/hooks/test_imap.py
+++ b/tests/providers/imap/hooks/test_imap.py
@@ -41,9 +41,11 @@ def _create_fake_imap(mock_imaplib, with_mail=False, attachment_name='test1.csv'
mock_conn.select.return_value = ('OK', [])
mock_conn.search.return_value = ('OK', [b'1'])
mail_string = (
- 'Content-Type: multipart/mixed; boundary=123\r\n--123\r\n'
- 'Content-Disposition: attachment; filename="{}";'
- 'Content-Transfer-Encoding: base64\r\nSWQsTmFtZQoxLEZlbGl4\r\n--123--'.format(attachment_name)
+ f'Content-Type: multipart/mixed; '
+ f'boundary=123\r\n--123\r\n'
+ f'Content-Disposition: attachment; '
+ f'filename="{attachment_name}";'
+ f'Content-Transfer-Encoding: base64\r\nSWQsTmFtZQoxLEZlbGl4\r\n--123--'
)
mock_conn.fetch.return_value = ('OK', [(b'', mail_string.encode('utf-8'))])
mock_conn.close.return_value = ('OK', [])
diff --git a/tests/security/test_kerberos.py b/tests/security/test_kerberos.py
index a60b4d6..d5bfece 100644
--- a/tests/security/test_kerberos.py
+++ b/tests/security/test_kerberos.py
@@ -98,9 +98,8 @@ class TestKerberos(unittest.TestCase):
with self.assertLogs(kerberos.log) as log:
assert (
- 'kinit: krb5_init_creds_set_keytab: Failed to find '
- 'airflow@LUPUS.GRIDDYNAMICS.NET in keytab FILE:{} '
- '(unknown enctype)'.format(self.args.keytab) in log.output
+ f'kinit: krb5_init_creds_set_keytab: Failed to find airflow@LUPUS.GRIDDYNAMICS.NET in '
+ f'keytab FILE:{self.args.keytab} (unknown enctype)' in log.output
)
assert ctx.value.code == 1
diff --git a/tests/sensors/test_python.py b/tests/sensors/test_python.py
index c4aecd4..40a0c28 100644
--- a/tests/sensors/test_python.py
+++ b/tests/sensors/test_python.py
@@ -132,8 +132,6 @@ class TestPythonSensor(TestPythonBase):
Call(
an_int=4,
a_date=date(2019, 1, 1),
- a_templated_string="dag {} ran on {}.".format(
- self.dag.dag_id, DEFAULT_DATE.date().isoformat()
- ),
+ a_templated_string=f"dag {self.dag.dag_id} ran on {DEFAULT_DATE.date().isoformat()}.",
),
)
diff --git a/tests/test_utils/system_tests_class.py b/tests/test_utils/system_tests_class.py
index a8d2bf5..000cfc9 100644
--- a/tests/test_utils/system_tests_class.py
+++ b/tests/test_utils/system_tests_class.py
@@ -140,13 +140,9 @@ class SystemTest(TestCase, LoggingMixin):
dag = dag_bag.get_dag(dag_id)
if dag is None:
raise AirflowException(
- "The Dag {dag_id} could not be found. It's either an import problem,"
- "wrong dag_id or DAG is not in provided dag_folder."
- "The content of the {dag_folder} folder is {content}".format(
- dag_id=dag_id,
- dag_folder=dag_folder,
- content=os.listdir(dag_folder),
- )
+ f"The Dag {dag_id} could not be found. It's either an import problem, wrong dag_id or DAG is "
+ "not in provided dag_folder.The content of "
+ f"the {dag_folder} folder is {os.listdir(dag_folder)}"
)
self.log.info("Attempting to run DAG: %s", dag_id)
diff --git a/tests/utils/test_helpers.py b/tests/utils/test_helpers.py
index 1bf94af..b0e2ca7 100644
--- a/tests/utils/test_helpers.py
+++ b/tests/utils/test_helpers.py
@@ -199,25 +199,25 @@ class TestHelpers:
("simple-key", None, None),
(
"group.simple_key",
- "The key (group.simple_key) has to be made of alphanumeric "
+ "The key ('group.simple_key') has to be made of alphanumeric "
"characters, dashes and underscores exclusively",
AirflowException,
),
(
"root.group-name.simple_key",
- "The key (root.group-name.simple_key) has to be made of alphanumeric "
+ "The key ('root.group-name.simple_key') has to be made of alphanumeric "
"characters, dashes and underscores exclusively",
AirflowException,
),
(
"key with space",
- "The key (key with space) has to be made of alphanumeric "
+ "The key ('key with space') has to be made of alphanumeric "
"characters, dashes and underscores exclusively",
AirflowException,
),
(
"key_with_!",
- "The key (key_with_!) has to be made of alphanumeric "
+ "The key ('key_with_!') has to be made of alphanumeric "
"characters, dashes and underscores exclusively",
AirflowException,
),
diff --git a/tests/utils/test_log_handlers.py b/tests/utils/test_log_handlers.py
index 4503dd8..d7cec84 100644
--- a/tests/utils/test_log_handlers.py
+++ b/tests/utils/test_log_handlers.py
@@ -232,8 +232,8 @@ def filename_rendering_ti(session, create_task_instance):
class TestFilenameRendering:
def test_python_formatting(self, filename_rendering_ti):
expected_filename = (
- 'dag_for_testing_filename_rendering/task_for_testing_filename_rendering/%s/42.log'
- % DEFAULT_DATE.isoformat()
+ f'dag_for_testing_filename_rendering/task_for_testing_filename_rendering/'
+ f'{DEFAULT_DATE.isoformat()}/42.log'
)
fth = FileTaskHandler('', '{dag_id}/{task_id}/{execution_date}/{try_number}.log')
@@ -242,8 +242,8 @@ class TestFilenameRendering:
def test_jinja_rendering(self, filename_rendering_ti):
expected_filename = (
- 'dag_for_testing_filename_rendering/task_for_testing_filename_rendering/%s/42.log'
- % DEFAULT_DATE.isoformat()
+ f'dag_for_testing_filename_rendering/task_for_testing_filename_rendering/'
+ f'{DEFAULT_DATE.isoformat()}/42.log'
)
fth = FileTaskHandler('', '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log')
diff --git a/tests/www/views/test_views_decorators.py b/tests/www/views/test_views_decorators.py
index 621d3a4..1a1c795 100644
--- a/tests/www/views/test_views_decorators.py
+++ b/tests/www/views/test_views_decorators.py
@@ -111,8 +111,9 @@ def _check_last_log(session, dag_id, event, execution_date):
def test_action_logging_get(session, admin_client):
- url = 'graph?dag_id=example_bash_operator&execution_date={}'.format(
- urllib.parse.quote_plus(str(EXAMPLE_DAG_DEFAULT_DATE))
+ url = (
+ f'graph?dag_id=example_bash_operator&'
+ f'execution_date={urllib.parse.quote_plus(str(EXAMPLE_DAG_DEFAULT_DATE))}'
)
resp = admin_client.get(url, follow_redirects=True)
check_content_in_response('runme_1', resp)
diff --git a/tests/www/views/test_views_extra_links.py b/tests/www/views/test_views_extra_links.py
index 8d7f1ac..72e17ee 100644
--- a/tests/www/views/test_views_extra_links.py
+++ b/tests/www/views/test_views_extra_links.py
@@ -131,12 +131,8 @@ def reset_task_instances():
def test_extra_links_works(dag, task_1, viewer_client):
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=foo-bar".format(
- ENDPOINT,
- dag.dag_id,
- task_1.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_1.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=foo-bar",
follow_redirects=True,
)
@@ -149,12 +145,8 @@ def test_extra_links_works(dag, task_1, viewer_client):
def test_global_extra_links_works(dag, task_1, viewer_client):
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=github".format(
- ENDPOINT,
- dag.dag_id,
- task_1.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_1.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=github",
follow_redirects=True,
)
@@ -192,12 +184,8 @@ def test_extra_link_in_gantt_view(dag, create_dag_run, viewer_client):
def test_operator_extra_link_override_global_extra_link(dag, task_1, viewer_client):
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=airflow".format(
- ENDPOINT,
- dag.dag_id,
- task_1.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_1.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=airflow",
follow_redirects=True,
)
@@ -210,12 +198,8 @@ def test_operator_extra_link_override_global_extra_link(dag, task_1, viewer_clie
def test_extra_links_error_raised(dag, task_1, viewer_client):
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=raise_error".format(
- ENDPOINT,
- dag.dag_id,
- task_1.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_1.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=raise_error",
follow_redirects=True,
)
@@ -228,12 +212,8 @@ def test_extra_links_error_raised(dag, task_1, viewer_client):
def test_extra_links_no_response(dag, task_1, viewer_client):
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=no_response".format(
- ENDPOINT,
- dag.dag_id,
- task_1.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_1.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=no_response",
follow_redirects=True,
)
@@ -253,12 +233,8 @@ def test_operator_extra_link_override_plugin(dag, task_2, viewer_client):
AirflowLink2 returns 'https://airflow.apache.org/1.10.5/' link
"""
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=airflow".format(
- ENDPOINT,
- dag.dag_id,
- task_2.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_2.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=airflow",
follow_redirects=True,
)
@@ -279,12 +255,8 @@ def test_operator_extra_link_multiple_operators(dag, task_2, task_3, viewer_clie
GoogleLink returns 'https://www.google.com'
"""
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=airflow".format(
- ENDPOINT,
- dag.dag_id,
- task_2.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_2.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=airflow",
follow_redirects=True,
)
@@ -295,12 +267,8 @@ def test_operator_extra_link_multiple_operators(dag, task_2, task_3, viewer_clie
assert json.loads(response_str) == {'url': 'https://airflow.apache.org/1.10.5/', 'error': None}
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=airflow".format(
- ENDPOINT,
- dag.dag_id,
- task_3.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_3.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=airflow",
follow_redirects=True,
)
@@ -312,12 +280,8 @@ def test_operator_extra_link_multiple_operators(dag, task_2, task_3, viewer_clie
# Also check that the other Operator Link defined for this operator exists
response = viewer_client.get(
- "{}?dag_id={}&task_id={}&execution_date={}&link_name=google".format(
- ENDPOINT,
- dag.dag_id,
- task_3.task_id,
- DEFAULT_DATE,
- ),
+ f"{ENDPOINT}?dag_id={dag.dag_id}&task_id={task_3.task_id}"
+ f"&execution_date={DEFAULT_DATE}&link_name=google",
follow_redirects=True,
)