🎨 Auto format

fastapi/_compat/shared.py

@@ -1,4 +1,3 @@
-import sys
 import types
 import typing
 import warnings
@@ -8,15 +7,17 @@ from dataclasses import is_dataclass
 from typing import (
     Annotated,
     Any,
+    TypeGuard,
     TypeVar,
     Union,
+    get_args,
+    get_origin,
 )
 
 from fastapi.types import UnionType
 from pydantic import BaseModel
 from pydantic.version import VERSION as PYDANTIC_VERSION
 from starlette.datastructures import UploadFile
-from typing_extensions import TypeGuard, get_args, get_origin
 
 _T = TypeVar("_T")
 
@@ -44,7 +45,7 @@ sequence_types: tuple[type[Any], ...] = tuple(sequence_annotation_to_type.keys()
 
 # Copy of Pydantic: pydantic/_internal/_utils.py with added TypeGuard
 def lenient_issubclass(
-    cls: Any, class_or_tuple: Union[type[_T], tuple[type[_T], ...], None]
+    cls: Any, class_or_tuple: type[_T] | tuple[type[_T], ...] | None
 ) -> TypeGuard[type[_T]]:
     try:
         return isinstance(cls, type) and issubclass(cls, class_or_tuple)  # type: ignore[arg-type]
@@ -54,13 +55,13 @@ def lenient_issubclass(
         raise  # pragma: no cover
 
 
-def _annotation_is_sequence(annotation: Union[type[Any], None]) -> bool:
+def _annotation_is_sequence(annotation: type[Any] | None) -> bool:
     if lenient_issubclass(annotation, (str, bytes)):
         return False
     return lenient_issubclass(annotation, sequence_types)
 
 
-def field_annotation_is_sequence(annotation: Union[type[Any], None]) -> bool:
+def field_annotation_is_sequence(annotation: type[Any] | None) -> bool:
     origin = get_origin(annotation)
     if origin is Union or origin is UnionType:
         for arg in get_args(annotation):
@@ -76,7 +77,7 @@ def value_is_sequence(value: Any) -> bool:
     return isinstance(value, sequence_types) and not isinstance(value, (str, bytes))
 
 
-def _annotation_is_complex(annotation: Union[type[Any], None]) -> bool:
+def _annotation_is_complex(annotation: type[Any] | None) -> bool:
     return (
         lenient_issubclass(annotation, (BaseModel, Mapping, UploadFile))
         or _annotation_is_sequence(annotation)
@@ -84,7 +85,7 @@ def _annotation_is_complex(annotation: Union[type[Any], None]) -> bool:
     )
 
 
-def field_annotation_is_complex(annotation: Union[type[Any], None]) -> bool:
+def field_annotation_is_complex(annotation: type[Any] | None) -> bool:
     origin = get_origin(annotation)
     if origin is Union or origin is UnionType:
         return any(field_annotation_is_complex(arg) for arg in get_args(annotation))
@@ -105,7 +106,7 @@ def field_annotation_is_scalar(annotation: Any) -> bool:
     return annotation is Ellipsis or not field_annotation_is_complex(annotation)
 
 
-def field_annotation_is_scalar_sequence(annotation: Union[type[Any], None]) -> bool:
+def field_annotation_is_scalar_sequence(annotation: type[Any] | None) -> bool:
     origin = get_origin(annotation)
     if origin is Union or origin is UnionType:
         at_least_one_scalar_sequence = False

fastapi/dependencies/utils.py

@@ -1,18 +1,19 @@
 import dataclasses
 import inspect
 import sys
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from contextlib import AsyncExitStack, contextmanager
 from copy import copy, deepcopy
 from dataclasses import dataclass
 from typing import (
     Annotated,
     Any,
-    Callable,
     ForwardRef,
-    Optional,
+    Literal,
     Union,
     cast,
+    get_args,
+    get_origin,
 )
 
 from fastapi import params
@@ -63,7 +64,6 @@ from starlette.datastructures import (
 from starlette.requests import HTTPConnection, Request
 from starlette.responses import Response
 from starlette.websockets import WebSocket
-from typing_extensions import Literal, get_args, get_origin
 from typing_inspection.typing_objects import is_typealiastype
 
 multipart_not_installed_error = (
@@ -127,8 +127,8 @@ def get_flat_dependant(
     dependant: Dependant,
     *,
     skip_repeats: bool = False,
-    visited: Optional[list[DependencyCacheKey]] = None,
-    parent_oauth_scopes: Optional[list[str]] = None,
+    visited: list[DependencyCacheKey] | None = None,
+    parent_oauth_scopes: list[str] | None = None,
 ) -> Dependant:
     if visited is None:
         visited = []
@@ -255,11 +255,11 @@ def get_dependant(
     *,
     path: str,
     call: Callable[..., Any],
-    name: Optional[str] = None,
-    own_oauth_scopes: Optional[list[str]] = None,
-    parent_oauth_scopes: Optional[list[str]] = None,
+    name: str | None = None,
+    own_oauth_scopes: list[str] | None = None,
+    parent_oauth_scopes: list[str] | None = None,
     use_cache: bool = True,
-    scope: Union[Literal["function", "request"], None] = None,
+    scope: Literal["function", "request"] | None = None,
 ) -> Dependant:
     dependant = Dependant(
         call=call,
@@ -328,7 +328,7 @@ def get_dependant(
 
 def add_non_field_param_to_dependency(
     *, param_name: str, type_annotation: Any, dependant: Dependant
-) -> Optional[bool]:
+) -> bool | None:
     if lenient_issubclass(type_annotation, Request):
         dependant.request_param_name = param_name
         return True
@@ -353,8 +353,8 @@ def add_non_field_param_to_dependency(
 @dataclass
 class ParamDetails:
     type_annotation: Any
-    depends: Optional[params.Depends]
-    field: Optional[ModelField]
+    depends: params.Depends | None
+    field: ModelField | None
 
 
 def analyze_param(
@@ -396,7 +396,7 @@ def analyze_param(
             )
         ]
         if fastapi_specific_annotations:
-            fastapi_annotation: Union[FieldInfo, params.Depends, None] = (
+            fastapi_annotation: FieldInfo | params.Depends | None = (
                 fastapi_specific_annotations[-1]
             )
         else:
@@ -557,20 +557,20 @@ async def _solve_generator(
 class SolvedDependency:
     values: dict[str, Any]
     errors: list[Any]
-    background_tasks: Optional[StarletteBackgroundTasks]
+    background_tasks: StarletteBackgroundTasks | None
     response: Response
     dependency_cache: dict[DependencyCacheKey, Any]
 
 
 async def solve_dependencies(
     *,
-    request: Union[Request, WebSocket],
+    request: Request | WebSocket,
     dependant: Dependant,
-    body: Optional[Union[dict[str, Any], FormData]] = None,
-    background_tasks: Optional[StarletteBackgroundTasks] = None,
-    response: Optional[Response] = None,
-    dependency_overrides_provider: Optional[Any] = None,
-    dependency_cache: Optional[dict[DependencyCacheKey, Any]] = None,
+    body: dict[str, Any] | FormData | None = None,
+    background_tasks: StarletteBackgroundTasks | None = None,
+    response: Response | None = None,
+    dependency_overrides_provider: Any | None = None,
+    dependency_cache: dict[DependencyCacheKey, Any] | None = None,
     # TODO: remove this parameter later, no longer used, not removing it yet as some
     # people might be monkey patching this function (although that's not supported)
     async_exit_stack: AsyncExitStack,
@@ -718,7 +718,7 @@ def _is_json_field(field: ModelField) -> bool:
 
 
 def _get_multidict_value(
-    field: ModelField, values: Mapping[str, Any], alias: Union[str, None] = None
+    field: ModelField, values: Mapping[str, Any], alias: str | None = None
 ) -> Any:
     alias = alias or get_validation_alias(field)
     if (
@@ -750,7 +750,7 @@ def _get_multidict_value(
 
 def request_params_to_args(
     fields: Sequence[ModelField],
-    received_params: Union[Mapping[str, Any], QueryParams, Headers],
+    received_params: Mapping[str, Any] | QueryParams | Headers,
 ) -> tuple[dict[str, Any], list[Any]]:
     values: dict[str, Any] = {}
     errors: list[dict[str, Any]] = []
@@ -898,7 +898,7 @@ async def _extract_form_body(
         ):
             # For types
             assert isinstance(value, sequence_types)
-            results: list[Union[bytes, str]] = []
+            results: list[bytes | str] = []
             for sub_value in value:
                 results.append(await sub_value.read())
             value = serialize_sequence_value(field=field, value=results)
@@ -917,7 +917,7 @@ async def _extract_form_body(
 
 async def request_body_to_args(
     body_fields: list[ModelField],
-    received_body: Optional[Union[dict[str, Any], FormData]],
+    received_body: dict[str, Any] | FormData | None,
     embed_body_fields: bool,
 ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
     values: dict[str, Any] = {}
@@ -947,7 +947,7 @@ async def request_body_to_args(
         return {first_field.name: v_}, errors_
     for field in body_fields:
         loc = ("body", get_validation_alias(field))
-        value: Optional[Any] = None
+        value: Any | None = None
         if body_to_process is not None:
             try:
                 value = body_to_process.get(get_validation_alias(field))
@@ -967,7 +967,7 @@ async def request_body_to_args(
 
 def get_body_field(
     *, flat_dependant: Dependant, name: str, embed_body_fields: bool
-) -> Optional[ModelField]:
+) -> ModelField | None:
     """
     Get a ModelField representing the request body for a path operation, combining
     all body parameters into a single field if necessary.