style: pre-commit.ci auto fixes [...]

taskcluster/fxci_config_taskgraph/util/integration.py

@@ -4,16 +4,17 @@
 
 import copy
 from functools import cache
-from typing import Any, Dict, List, Union
+from typing import Any
 
 import requests
 import taskcluster
-from taskgraph.util.taskcluster import logging, get_task_url, get_session
 from taskgraph.util.memoize import memoize
+from taskgraph.util.taskcluster import get_session, get_task_url, logging
 
 FIREFOXCI_ROOT_URL = "https://firefox-ci-tc.services.mozilla.com"
 STAGING_ROOT_URL = "https://stage.taskcluster.nonprod.cloudops.mozgcp.net"
 
+
 def _do_request(url, method=None, **kwargs):
     if method is None:
         method = "post" if kwargs else "get"
@@ -60,9 +61,7 @@ def _get_deps(task_ids, use_proxy):
     return upstream_tasks
 
 
-def get_ancestors(
-    task_ids: Union[List[str], str], use_proxy: bool = False
-) -> Dict[str, str]:
+def get_ancestors(task_ids: list[str] | str, use_proxy: bool = False) -> dict[str, str]:
     """Gets the ancestor tasks of the given task_ids as a dictionary of label -> taskid.
 
     Args:
@@ -72,7 +71,7 @@ def get_ancestors(
     Returns:
         dict: A dict whose keys are task labels and values are task ids.
     """
-    upstream_tasks: Dict[str, str] = {}
+    upstream_tasks: dict[str, str] = {}
     logging.info("A")
 
     if isinstance(task_ids, str):
@@ -87,14 +86,17 @@ def get_ancestors(
 
     return copy.deepcopy(upstream_tasks)
 
+
 @cache
 def get_taskcluster_client(service: str):
     options = {"rootUrl": FIREFOXCI_ROOT_URL}
     return getattr(taskcluster, service.capitalize())(options)
 
 
 @cache
-def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dict[str, Any]]:
+def find_tasks(
+    decision_index_path: str, include_deps: bool = False
+) -> list[dict[str, Any]]:
     """Find tasks targeted by the Decision task pointed to by `decision_index_path`."""
     queue = get_taskcluster_client("queue")
     index = get_taskcluster_client("index")
@@ -130,6 +132,7 @@ def find_tasks(decision_index_path: str, include_deps: bool = False) -> list[dic
             logging.info("including deps")
             # TODO: remove hack
             import os
+
             orig = os.environ["TASKCLUSTER_ROOT_URL"]
             os.environ["TASKCLUSTER_ROOT_URL"] = FIREFOXCI_ROOT_URL
             logging.info(f"fetching ancestors for {task['task_id']}")