Switch from Black to Ruff formatter (#35287)

This PR switches the formatter we use from Black to Ruff, now that Ruff's introduced a [formatter](https://docs.astral.sh/ruff/formatter/). The Ruff formatter is ~30x as fast as Black, and we already use Ruff for linting. This PR also upgrades the version of Ruff to the latest. Note that this doesn't swap the formatter used in inline Python code in the docs as I haven't seen an easy way of getting the Ruff formatter working with it. Because of this, a lot of the Black code hasn't been removed (i.e. the code to upgrade black, etc).
apache · Oct 31, 2023 · 63cc915 · 63cc915
1 parent 69bac3f
commit 63cc915
Show file tree

Hide file tree

Showing 62 changed files with 248 additions and 270 deletions.
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -155,13 +155,6 @@ repos:
           - --fuzzy-match-generates-todo
         files: >
           \.cfg$|\.conf$|\.ini$|\.ldif$|\.properties$|\.readthedocs$|\.service$|\.tf$|Dockerfile.*$
-  - repo: https://github.com/psf/black
-    rev: 23.10.0
-    hooks:
-      - id: black
-        name: Run black (Python formatter)
-        args: [--config=./pyproject.toml]
-        exclude: ^.*/.*_vendor/|^airflow/contrib/
   - repo: local
     hooks:
       - id: update-common-sql-api-stubs
@@ -181,14 +174,17 @@ repos:
         pass_filenames: false
         require_serial: true
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.0.292
+    rev: v0.1.3
     hooks:
       # Since ruff makes use of multiple cores we _purposefully_ don't run this in docker so it can use the
       # host CPU to it's fullest
       - id: ruff
-        name: ruff
-        args: [ --fix ]
+        name: ruff-lint
+        args: [--fix]
         exclude: ^.*/.*_vendor/|^tests/dags/test_imports.py
+      - id: ruff-format
+        name: ruff-format
+        exclude: ^.*/.*_vendor/|^tests/dags/test_imports.py|^airflow/contrib/
   - repo: https://github.com/asottile/blacken-docs
     rev: 1.16.0
     hooks:
@@ -200,7 +196,7 @@ repos:
           - --target-version=py38
           - --target-version=py39
           - --target-version=py310
-        alias: black
+        alias: blacken-docs
         additional_dependencies: [black==23.10.0]
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.5.0

diff --git a/STATIC_CODE_CHECKS.rst b/STATIC_CODE_CHECKS.rst
@@ -140,8 +140,6 @@ require Breeze Docker image to be built locally.
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | ID                                                        | Description                                                  | Image   |
 +===========================================================+==============================================================+=========+
-| black                                                     | Run black (Python formatter)                                 |         |
-+-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | blacken-docs                                              | Run black on Python code blocks in documentation files       |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | check-aiobotocore-optional                                | Check if aiobotocore is an optional dependency only          |         |
@@ -319,7 +317,9 @@ require Breeze Docker image to be built locally.
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | rst-backticks                                             | Check if RST files use double backticks for code             |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
-| ruff                                                      | ruff                                                         |         |
+| ruff                                                      | ruff-lint                                                    |         |
++-----------------------------------------------------------+--------------------------------------------------------------+---------+
+| ruff-format                                               | ruff-format                                                  |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | shellcheck                                                | Check Shell scripts syntax correctness                       |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+

diff --git a/airflow/api/common/delete_dag.py b/airflow/api/common/delete_dag.py
@@ -72,7 +72,7 @@ def delete_dag(dag_id: str, keep_records_in_log: bool = True, session: Session =
         )
     )
 
-    dags_to_delete = [dag_id for dag_id, in dags_to_delete_query]
+    dags_to_delete = [dag_id for (dag_id,) in dags_to_delete_query]
 
     # Scheduler removes DAGs without files from serialized_dag table every dag_dir_list_interval.
     # There may be a lag, so explicitly removes serialized DAG here.

diff --git a/airflow/lineage/__init__.py b/airflow/lineage/__init__.py
@@ -119,11 +119,9 @@ def wrapper(self, context, *args, **kwargs):
 
         if self.inlets and isinstance(self.inlets, list):
             # get task_ids that are specified as parameter and make sure they are upstream
-            task_ids = (
-                {o for o in self.inlets if isinstance(o, str)}
-                .union(op.task_id for op in self.inlets if isinstance(op, AbstractOperator))
-                .intersection(self.get_flat_relative_ids(upstream=True))
-            )
+            task_ids = {o for o in self.inlets if isinstance(o, str)}.union(
+                op.task_id for op in self.inlets if isinstance(op, AbstractOperator)
+            ).intersection(self.get_flat_relative_ids(upstream=True))
 
             # pick up unique direct upstream task_ids if AUTO is specified
             if AUTO.upper() in self.inlets or AUTO.lower() in self.inlets:

diff --git a/airflow/models/dag.py b/airflow/models/dag.py
@@ -3618,7 +3618,7 @@ def get_paused_dag_ids(dag_ids: list[str], session: Session = NEW_SESSION) -> se
             .where(DagModel.dag_id.in_(dag_ids))
         )
 
-        paused_dag_ids = {paused_dag_id for paused_dag_id, in paused_dag_ids}
+        paused_dag_ids = {paused_dag_id for (paused_dag_id,) in paused_dag_ids}
         return paused_dag_ids
 
     def get_default_view(self) -> str:

diff --git a/airflow/providers/apache/cassandra/__init__.py b/airflow/providers/apache/cassandra/__init__.py
@@ -39,5 +39,5 @@
     "2.5.0"
 ):
     raise RuntimeError(
-        f"The package `apache-airflow-providers-apache-cassandra:{__version__}` requires Apache Airflow 2.5.0+"  # NOQA: E501
+        f"The package `apache-airflow-providers-apache-cassandra:{__version__}` requires Apache Airflow 2.5.0+"
     )
diff --git a/airflow/providers/apache/kafka/operators/consume.py b/airflow/providers/apache/kafka/operators/consume.py
@@ -137,7 +137,9 @@ def execute(self, context) -> Any:
 
         if self.apply_function:
             apply_callable = partial(
-                self.apply_function, *self.apply_function_args, **self.apply_function_kwargs  # type: ignore
+                self.apply_function,  # type: ignore
+                *self.apply_function_args,
+                **self.apply_function_kwargs,
             )
 
         if self.apply_function_batch:

diff --git a/airflow/providers/cncf/kubernetes/utils/pod_manager.py b/airflow/providers/cncf/kubernetes/utils/pod_manager.py
@@ -719,7 +719,7 @@ def extract_xcom_json(self, pod: V1Pod) -> str:
         ) as resp:
             result = self._exec_pod_command(
                 resp,
-                f"if [ -s {PodDefaults.XCOM_MOUNT_PATH}/return.json ]; then cat {PodDefaults.XCOM_MOUNT_PATH}/return.json; else echo __airflow_xcom_result_empty__; fi",  # noqa
+                f"if [ -s {PodDefaults.XCOM_MOUNT_PATH}/return.json ]; then cat {PodDefaults.XCOM_MOUNT_PATH}/return.json; else echo __airflow_xcom_result_empty__; fi",
             )
             if result and result.rstrip() != "__airflow_xcom_result_empty__":
                 # Note: result string is parsed to check if its valid json.

diff --git a/airflow/providers/ftp/operators/ftp.py b/airflow/providers/ftp/operators/ftp.py
@@ -156,7 +156,7 @@ def get_openlineage_facets_on_start(self):
             local_host = socket.gethostbyname(local_host)
         except Exception as e:
             self.log.warning(
-                f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}",  # noqa: E501
+                f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}",
                 exc_info=True,
             )
 

diff --git a/airflow/providers/google/cloud/hooks/bigquery.py b/airflow/providers/google/cloud/hooks/bigquery.py
@@ -1725,7 +1725,7 @@ def run_load(
         # we check to make sure the passed source format is valid
         # if it's not, we raise a ValueError
         # Refer to this link for more details:
-        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.tableDefinitions.(key).sourceFormat # noqa
+        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.tableDefinitions.(key).sourceFormat
 
         if schema_fields is None and not autodetect:
             raise ValueError("You must either pass a schema or autodetect=True.")
@@ -2137,7 +2137,7 @@ def run_query(
         # BigQuery also allows you to define how you want a table's schema to change
         # as a side effect of a query job
         # for more details:
-        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.schemaUpdateOptions  # noqa
+        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.schemaUpdateOptions
 
         allowed_schema_update_options = ["ALLOW_FIELD_ADDITION", "ALLOW_FIELD_RELAXATION"]
 

diff --git a/airflow/providers/google/cloud/hooks/mlengine.py b/airflow/providers/google/cloud/hooks/mlengine.py
@@ -590,7 +590,9 @@ async def get_job_status(
         async with ClientSession() as session:
             try:
                 job = await self.get_job(
-                    project_id=project_id, job_id=job_id, session=session  #  type: ignore
+                    project_id=project_id,
+                    job_id=job_id,
+                    session=session,  #  type: ignore
                 )
                 job = await job.json(content_type=None)
                 self.log.info("Retrieving json_response: %s", job)

diff --git a/airflow/providers/google/cloud/hooks/secret_manager.py b/airflow/providers/google/cloud/hooks/secret_manager.py
@@ -81,5 +81,7 @@ def get_secret(
         :param project_id: Project id (if you want to override the project_id from credentials)
         """
         return self.get_conn().get_secret(
-            secret_id=secret_id, secret_version=secret_version, project_id=project_id  # type: ignore
+            secret_id=secret_id,
+            secret_version=secret_version,
+            project_id=project_id,  # type: ignore
         )
diff --git a/airflow/providers/google/cloud/operators/bigquery.py b/airflow/providers/google/cloud/operators/bigquery.py
@@ -2850,7 +2850,9 @@ def execute(self, context: Any):
         project_id = self.project_id or self.hook.project_id
         if project_id:
             job_id_path = convert_job_id(
-                job_id=self.job_id, project_id=project_id, location=self.location  # type: ignore[arg-type]
+                job_id=self.job_id,  # type: ignore[arg-type]
+                project_id=project_id,
+                location=self.location,
             )
             context["ti"].xcom_push(key="job_id_path", value=job_id_path)
         # Wait for the job to complete

diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py
@@ -152,7 +152,7 @@ class ClusterGenerator:
         ``projects/[PROJECT_STORING_KEYS]/locations/[LOCATION]/keyRings/[KEY_RING_NAME]/cryptoKeys/[KEY_NAME]`` # noqa
     :param enable_component_gateway: Provides access to the web interfaces of default and selected optional
         components on the cluster.
-    """  # noqa: E501
+    """
 
     def __init__(
         self,

diff --git a/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py b/airflow/providers/google/cloud/transfers/bigquery_to_mssql.py
@@ -62,10 +62,7 @@ def __init__(
     ) -> None:
         if mssql_table is not None:
             warnings.warn(
-                # fmt: off
-                "The `mssql_table` parameter has been deprecated. "
-                "Use `target_table_name` instead.",
-                # fmt: on
+                "The `mssql_table` parameter has been deprecated. Use `target_table_name` instead.",
                 AirflowProviderDeprecationWarning,
             )
 

diff --git a/airflow/providers/microsoft/azure/hooks/asb.py b/airflow/providers/microsoft/azure/hooks/asb.py
@@ -64,7 +64,7 @@ def get_ui_field_behaviour() -> dict[str, Any]:
                     "<Resource group>.servicebus.windows.net (for Azure AD authenticaltion)"
                 ),
                 "credential": "credential",
-                "schema": "Endpoint=sb://<Resource group>.servicebus.windows.net/;SharedAccessKeyName=<AccessKeyName>;SharedAccessKey=<SharedAccessKey>",  # noqa
+                "schema": "Endpoint=sb://<Resource group>.servicebus.windows.net/;SharedAccessKeyName=<AccessKeyName>;SharedAccessKey=<SharedAccessKey>",
             },
         }
 

diff --git a/airflow/providers/microsoft/azure/hooks/wasb.py b/airflow/providers/microsoft/azure/hooks/wasb.py
@@ -600,7 +600,9 @@ async def get_async_conn(self) -> AsyncBlobServiceClient:
                 tenant, app_id, app_secret, **client_secret_auth_config
             )
             self.blob_service_client = AsyncBlobServiceClient(
-                account_url=account_url, credential=token_credential, **extra  # type:ignore[arg-type]
+                account_url=account_url,
+                credential=token_credential,
+                **extra,  # type:ignore[arg-type]
             )
             return self.blob_service_client
 

diff --git a/airflow/providers/openlineage/plugins/listener.py b/airflow/providers/openlineage/plugins/listener.py
@@ -51,7 +51,10 @@ def __init__(self):
 
     @hookimpl
     def on_task_instance_running(
-        self, previous_state, task_instance: TaskInstance, session: Session  # This will always be QUEUED
+        self,
+        previous_state,
+        task_instance: TaskInstance,
+        session: Session,  # This will always be QUEUED
     ):
         if not hasattr(task_instance, "task"):
             self.log.warning(

diff --git a/airflow/providers/openlineage/utils/utils.py b/airflow/providers/openlineage/utils/utils.py
@@ -356,7 +356,8 @@ def _redact(self, item: Redactable, name: str | None, depth: int, max_depth: int
                 if attrs.has(type(item)):
                     # TODO: fixme when mypy gets compatible with new attrs
                     for dict_key, subval in attrs.asdict(
-                        item, recurse=False  # type: ignore[arg-type]
+                        item,  # type: ignore[arg-type]
+                        recurse=False,
                     ).items():
                         if _is_name_redactable(dict_key, item):
                             setattr(

diff --git a/airflow/providers/qubole/hooks/qubole.py b/airflow/providers/qubole/hooks/qubole.py
@@ -175,12 +175,16 @@ def execute(self, context: Context) -> None:
             time.sleep(Qubole.poll_interval)
             self.cmd = self.cls.find(self.cmd.id)  # type: ignore[attr-defined]
             self.log.info(
-                "Command Id: %s and Status: %s", self.cmd.id, self.cmd.status  # type: ignore[attr-defined]
+                "Command Id: %s and Status: %s",
+                self.cmd.id,
+                self.cmd.status,  # type: ignore[attr-defined]
             )
 
         if "fetch_logs" in self.kwargs and self.kwargs["fetch_logs"] is True:
             self.log.info(
-                "Logs for Command Id: %s \n%s", self.cmd.id, self.cmd.get_log()  # type: ignore[attr-defined]
+                "Logs for Command Id: %s \n%s",
+                self.cmd.id,
+                self.cmd.get_log(),  # type: ignore[attr-defined]
             )
 
         if self.cmd.status != "done":  # type: ignore[attr-defined]
@@ -236,9 +240,7 @@ def get_results(
             self.cmd = self.cls.find(cmd_id)
 
         include_headers_str = "true" if include_headers else "false"
-        self.cmd.get_results(
-            fp, inline, delim, fetch, arguments=[include_headers_str]
-        )  # type: ignore[attr-defined]
+        self.cmd.get_results(fp, inline, delim, fetch, arguments=[include_headers_str])  # type: ignore[attr-defined]
         fp.flush()
         fp.close()
         return fp.name

diff --git a/airflow/providers/qubole/operators/qubole.py b/airflow/providers/qubole/operators/qubole.py
@@ -54,8 +54,7 @@ def get_link(
         :return: url link
         """
         conn = BaseHook.get_connection(
-            getattr(operator, "qubole_conn_id", None)
-            or operator.kwargs["qubole_conn_id"]  # type: ignore[attr-defined]
+            getattr(operator, "qubole_conn_id", None) or operator.kwargs["qubole_conn_id"]  # type: ignore[attr-defined]
         )
         if conn and conn.host:
             host = re.sub(r"api$", "v2/analyze?command_id=", conn.host)

diff --git a/airflow/providers/sftp/operators/sftp.py b/airflow/providers/sftp/operators/sftp.py
@@ -210,7 +210,7 @@ def get_openlineage_facets_on_start(self):
             local_host = socket.gethostbyname(local_host)
         except Exception as e:
             self.log.warning(
-                f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}",  # noqa: E501
+                f"Failed to resolve local hostname. Using the hostname got by socket.gethostbyname() without resolution. {e}",
                 exc_info=True,
             )
 

diff --git a/airflow/providers/snowflake/operators/snowflake.py b/airflow/providers/snowflake/operators/snowflake.py
@@ -505,7 +505,9 @@ def execute(self, context: Context) -> None:
             deferrable=self.deferrable,
         )
         self.query_ids = self._hook.execute_query(
-            self.sql, statement_count=self.statement_count, bindings=self.bindings  # type: ignore[arg-type]
+            self.sql,  # type: ignore[arg-type]
+            statement_count=self.statement_count,
+            bindings=self.bindings,
         )
         self.log.info("List of query ids %s", self.query_ids)
 

diff --git a/airflow/providers/trino/hooks/trino.py b/airflow/providers/trino/hooks/trino.py
@@ -177,9 +177,7 @@ def get_first(
         except DatabaseError as e:
             raise TrinoException(e)
 
-    def get_pandas_df(
-        self, sql: str = "", parameters: Iterable | Mapping[str, Any] | None = None, **kwargs
-    ):  # type: ignore[override]
+    def get_pandas_df(self, sql: str = "", parameters: Iterable | Mapping[str, Any] | None = None, **kwargs):  # type: ignore[override]
         import pandas as pd
 
         cursor = self.get_cursor()

diff --git a/airflow/settings.py b/airflow/settings.py
@@ -293,9 +293,7 @@ def prepare_engine_args(disable_connection_pool=False, pool_class=None):
             default_args = default.copy()
             break
 
-    engine_args: dict = conf.getjson(
-        "database", "sql_alchemy_engine_args", fallback=default_args
-    )  # type: ignore
+    engine_args: dict = conf.getjson("database", "sql_alchemy_engine_args", fallback=default_args)  # type: ignore
 
     if pool_class:
         # Don't use separate settings for size etc, only those from sql_alchemy_engine_args

diff --git a/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow/ti_deps/deps/trigger_rule_dep.py
@@ -275,9 +275,12 @@ def _evaluate_setup_constraint(*, relevant_setups) -> Iterator[tuple[TIDepStatus
                         task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session, default=False
                     )
                     if not past_depends_met:
-                        yield self._failing_status(
-                            reason="Task should be skipped but the past depends are not met"
-                        ), changed
+                        yield (
+                            self._failing_status(
+                                reason="Task should be skipped but the past depends are not met"
+                            ),
+                            changed,
+                        )
                         return
                 changed = ti.set_state(new_state, session)
 
@@ -288,13 +291,16 @@ def _evaluate_setup_constraint(*, relevant_setups) -> Iterator[tuple[TIDepStatus
             if ti.map_index > -1:
                 non_successes -= removed
             if non_successes > 0:
-                yield self._failing_status(
-                    reason=(
-                        f"All setup tasks must complete successfully. Relevant setups: {relevant_setups}: "
-                        f"upstream_states={upstream_states}, "
-                        f"upstream_task_ids={task.upstream_task_ids}"
+                yield (
+                    self._failing_status(
+                        reason=(
+                            f"All setup tasks must complete successfully. Relevant setups: {relevant_setups}: "
+                            f"upstream_states={upstream_states}, "
+                            f"upstream_task_ids={task.upstream_task_ids}"
+                        ),
                     ),
-                ), changed
+                    changed,
+                )
 
         def _evaluate_direct_relatives() -> Iterator[TIDepStatus]:
             """Evaluate whether ``ti``'s trigger rule was met.

diff --git a/airflow/timetables/events.py b/airflow/timetables/events.py
@@ -83,7 +83,8 @@ def next_dagrun_info(
             next_event = self.event_dates[0]
         else:
             future_dates = itertools.dropwhile(
-                lambda when: when <= last_automated_data_interval.end, self.event_dates  # type: ignore
+                lambda when: when <= last_automated_data_interval.end,  # type: ignore
+                self.event_dates,
             )
             next_event = next(future_dates, None)  # type: ignore
             if next_event is None: