🎨 [pre-commit.ci] Auto format from pre-commit.com hooks

fastapi/_compat/__init__.py

@@ -104,19 +104,24 @@ CoreSchema = Any
 GetJsonSchemaHandler = Any
 JsonSchemaValue = dict[str, Any]
 
+
 def _normalize_errors(errors: Any) -> Any:
     from importlib import import_module
+
     v1 = import_module("fastapi._compat.v1")  # proxy lazy
     return v1._normalize_errors(errors)
 
+
 # Make v1 available as an attribute
 def __getattr__(name: str) -> Any:
     if name == "v1":
         # Import directly to avoid recursion
         import importlib
+
         return importlib.import_module("fastapi._compat.v1")
     raise AttributeError(f"module 'fastapi._compat' has no attribute '{name}'")
 
+
 # Explicitly export all compatibility symbols
 __all__ = [
     "PYDANTIC_V2",

fastapi/_compat/_v1_params.py

@@ -1,4 +1,3 @@
-
 """
 Internal v1-params shim. Will be removed when v1 is dropped.
 
@@ -13,28 +12,47 @@ from typing import Any
 
 _SENTINEL = object()
 
+
 def _v1() -> Any:
     """Lazy import of v1 module to avoid warnings."""
     from . import v1  # lazy proxy; only warns/errors if actually using v1
+
     return v1
 
+
 class _BaseParam:
     """Minimal wrapper that delegates to v1.FieldInfo without importing v1 at import-time."""
+
     def __init__(self, default: Any = _SENTINEL, **kwargs: Any):
         v1 = _v1()
         if default is _SENTINEL:
-            default = getattr(v1, 'Undefined', None)
+            default = getattr(v1, "Undefined", None)
         self._fi = v1.FieldInfo(default=default, **kwargs)
 
     def __getattr__(self, name: str) -> Any:
         return getattr(self._fi, name)
 
+
 # Types used in isinstance() checks in core
 class Param(_BaseParam): ...
+
+
 class Body(_BaseParam): ...
+
+
 class Form(Body): ...
+
+
 class File(Form): ...
+
+
 class Path(Param): ...
+
+
 class Query(Param): ...
+
+
 class Header(Param): ...
+
+
 class Cookie(Param): ...

fastapi/_compat/_v1_params.pyi

@@ -1,29 +1,44 @@
-
 from typing import Any
 
 class Param:
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
     in_: Any
 
 class Body(Param):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
 
 class Form(Body):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
 
 class File(Form):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
 
 class Path(Param):
-    def __init__(self, annotation: Any = ..., default: Any = ..., alias: str = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., alias: str = ..., **kwargs: Any
+    ) -> None: ...
     alias: str
     default: Any
 
 class Query(Param):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
 
 class Header(Param):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...
 
 class Cookie(Param):
-    def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ...
+    def __init__(
+        self, annotation: Any = ..., default: Any = ..., **kwargs: Any
+    ) -> None: ...

fastapi/_compat/lazy_import.py

@@ -8,7 +8,8 @@ from __future__ import annotations
 import sys
 from typing import Any, Optional, TypeVar
 
-T = TypeVar('T')
+T = TypeVar("T")
+
 
 def get_v1_if_loaded() -> Optional[Any]:
     """
@@ -17,20 +18,25 @@ def get_v1_if_loaded() -> Optional[Any]:
     """
     if "pydantic.v1" in sys.modules:
         import pydantic.v1
+
         return pydantic.v1
     return None
 
+
 def with_v1_guard(func: Any) -> Any:
     """
     Decorator that only executes function if pydantic.v1 is loaded.
     """
+
     def wrapper(*args: Any, **kwargs: Any) -> Any:
         v1 = get_v1_if_loaded()
         if v1 is not None:
             return func(v1, *args, **kwargs)
         return None
+
     return wrapper
 
+
 def v1_isinstance(obj: Any, v1_class: str) -> bool:
     """
     Check isinstance with v1 class only if v1 is loaded.
@@ -42,6 +48,7 @@ def v1_isinstance(obj: Any, v1_class: str) -> bool:
             return isinstance(obj, cls)
     return False
 
+
 def v1_lenient_issubclass(cls: Any, v1_class: str) -> bool:
     """
     Check lenient_issubclass with v1 class only if v1 is loaded.
@@ -51,9 +58,11 @@ def v1_lenient_issubclass(cls: Any, v1_class: str) -> bool:
         v1_cls = getattr(v1, v1_class, None)
         if v1_cls is not None:
             from .shared import lenient_issubclass
+
             return lenient_issubclass(cls, v1_cls)
     return False
 
+
 def v1_call_method(method_name: str, *args: Any, **kwargs: Any) -> Any:
     """
     Call a v1 method only if v1 is loaded.
@@ -65,6 +74,7 @@ def v1_call_method(method_name: str, *args: Any, **kwargs: Any) -> Any:
             return method(*args, **kwargs)
     return None
 
+
 def v1_get_attr(attr_name: str) -> Any:
     """
     Get v1 attribute only if v1 is loaded.

fastapi/_compat/main.py

@@ -34,6 +34,7 @@ def get_cached_model_fields(model: Type[BaseModel]) -> Any:
         return v1.get_model_fields(model)
 
     from . import v2
+
     return v2.get_model_fields(model)
 
 
@@ -42,6 +43,7 @@ def _is_undefined(value: object) -> bool:
         return True
     elif PYDANTIC_V2:
         from pydantic_core import PydanticUndefined
+
         return value is PydanticUndefined
     else:
         return False
@@ -56,6 +58,7 @@ def _get_model_config(model: BaseModel) -> Any:
 
     if PYDANTIC_V2:
         from . import v2
+
         return v2._get_model_config(model)
 
     return getattr(model, "__config__", None)
@@ -72,6 +75,7 @@ def _model_dump(
 
     if PYDANTIC_V2:
         from . import v2
+
         return v2._model_dump(model, mode=mode, **kwargs)
 
     return model.dict(**kwargs)
@@ -96,12 +100,11 @@ def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
         return v1.copy_field_info(field_info=field_info, annotation=annotation)
 
     from . import v2
+
     return v2.copy_field_info(field_info=field_info, annotation=annotation)
 
 
-def create_body_model(
+def create_body_model(*, fields: List[ModelField], model_name: str) -> Any:
-    *, fields: List[ModelField], model_name: str
-) -> Any:
     if fields and v1_isinstance(fields[0], "ModelField"):
         v1 = get_v1_if_loaded()
         if v1 is None:
@@ -109,6 +112,7 @@ def create_body_model(
         return v1.create_body_model(fields=fields, model_name=model_name)
 
     from . import v2
+
     return v2.create_body_model(fields=fields, model_name=model_name)
 
 
@@ -124,6 +128,7 @@ def get_annotation_from_field_info(
         )
     else:
         from . import v2
+
         return v2.get_annotation_from_field_info(
             annotation=annotation, field_info=field_info, field_name=field_name
         )
@@ -137,6 +142,7 @@ def is_bytes_field(field: ModelField) -> Any:
         return v1.is_bytes_field(field)
 
     from . import v2
+
     return v2.is_bytes_field(field)
 
 
@@ -148,6 +154,7 @@ def is_bytes_sequence_field(field: ModelField) -> Any:
         return v1.is_bytes_sequence_field(field)
 
     from . import v2
+
     return v2.is_bytes_sequence_field(field)
 
 
@@ -159,6 +166,7 @@ def is_scalar_field(field: ModelField) -> Any:
         return v1.is_scalar_field(field)
 
     from . import v2
+
     return v2.is_scalar_field(field)
 
 
@@ -170,6 +178,7 @@ def is_scalar_sequence_field(field: ModelField) -> Any:
         return v1.is_scalar_sequence_field(field)
 
     from . import v2
+
     return v2.is_scalar_sequence_field(field)
 
 
@@ -181,6 +190,7 @@ def is_sequence_field(field: ModelField) -> Any:
         return v1.is_sequence_field(field)
 
     from . import v2
+
     return v2.is_sequence_field(field)
 
 
@@ -192,18 +202,30 @@ def serialize_sequence_value(*, field: ModelField, value: Any) -> Any:
         return v1.serialize_sequence_value(field=field, value=value)
 
     from . import v2
+
     return v2.serialize_sequence_value(field=field, value=value)
 
 
 def get_compat_model_name_map(fields: List[ModelField]) -> Any:
     v1 = get_v1_if_loaded()
-    v1_model_fields = [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
+    v1_model_fields = (
-    v1_flat_models = v1.get_flat_models_from_fields(v1_model_fields, known_models=set()) if v1 and v1_model_fields else set()
+        [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
+    )
+    v1_flat_models = (
+        v1.get_flat_models_from_fields(v1_model_fields, known_models=set())
+        if v1 and v1_model_fields
+        else set()
+    )
     all_flat_models = v1_flat_models
     if PYDANTIC_V2:
         from . import v2
-        v2_model_fields = [field for field in fields if not v1_isinstance(field, "ModelField")]
+
-        v2_flat_models = v2.get_flat_models_from_fields(v2_model_fields, known_models=set())
+        v2_model_fields = [
+            field for field in fields if not v1_isinstance(field, "ModelField")
+        ]
+        v2_flat_models = v2.get_flat_models_from_fields(
+            v2_model_fields, known_models=set()
+        )
         all_flat_models = v1_flat_models | v2_flat_models
         model_name_map = v2.get_model_name_map(all_flat_models)
         return model_name_map
@@ -221,7 +243,9 @@ def get_definitions(
     Dict[str, Dict[str, Any]],
 ]:
     v1 = get_v1_if_loaded()
-    v1_fields = [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
+    v1_fields = (
+        [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
+    )
     if v1_fields and v1:
         v1_field_maps, v1_definitions = v1.get_definitions(
             fields=v1_fields,
@@ -233,7 +257,10 @@ def get_definitions(
         v1_definitions = {}
     if PYDANTIC_V2:
         from . import v2
-        v2_fields = [field for field in fields if not v1_isinstance(field, "ModelField")]
+
+        v2_fields = [
+            field for field in fields if not v1_isinstance(field, "ModelField")
+        ]
         v2_field_maps, v2_definitions = v2.get_definitions(
             fields=v2_fields,
             model_name_map=model_name_map,
@@ -266,6 +293,7 @@ def get_schema_from_model_field(
         )
     else:
         from . import v2
+
         return v2.get_schema_from_model_field(
             field=field,
             model_name_map=model_name_map,
@@ -279,6 +307,7 @@ def _is_model_field(value: Any) -> bool:
         return True
     elif PYDANTIC_V2:
         from . import v2
+
         return v2._is_model_field(value)
     else:
         return False
@@ -289,6 +318,7 @@ def _is_model_class(value: Any) -> bool:
         return True
     elif PYDANTIC_V2:
         from . import v2
+
         return v2._is_model_class(value)
     else:
         return False
@@ -302,12 +332,16 @@ def get_missing_field_error(loc: Tuple[str, ...], field: ModelField) -> Any:
         return v1.get_missing_field_error(loc=loc, field=field)
     else:
         from . import v2
+
         return v2.get_missing_field_error(loc=loc, field=field)
 
 
-def evaluate_forwardref(type_: Any, globalns: Dict[str, Any], localns: Dict[str, Any]) -> Any:
+def evaluate_forwardref(
+    type_: Any, globalns: Dict[str, Any], localns: Dict[str, Any]
+) -> Any:
     if PYDANTIC_V2:
         from . import v2
+
         return v2.evaluate_forwardref(type_, globalns, localns)
     else:
         v1 = get_v1_if_loaded()
@@ -334,7 +368,9 @@ def with_info_plain_validator_function(
         v1 = get_v1_if_loaded()
         if v1 is None:
             return func
-        return v1.with_info_plain_validator_function(func=func, info_argname=info_argname)
+        return v1.with_info_plain_validator_function(
+            func=func, info_argname=info_argname
+        )
 
 
 def _model_rebuild(model: Any) -> None:
@@ -344,6 +380,7 @@ def _model_rebuild(model: Any) -> None:
             v1._model_rebuild(model)
     elif PYDANTIC_V2:
         from . import v2
+
         v2._model_rebuild(model)
     else:
         model.update_forward_refs()

fastapi/_compat/shared.py

@@ -205,7 +205,9 @@ def annotation_is_pydantic_v1(annotation: Any) -> bool:
         return True
     origin = get_origin(annotation)
     if origin in (Union, UnionType):
-        return any(lenient_issubclass(arg, _v1.BaseModel) for arg in get_args(annotation))
+        return any(
+            lenient_issubclass(arg, _v1.BaseModel) for arg in get_args(annotation)
+        )
     if field_annotation_is_sequence(annotation):
         return any(annotation_is_pydantic_v1(sa) for sa in get_args(annotation))
     return False

fastapi/_compat/v1.py

@@ -1,4 +1,3 @@
-
 """
 Pydantic V1 compatibility module with lazy loading and controlled warnings.
 
@@ -26,6 +25,7 @@ RequiredParam = Ellipsis
 _pv1 = None
 _warned = False
 
+
 def _load() -> Any:
     global _pv1, _warned
     if _pv1 is not None:
@@ -42,6 +42,7 @@ def _load() -> Any:
     _pv1 = importlib.import_module("pydantic.v1")
     return _pv1
 
+
 def __getattr__(name: str) -> Any:
     if name == "RequiredParam":
         return Ellipsis
@@ -50,92 +51,176 @@ def __getattr__(name: str) -> Any:
     if hasattr(mod, name):
         return getattr(mod, name)
     # try common submodules
-    for sub in ("fields","schema","networks","types","color","class_validators",
+    for sub in (
-                "error_wrappers","errors","typing","utils"):
+        "fields",
+        "schema",
+        "networks",
+        "types",
+        "color",
+        "class_validators",
+        "error_wrappers",
+        "errors",
+        "typing",
+        "utils",
+    ):
         submod = getattr(mod, sub, None)
         if submod and hasattr(submod, name):
             return getattr(submod, name)
     raise AttributeError(name)
 
+
 # ---------- Wrappers used by FastAPI core (minimal) ----------
 
+
 def _normalize_errors(errors: Sequence[Any]) -> List[Dict[str, Any]]:
     pv1 = _load()
     RequestErrorModel = pv1.create_model("Request")
     out: List[Any] = []
     for err in errors:
         if isinstance(err, pv1.error_wrappers.ErrorWrapper):
-            out.extend(pv1.ValidationError(errors=[err], model=RequestErrorModel).errors())
+            out.extend(
+                pv1.ValidationError(errors=[err], model=RequestErrorModel).errors()
+            )
         elif isinstance(err, list):
             out.extend(_normalize_errors(err))
         else:
             out.append(err)
     return out
 
-def _regenerate_error_with_loc(*, errors: Sequence[Any], loc_prefix: Tuple[Any, ...]) -> List[Dict[str, Any]]:
+
-    return [{**e, "loc": loc_prefix + tuple(e.get("loc", ())) } for e in _normalize_errors(errors)]
+def _regenerate_error_with_loc(
+    *, errors: Sequence[Any], loc_prefix: Tuple[Any, ...]
+) -> List[Dict[str, Any]]:
+    return [
+        {**e, "loc": loc_prefix + tuple(e.get("loc", ()))}
+        for e in _normalize_errors(errors)
+    ]
+
 
 def _model_rebuild(model: Any) -> None:
     model.update_forward_refs()
 
-def _model_dump(model: Any, mode: Literal["json","python"]="json", **kwargs: Any) -> Any:
+
+def _model_dump(
+    model: Any, mode: Literal["json", "python"] = "json", **kwargs: Any
+) -> Any:
     return model.dict(**kwargs)
 
+
 def _get_model_config(model: Any) -> Any:
     return getattr(model, "__config__", None)
 
-def get_schema_from_model_field(*, field: Any, model_name_map: Any, field_mapping: Dict[Tuple[Any, Literal["validation","serialization"]], Dict[str, Any]], separate_input_output_schemas: bool=True) -> Dict[str, Any]:
+
+def get_schema_from_model_field(
+    *,
+    field: Any,
+    model_name_map: Any,
+    field_mapping: Dict[
+        Tuple[Any, Literal["validation", "serialization"]], Dict[str, Any]
+    ],
+    separate_input_output_schemas: bool = True,
+) -> Dict[str, Any]:
     schema = _load().schema
     ref = "#/components/schemas"
     return schema.field_schema(field, model_name_map=model_name_map, ref_prefix=ref)[0]
 
-def get_definitions(*, fields: List[Any], model_name_map: Any, separate_input_output_schemas: bool=True) -> Any:
+
+def get_definitions(
+    *,
+    fields: List[Any],
+    model_name_map: Any,
+    separate_input_output_schemas: bool = True,
+) -> Any:
     schema = _load().schema
     models = schema.get_flat_models_from_fields(fields, known_models=set())
     definitions: Dict[str, Dict[str, Any]] = {}
     for m in models:
-        m_schema, m_defs, _ = schema.model_process_schema(m, model_name_map=model_name_map, ref_prefix="#/components/schemas")
+        m_schema, m_defs, _ = schema.model_process_schema(
+            m, model_name_map=model_name_map, ref_prefix="#/components/schemas"
+        )
         definitions.update(m_defs)
         definitions[model_name_map[m]] = m_schema
     return {}, definitions
 
+
 def get_model_fields(model: Any) -> List[Any]:
     return list(getattr(model, "__fields__", {}).values())
 
+
 def is_bytes_field(field: Any) -> bool:
     return _load().utils.lenient_issubclass(field.type_, bytes)
 
+
 def is_bytes_sequence_field(field: Any) -> bool:
     f = _load().fields
-    shapes = {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}
+    shapes = {
-    return field.shape in shapes and _load().utils.lenient_issubclass(field.type_, bytes)
+        f.SHAPE_LIST,
+        f.SHAPE_SET,
+        f.SHAPE_FROZENSET,
+        f.SHAPE_TUPLE,
+        f.SHAPE_SEQUENCE,
+        f.SHAPE_TUPLE_ELLIPSIS,
+    }
+    return field.shape in shapes and _load().utils.lenient_issubclass(
+        field.type_, bytes
+    )
+
 
 def is_scalar_field(field: Any) -> bool:
     f = _load().fields
     pv1 = _load()
-    return (field.shape == f.SHAPE_SINGLETON
+    return (
-            and not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel)
+        field.shape == f.SHAPE_SINGLETON
-            and not pv1.utils.lenient_issubclass(field.type_, dict))
+        and not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel)
+        and not pv1.utils.lenient_issubclass(field.type_, dict)
+    )
+
 
 def is_sequence_field(field: Any) -> bool:
     f = _load().fields
-    return field.shape in {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}
+    return field.shape in {
+        f.SHAPE_LIST,
+        f.SHAPE_SET,
+        f.SHAPE_FROZENSET,
+        f.SHAPE_TUPLE,
+        f.SHAPE_SEQUENCE,
+        f.SHAPE_TUPLE_ELLIPSIS,
+    }
+
 
 def is_scalar_sequence_field(field: Any) -> bool:
     f = _load().fields
     pv1 = _load()
-    if field.shape in {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}:
+    if field.shape in {
-        return not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel) and all(is_scalar_field(sf) for sf in (field.sub_fields or []))
+        f.SHAPE_LIST,
+        f.SHAPE_SET,
+        f.SHAPE_FROZENSET,
+        f.SHAPE_TUPLE,
+        f.SHAPE_SEQUENCE,
+        f.SHAPE_TUPLE_ELLIPSIS,
+    }:
+        return not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel) and all(
+            is_scalar_field(sf) for sf in (field.sub_fields or [])
+        )
     return False
 
+
 def copy_field_info(*, field_info: Any, annotation: Any) -> Any:
     return _copy(field_info)
 
+
 def serialize_sequence_value(*, field: Any, value: Any) -> Any:
     f = _load().fields
-    mapping = { f.SHAPE_LIST: list, f.SHAPE_SET: set, f.SHAPE_TUPLE: tuple, f.SHAPE_SEQUENCE: list, f.SHAPE_TUPLE_ELLIPSIS: list }
+    mapping = {
+        f.SHAPE_LIST: list,
+        f.SHAPE_SET: set,
+        f.SHAPE_TUPLE: tuple,
+        f.SHAPE_SEQUENCE: list,
+        f.SHAPE_TUPLE_ELLIPSIS: list,
+    }
     return mapping[field.shape](value)
 
+
 # Type aliases for backward compatibility
 GetJsonSchemaHandler = Any
 JsonSchemaValue = dict[str, Any]

fastapi/_compat/v1.pyi

@@ -16,12 +16,24 @@ class Validator: ...
 
 def create_model(__name__: str, **kwargs: Any) -> Any: ...
 def _model_rebuild(model: Any) -> None: ...
-def _model_dump(model: Any, mode: Literal["json","python"]="json", **kwargs: Any) -> Any: ...
+def _model_dump(
+    model: Any, mode: Literal["json", "python"] = "json", **kwargs: Any
+) -> Any: ...
 def _get_model_config(model: Any) -> Any: ...
 def _normalize_errors(errors: Any) -> Any: ...
-def _regenerate_error_with_loc(*, errors: Any, loc_prefix: tuple[Any, ...]) -> list[dict[str, Any]]: ...
+def _regenerate_error_with_loc(
-def get_schema_from_model_field(*, field: Any, model_name_map: Any, field_mapping: Any, separate_input_output_schemas: bool=True) -> dict[str, Any]: ...
+    *, errors: Any, loc_prefix: tuple[Any, ...]
-def get_definitions(*, fields: Any, model_name_map: Any, separate_input_output_schemas: bool=True) -> tuple[Any, dict[str, dict[str, Any]]]: ...
+) -> list[dict[str, Any]]: ...
+def get_schema_from_model_field(
+    *,
+    field: Any,
+    model_name_map: Any,
+    field_mapping: Any,
+    separate_input_output_schemas: bool = True,
+) -> dict[str, Any]: ...
+def get_definitions(
+    *, fields: Any, model_name_map: Any, separate_input_output_schemas: bool = True
+) -> tuple[Any, dict[str, dict[str, Any]]]: ...
 def get_model_fields(model: Any) -> list[Any]: ...
 def is_bytes_field(field: Any) -> bool: ...
 def is_bytes_sequence_field(field: Any) -> bool: ...
@@ -30,8 +42,12 @@ def is_scalar_sequence_field(field: Any) -> bool: ...
 def is_sequence_field(field: Any) -> bool: ...
 def copy_field_info(field_info: Any, annotation: Any, **kwargs: Any) -> Any: ...
 def create_body_model(fields: Any, model_name: str) -> Any: ...
-def evaluate_forwardref(type_: Any, globalns: dict[str, Any], localns: dict[str, Any]) -> Any: ...
+def evaluate_forwardref(
-def get_annotation_from_field_info(annotation: Any, field_info: Any, field_name: str) -> Any: ...
+    type_: Any, globalns: dict[str, Any], localns: dict[str, Any]
+) -> Any: ...
+def get_annotation_from_field_info(
+    annotation: Any, field_info: Any, field_name: str
+) -> Any: ...
 def get_missing_field_error(loc: tuple[str, ...], field: Any) -> dict[str, Any]: ...
 def serialize_sequence_value(*, field: Any, value: Any) -> Any: ...
 def with_info_plain_validator_function(func: Any) -> Any: ...

fastapi/_compat/v2.py

@@ -40,8 +40,10 @@ from typing_extensions import Annotated, Literal, get_args, get_origin
 def _get_v1() -> Any:
     """Lazy import of v1 module to avoid warnings."""
     from fastapi._compat import v1
+
     return v1
 
+
 try:
     from pydantic_core.core_schema import (
         with_info_plain_validator_function as with_info_plain_validator_function,

fastapi/_compat/v2.pyi

@@ -12,15 +12,27 @@ class Validator: ...
 
 def _is_model_class(value: Any) -> bool: ...
 def _model_rebuild(model: Any) -> None: ...
-def evaluate_forwardref(type_: Any, globalns: dict[str, Any], localns: dict[str, Any]) -> Any: ...
+def evaluate_forwardref(
+    type_: Any, globalns: dict[str, Any], localns: dict[str, Any]
+) -> Any: ...
 def _get_model_config(model: Any) -> Any: ...
 def _model_dump(model: Any, **kwargs: Any) -> Any: ...
 def copy_field_info(field_info: Any, annotation: Any, **kwargs: Any) -> Any: ...
 def create_body_model(fields: Any, model_name: str) -> Any: ...
-def get_annotation_from_field_info(annotation: Any, field_info: Any, field_name: str) -> Any: ...
+def get_annotation_from_field_info(
-def get_definitions(*, fields: Any, model_name_map: Any, separate_input_output_schemas: bool=True) -> tuple[Any, dict[str, dict[str, Any]]]: ...
+    annotation: Any, field_info: Any, field_name: str
+) -> Any: ...
+def get_definitions(
+    *, fields: Any, model_name_map: Any, separate_input_output_schemas: bool = True
+) -> tuple[Any, dict[str, dict[str, Any]]]: ...
 def get_missing_field_error(loc: tuple[str, ...], field: Any) -> dict[str, Any]: ...
-def get_schema_from_model_field(*, field: Any, model_name_map: Any, field_mapping: Any, separate_input_output_schemas: bool=True) -> dict[str, Any]: ...
+def get_schema_from_model_field(
+    *,
+    field: Any,
+    model_name_map: Any,
+    field_mapping: Any,
+    separate_input_output_schemas: bool = True,
+) -> dict[str, Any]: ...
 def is_bytes_field(field: Any) -> bool: ...
 def is_bytes_sequence_field(field: Any) -> bool: ...
 def is_scalar_field(field: Any) -> bool: ...

fastapi/datastructures.py

@@ -168,6 +168,7 @@ class UploadFile(StarletteUploadFile):
         cls, source: Type[Any], handler: Callable[[Any], CoreSchema]
     ) -> CoreSchema:
         from fastapi._compat.v2 import with_info_plain_validator_function  # noqa: I001
+
         return with_info_plain_validator_function(cls._validate)
 
 

fastapi/dependencies/utils.py

@@ -83,8 +83,10 @@ from .._compat import _v1_params as temp_pydantic_v1_params
 def _get_v1() -> Any:
     """Lazy import of v1 module to avoid warnings."""
     from fastapi._compat import v1
+
     return v1
 
+
 if sys.version_info >= (3, 13):  # pragma: no cover
     from inspect import iscoroutinefunction
 else:  # pragma: no cover
@@ -531,7 +533,8 @@ def analyze_param(
             type_=use_annotation_from_field_info,
             default=field_info.default,
             alias=alias,
-            required=field_info.default in (RequiredParam, _get_v1().RequiredParam, Undefined),
+            required=field_info.default
+            in (RequiredParam, _get_v1().RequiredParam, Undefined),
             field_info=field_info,  # type: ignore[arg-type]
         )
         if is_path_param:

fastapi/encoders.py

@@ -86,6 +86,7 @@ ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
     AnyUrl: str,
 }
 
+
 def _ensure_v1_encoders_registered() -> None:
     """Register V1 encoders only when needed (lazy loading)."""
     # Only register if pydantic.v1 is already loaded (app actually used v1)
@@ -236,7 +237,7 @@ def jsonable_encoder(
         # Check if it's a v1 model using lazy loading
         if "pydantic.v1" in sys.modules:
             if isinstance(obj, v1.BaseModel):
-                    encoders = getattr(obj.__config__, "json_encoders", {})  # type: ignore[attr-defined]
+                encoders = getattr(obj.__config__, "json_encoders", {})  # type: ignore[attr-defined]
         if custom_encoder:
             encoders = {**encoders, **custom_encoder}
         obj_dict = _model_dump(

fastapi/utils.py

@@ -33,12 +33,15 @@ from typing_extensions import Literal
 if TYPE_CHECKING:  # pragma: nocover
     from .routing import APIRoute
 
+
 # Lazy import of v1 to avoid warnings
 def _get_v1() -> Any:
     """Lazy import of v1 module to avoid warnings."""
     from fastapi._compat import v1
+
     return v1
 
+
 # Cache for `create_cloned_field`
 _CLONED_TYPES_CACHE: MutableMapping[Type[BaseModel], Type[BaseModel]] = (
     WeakKeyDictionary()
@@ -146,7 +149,9 @@ def create_cloned_field(
         original_type = cast(Type[Any], original_type)
         use_type = cloned_types.get(original_type)
         if use_type is None:
-            use_type = _get_v1().create_model(original_type.__name__, __base__=original_type)
+            use_type = _get_v1().create_model(
+                original_type.__name__, __base__=original_type
+            )
             cloned_types[original_type] = use_type
             for f in original_type.__fields__.values():
                 use_type.__fields__[f.name] = create_cloned_field(

tests/test_pydantic_v2_first_compat.py

@@ -45,21 +45,23 @@ class TestV2FirstCompatibility:
             result = jsonable_encoder(model)
 
             assert result == {"name": "test", "value": 42}
-            assert len(w) == 0, f"Unexpected warnings: {[str(warning.message) for warning in w]}"
+            assert len(w) == 0, (
+                f"Unexpected warnings: {[str(warning.message) for warning in w]}"
+            )
 
     def test_proxy_classes_isinstance(self):
         """Test that proxy classes work correctly with isinstance()."""
         import fastapi.temp_pydantic_v1_params as T
 
         # Test that proxy classes are available
-        assert hasattr(T, 'Param')
+        assert hasattr(T, "Param")
-        assert hasattr(T, 'Body')
+        assert hasattr(T, "Body")
-        assert hasattr(T, 'Form')
+        assert hasattr(T, "Form")
-        assert hasattr(T, 'File')
+        assert hasattr(T, "File")
-        assert hasattr(T, 'Query')
+        assert hasattr(T, "Query")
-        assert hasattr(T, 'Header')
+        assert hasattr(T, "Header")
-        assert hasattr(T, 'Cookie')
+        assert hasattr(T, "Cookie")
-        assert hasattr(T, 'Path')
+        assert hasattr(T, "Path")
 
         # Test isinstance() with proxy classes (expect DeprecationWarning)
         with warnings.catch_warnings():
@@ -69,6 +71,7 @@ class TestV2FirstCompatibility:
 
         # Test that proxy classes are different from v2 params
         from fastapi import params as v2_params
+
         assert T.Param is not v2_params.Param
         assert T.Body is not v2_params.Body
 
@@ -99,14 +102,14 @@ class TestV2FirstCompatibility:
             path = Path()
 
         # Test that they have expected attributes
-        assert hasattr(body, 'default')
+        assert hasattr(body, "default")
-        assert hasattr(query, 'default')
+        assert hasattr(query, "default")
-        assert hasattr(form, 'default')
+        assert hasattr(form, "default")
-        assert hasattr(file_param, 'default')
+        assert hasattr(file_param, "default")
-        assert hasattr(param, 'default')
+        assert hasattr(param, "default")
-        assert hasattr(header, 'default')
+        assert hasattr(header, "default")
-        assert hasattr(cookie, 'default')
+        assert hasattr(cookie, "default")
-        assert hasattr(path, 'default')
+        assert hasattr(path, "default")
 
     def test_lazy_loading_behavior(self):
         """Test that v1 is only loaded when actually used."""
@@ -124,6 +127,7 @@ class TestV2FirstCompatibility:
 
         # Test that we can still access v1 proxy
         from fastapi._compat import v1
+
         assert v1 is not None
 
     def test_encoders_lazy_registration(self):
@@ -148,14 +152,14 @@ class TestV2FirstCompatibility:
             from fastapi._compat import v1
 
             # Test that v1 proxy has expected attributes
-            assert hasattr(v1, 'BaseModel')
+            assert hasattr(v1, "BaseModel")
-            assert hasattr(v1, 'FieldInfo')
+            assert hasattr(v1, "FieldInfo")
-            assert hasattr(v1, 'ValidationError')
+            assert hasattr(v1, "ValidationError")
 
             # Test that wrapper functions are available
-            assert hasattr(v1, '_normalize_errors')
+            assert hasattr(v1, "_normalize_errors")
-            assert hasattr(v1, '_model_dump')
+            assert hasattr(v1, "_model_dump")
-            assert hasattr(v1, '_model_rebuild')
+            assert hasattr(v1, "_model_rebuild")
 
     def test_strict_mode_environment_variable(self):
         """Test FASTAPI_PYDANTIC_V1_STRICT environment variable behavior."""
@@ -170,6 +174,7 @@ class TestV2FirstCompatibility:
 
             # This should not raise an error unless we actually try to use v1
             from fastapi import FastAPI
+
             _ = FastAPI()  # Create app but don't use it
 
             # The strict mode only affects actual v1 usage, not imports
@@ -194,14 +199,14 @@ class TestV2FirstCompatibility:
             form = Form()
 
         # Test that they delegate to internal v1.FieldInfo
-        assert hasattr(param, '_fi')
+        assert hasattr(param, "_fi")
-        assert hasattr(body, '_fi')
+        assert hasattr(body, "_fi")
-        assert hasattr(form, '_fi')
+        assert hasattr(form, "_fi")
 
         # Test that they have expected interface
-        assert hasattr(param, 'default')
+        assert hasattr(param, "default")
-        assert hasattr(body, 'default')
+        assert hasattr(body, "default")
-        assert hasattr(form, 'default')
+        assert hasattr(form, "default")
 
 
 if __name__ == "__main__":