adopt ruff formatter (PrefectHQ#11959)

.pre-commit-config.yaml

@@ -1,12 +1,13 @@
 repos:
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.0.262"
+    rev: "v0.2.1"
     hooks:
+      - id: ruff-format
       - id: ruff
         language_version: python3
         args: [--fix, --exit-non-zero-on-fix, --show-fixes]
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.5
+    rev: v2.2.6
     hooks:
     - id: codespell
       exclude: package-lock.json|_vendor/.*

.ruff.toml

@@ -1,12 +1,12 @@
 src = ["src"]
 
 # Use Ruff for sorting imports
-extend-select = ["I"]
+lint.extend-select = ["I"]
 
 # Do not enforce line length; black does this for code and we do not care about comments / docs
-ignore = ["E501"]
+lint.ignore = ["E501"]
 
-[per-file-ignores]
+[lint.per-file-ignores]
 # Do not enforce usage and import order rules in init files
 "__init__.py" = ["E402", "F401", "I"]
 
@@ -22,5 +22,5 @@ ignore = ["E501"]
 # Do not enforce line length limits in migrations
 "src/prefect/server/database/migrations/**/*" = ["E501"]
 
-[isort]
+[lint.isort]
 known-third-party = ["prefect._vendor"]

src/prefect/_internal/concurrency/cancellation.py

@@ -230,7 +230,9 @@ def __repr__(self) -> str:
                 else (
                     "cancelled"
                     if self._cancelled
-                    else "running" if self._started else "pending"
+                    else "running"
+                    if self._started
+                    else "pending"
                 )
             ).upper()
             timeout = f", timeout={self._timeout:.2f}" if self._timeout else ""

src/prefect/_internal/concurrency/event_loop.py

@@ -29,7 +29,7 @@ def call_in_loop(
     __loop: asyncio.AbstractEventLoop,
     __fn: Callable[P, T],
     *args: P.args,
-    **kwargs: P.kwargs
+    **kwargs: P.kwargs,
 ) -> T:
     """
     Run a synchronous call in event loop's thread from another thread.
@@ -49,7 +49,7 @@ def call_soon_in_loop(
     __loop: asyncio.AbstractEventLoop,
     __fn: Callable[P, T],
     *args: P.args,
-    **kwargs: P.kwargs
+    **kwargs: P.kwargs,
 ) -> concurrent.futures.Future:
     """
     Run a synchronous call in an event loop's thread from another thread.

src/prefect/_internal/pydantic/annotations/pendulum.py

@@ -13,9 +13,9 @@
 
 
 class _PendulumDateTimeAnnotation:
-    _pendulum_type: t.Type[t.Union[pendulum.DateTime, pendulum.Date, pendulum.Time]] = (
-        pendulum.DateTime
-    )
+    _pendulum_type: t.Type[
+        t.Union[pendulum.DateTime, pendulum.Date, pendulum.Time]
+    ] = pendulum.DateTime
 
     _pendulum_types_to_schemas = {
         pendulum.DateTime: core_schema.datetime_schema(),
@@ -35,7 +35,7 @@ def validate_from_str(
             return pendulum.parse(value)
 
         def to_str(
-            value: t.Union[pendulum.DateTime, pendulum.Date, pendulum.Time]
+            value: t.Union[pendulum.DateTime, pendulum.Date, pendulum.Time],
         ) -> str:
             return value.isoformat()
 

src/prefect/_internal/pydantic/v2_schema.py

@@ -59,7 +59,7 @@ def process_v2_params(
     *,
     position: int,
     docstrings: t.Dict[str, str],
-    aliases: t.Dict
+    aliases: t.Dict,
 ) -> t.Tuple[str, t.Any, "pydantic.Field"]:
     """
     Generate a sanitized name, type, and pydantic.Field for a given parameter.
@@ -98,7 +98,7 @@ def create_v2_schema(
     name_: str,
     model_cfg: t.Optional[ConfigDict] = None,
     model_base: t.Optional[t.Type[V2BaseModel]] = None,
-    **model_fields
+    **model_fields,
 ):
     """
     Create a pydantic v2 model and craft a v1 compatible schema from it.

src/prefect/_vendor/fastapi/applications.py

@@ -150,7 +150,7 @@ def __init__(
         )
         self.exception_handlers: Dict[
             Any, Callable[[Request, Any], Union[Response, Awaitable[Response]]]
-        ] = ({} if exception_handlers is None else dict(exception_handlers))
+        ] = {} if exception_handlers is None else dict(exception_handlers)
         self.exception_handlers.setdefault(HTTPException, http_exception_handler)
         self.exception_handlers.setdefault(
             RequestValidationError, request_validation_exception_handler

src/prefect/_vendor/fastapi/dependencies/utils.py

@@ -803,7 +803,7 @@ async def request_body_to_args(
                 results: List[Union[bytes, str]] = []
 
                 async def process_fn(
-                    fn: Callable[[], Coroutine[Any, Any, Any]]
+                    fn: Callable[[], Coroutine[Any, Any, Any]],
                 ) -> None:
                     result = await fn()
                     results.append(result)  # noqa: B023

src/prefect/_vendor/fastapi/encoders.py

@@ -19,7 +19,7 @@
 
 
 def generate_encoders_by_class_tuples(
-    type_encoder_map: Dict[Any, Callable[[Any], Any]]
+    type_encoder_map: Dict[Any, Callable[[Any], Any]],
 ) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]:
     encoders_by_class_tuples: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(
         tuple

src/prefect/_vendor/fastapi/routing.py

@@ -408,9 +408,9 @@ def __init__(
             methods = ["GET"]
         self.methods: Set[str] = {method.upper() for method in methods}
         if isinstance(generate_unique_id_function, DefaultPlaceholder):
-            current_generate_unique_id: Callable[["APIRoute"], str] = (
-                generate_unique_id_function.value
-            )
+            current_generate_unique_id: Callable[
+                ["APIRoute"], str
+            ] = generate_unique_id_function.value
         else:
             current_generate_unique_id = generate_unique_id_function
         self.unique_id = self.operation_id or current_generate_unique_id(self)
@@ -433,9 +433,9 @@ def __init__(
             # would pass the validation and be returned as is.
             # By being a new field, no inheritance will be passed as is. A new model
             # will be always created.
-            self.secure_cloned_response_field: Optional[ModelField] = (
-                create_cloned_field(self.response_field)
-            )
+            self.secure_cloned_response_field: Optional[
+                ModelField
+            ] = create_cloned_field(self.response_field)
         else:
             self.response_field = None  # type: ignore
             self.secure_cloned_response_field = None

src/prefect/_vendor/fastapi/utils.py

@@ -39,9 +39,9 @@
     from .routing import APIRoute
 
 # Cache for `create_cloned_field`
-_CLONED_TYPES_CACHE: MutableMapping[Type[BaseModel], Type[BaseModel]] = (
-    WeakKeyDictionary()
-)
+_CLONED_TYPES_CACHE: MutableMapping[
+    Type[BaseModel], Type[BaseModel]
+] = WeakKeyDictionary()
 
 
 def is_body_allowed_for_status_code(status_code: Union[int, str, None]) -> bool:

src/prefect/_vendor/starlette/authentication.py

@@ -35,7 +35,7 @@ def requires(
     scopes_list = [scopes] if isinstance(scopes, str) else list(scopes)
 
     def decorator(
-        func: typing.Callable[_P, typing.Any]
+        func: typing.Callable[_P, typing.Any],
     ) -> typing.Callable[_P, typing.Any]:
         sig = inspect.signature(func)
         for idx, parameter in enumerate(sig.parameters.values()):

src/prefect/_vendor/starlette/routing.py

@@ -59,7 +59,9 @@ def iscoroutinefunction_or_partial(obj: typing.Any) -> bool:  # pragma: no cover
 
 
 def request_response(
-    func: typing.Callable[[Request], typing.Union[typing.Awaitable[Response], Response]]
+    func: typing.Callable[
+        [Request], typing.Union[typing.Awaitable[Response], Response]
+    ],
 ) -> ASGIApp:
     """
     Takes a function or coroutine `func(request) -> response`,
@@ -82,7 +84,7 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
 
 
 def websocket_session(
-    func: typing.Callable[[WebSocket], typing.Awaitable[None]]
+    func: typing.Callable[[WebSocket], typing.Awaitable[None]],
 ) -> ASGIApp:
     """
     Takes a coroutine `func(session)`, and returns an ASGI application.

src/prefect/blocks/core.py

@@ -257,9 +257,9 @@ def schema_extra(schema: Dict[str, Any], model: Type["Block"]):
                                     type_._to_block_schema_reference_dict(),
                                 ]
                             else:
-                                refs[field.name] = (
-                                    type_._to_block_schema_reference_dict()
-                                )
+                                refs[
+                                    field.name
+                                ] = type_._to_block_schema_reference_dict()
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)

src/prefect/blocks/notifications.py

@@ -24,14 +24,14 @@ class AbstractAppriseNotificationBlock(NotificationBlock, ABC):
     An abstract class for sending notifications using Apprise.
     """
 
-    notify_type: Literal["prefect_default", "info", "success", "warning", "failure"] = (
-        Field(
-            default=PREFECT_NOTIFY_TYPE_DEFAULT,
-            description=(
-                "The type of notification being performed; the prefect_default "
-                "is a plain notification that does not attach an image."
-            ),
-        )
+    notify_type: Literal[
+        "prefect_default", "info", "success", "warning", "failure"
+    ] = Field(
+        default=PREFECT_NOTIFY_TYPE_DEFAULT,
+        description=(
+            "The type of notification being performed; the prefect_default "
+            "is a plain notification that does not attach an image."
+        ),
     )
 
     def __init__(self, *args, **kwargs):

src/prefect/cli/_prompts.py

@@ -487,14 +487,14 @@ async def prompt_push_custom_docker_image(
                 import prefect_docker
 
             credentials_block = prefect_docker.DockerRegistryCredentials
-            push_step["credentials"] = (
-                "{{ prefect_docker.docker-registry-credentials.docker_registry_creds_name }}"
-            )
+            push_step[
+                "credentials"
+            ] = "{{ prefect_docker.docker-registry-credentials.docker_registry_creds_name }}"
         else:
             credentials_block = DockerRegistry
-            push_step["credentials"] = (
-                "{{ prefect.docker-registry.docker_registry_creds_name }}"
-            )
+            push_step[
+                "credentials"
+            ] = "{{ prefect.docker-registry.docker_registry_creds_name }}"
         docker_registry_creds_name = f"deployment-{slugify(deployment_config['name'])}-{slugify(deployment_config['work_pool']['name'])}-registry-creds"
         create_new_block = False
         try:

src/prefect/cli/dev.py

@@ -233,7 +233,8 @@ async def api(
         try:
             server_pid = await tg.start(start_command)
             async for _ in watchfiles.awatch(
-                prefect.__module_path__, stop_event=stop_event  # type: ignore
+                prefect.__module_path__,
+                stop_event=stop_event,  # type: ignore
             ):
                 # when any watched files change, restart the server
                 app.console.print("Restarting Prefect Server...")

src/prefect/cli/flow_run.py

@@ -268,7 +268,7 @@ async def logs(
 
 @flow_run_app.command()
 async def execute(
-    id: Optional[UUID] = typer.Argument(None, help="ID of the flow run to execute")
+    id: Optional[UUID] = typer.Argument(None, help="ID of the flow run to execute"),
 ):
     if id is None:
         environ_flow_id = os.environ.get("PREFECT__FLOW_RUN_ID")

src/prefect/cli/profile.py

@@ -225,7 +225,7 @@ def rename(name: str, new_name: str):
 def inspect(
     name: Optional[str] = typer.Argument(
         None, help="Name of profile to inspect; defaults to active profile."
-    )
+    ),
 ):
     """
     Display settings from a given profile; defaults to active.

src/prefect/context.py

@@ -137,9 +137,9 @@ class PrefectObjectRegistry(ContextModel):
     )
 
     # Failures will be a tuple of (exception, instance, args, kwargs)
-    _instance_init_failures: Dict[Type[T], List[Tuple[Exception, T, Tuple, Dict]]] = (
-        PrivateAttr(default_factory=lambda: defaultdict(list))
-    )
+    _instance_init_failures: Dict[
+        Type[T], List[Tuple[Exception, T, Tuple, Dict]]
+    ] = PrivateAttr(default_factory=lambda: defaultdict(list))
 
     block_code_execution: bool = False
     capture_failures: bool = False

src/prefect/engine.py

@@ -2500,8 +2500,7 @@ def resolve_input(expr, context):
             #       incorrectly evaluate to false — to resolve this, we must track all
             #       annotations wrapping the current expression but this is not yet
             #       implemented.
-            isinstance(context.get("annotation"), allow_failure)
-            and state.is_failed()
+            isinstance(context.get("annotation"), allow_failure) and state.is_failed()
         ):
             raise UpstreamTaskError(
                 f"Upstream task run '{state.state_details.task_run_id}' did not reach a"

src/prefect/filesystems.py

@@ -709,23 +709,23 @@ def basepath(self) -> str:
     def filesystem(self) -> RemoteFileSystem:
         settings = {}
         if self.azure_storage_connection_string:
-            settings["connection_string"] = (
-                self.azure_storage_connection_string.get_secret_value()
-            )
+            settings[
+                "connection_string"
+            ] = self.azure_storage_connection_string.get_secret_value()
         if self.azure_storage_account_name:
-            settings["account_name"] = (
-                self.azure_storage_account_name.get_secret_value()
-            )
+            settings[
+                "account_name"
+            ] = self.azure_storage_account_name.get_secret_value()
         if self.azure_storage_account_key:
             settings["account_key"] = self.azure_storage_account_key.get_secret_value()
         if self.azure_storage_tenant_id:
             settings["tenant_id"] = self.azure_storage_tenant_id.get_secret_value()
         if self.azure_storage_client_id:
             settings["client_id"] = self.azure_storage_client_id.get_secret_value()
         if self.azure_storage_client_secret:
-            settings["client_secret"] = (
-                self.azure_storage_client_secret.get_secret_value()
-            )
+            settings[
+                "client_secret"
+            ] = self.azure_storage_client_secret.get_secret_value()
         settings["anon"] = self.azure_storage_anon
         self._remote_file_system = RemoteFileSystem(
             basepath=self.basepath, settings=settings

src/prefect/futures.py

@@ -360,7 +360,7 @@ def replace_futures_with_results(expr, context):
 
 
 async def resolve_futures_to_states(
-    expr: Union[PrefectFuture[R, Any], Any]
+    expr: Union[PrefectFuture[R, Any], Any],
 ) -> Union[State[R], Any]:
     """
     Given a Python built-in collection, recursively find `PrefectFutures` and build a

src/prefect/infrastructure/container.py

@@ -398,9 +398,9 @@ async def generate_work_pool_base_job_template(self):
             return await super().generate_work_pool_base_job_template()
         for key, value in self.dict(exclude_unset=True, exclude_defaults=True).items():
             if key == "command":
-                base_job_template["variables"]["properties"]["command"]["default"] = (
-                    shlex.join(value)
-                )
+                base_job_template["variables"]["properties"]["command"][
+                    "default"
+                ] = shlex.join(value)
             elif key == "image_registry":
                 self.logger.warning(
                     "Image registry blocks are not supported by Docker"

src/prefect/infrastructure/kubernetes.py

@@ -376,9 +376,9 @@ async def generate_work_pool_base_job_template(self):
         ), "Failed to retrieve default base job template."
         for key, value in self.dict(exclude_unset=True, exclude_defaults=True).items():
             if key == "command":
-                base_job_template["variables"]["properties"]["command"]["default"] = (
-                    shlex.join(value)
-                )
+                base_job_template["variables"]["properties"]["command"][
+                    "default"
+                ] = shlex.join(value)
             elif key in [
                 "type",
                 "block_type_slug",
@@ -892,9 +892,7 @@ def _slugify_label_key(self, key: str) -> str:
                     prefix,
                     max_length=253,
                     regex_pattern=r"[^a-zA-Z0-9-\.]+",
-                ).strip(
-                    "_-."
-                )  # Must start or end with alphanumeric characters
+                ).strip("_-.")  # Must start or end with alphanumeric characters
                 or prefix
             )