Switch from Black to Ruff formatter

.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

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                       |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+

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.

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:

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:

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+"
     )

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:

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.

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,
             )
 

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"]
 

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)

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
         )

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

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,

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,
             )
 

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>",
             },
         }
 

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
 

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(

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(

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

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)

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,
             )
 

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)
 

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()

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

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.

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: