Загрузить файлы в «venv/Lib/site-packages/fastapi»

venv/Lib/site-packages/fastapi/concurrency.py

@@ -0,0 +1,41 @@
+from collections.abc import AsyncGenerator
+from contextlib import AbstractContextManager
+from contextlib import asynccontextmanager as asynccontextmanager
+from typing import TypeVar
+
+import anyio.to_thread
+from anyio import CapacityLimiter
+from starlette.concurrency import iterate_in_threadpool as iterate_in_threadpool  # noqa
+from starlette.concurrency import run_in_threadpool as run_in_threadpool  # noqa
+from starlette.concurrency import (  # noqa
+    run_until_first_complete as run_until_first_complete,
+)
+
+_T = TypeVar("_T")
+
+
+@asynccontextmanager
+async def contextmanager_in_threadpool(
+    cm: AbstractContextManager[_T],
+) -> AsyncGenerator[_T, None]:
+    # blocking __exit__ from running waiting on a free thread
+    # can create race conditions/deadlocks if the context manager itself
+    # has its own internal pool (e.g. a database connection pool)
+    # to avoid this we let __exit__ run without a capacity limit
+    # since we're creating a new limiter for each call, any non-zero limit
+    # works (1 is arbitrary)
+    exit_limiter = CapacityLimiter(1)
+    try:
+        yield await run_in_threadpool(cm.__enter__)
+    except Exception as e:
+        ok = bool(
+            await anyio.to_thread.run_sync(
+                cm.__exit__, type(e), e, e.__traceback__, limiter=exit_limiter
+            )
+        )
+        if not ok:
+            raise e
+    else:
+        await anyio.to_thread.run_sync(
+            cm.__exit__, None, None, None, limiter=exit_limiter
+        )

venv/Lib/site-packages/fastapi/datastructures.py

@@ -0,0 +1,186 @@
+from collections.abc import Callable, Mapping
+from typing import (
+    Annotated,
+    Any,
+    BinaryIO,
+    TypeVar,
+    cast,
+)
+
+from annotated_doc import Doc
+from pydantic import GetJsonSchemaHandler
+from starlette.datastructures import URL as URL  # noqa: F401
+from starlette.datastructures import Address as Address  # noqa: F401
+from starlette.datastructures import FormData as FormData  # noqa: F401
+from starlette.datastructures import Headers as Headers  # noqa: F401
+from starlette.datastructures import QueryParams as QueryParams  # noqa: F401
+from starlette.datastructures import State as State  # noqa: F401
+from starlette.datastructures import UploadFile as StarletteUploadFile
+
+
+class UploadFile(StarletteUploadFile):
+    """
+    A file uploaded in a request.
+
+    Define it as a *path operation function* (or dependency) parameter.
+
+    If you are using a regular `def` function, you can use the `upload_file.file`
+    attribute to access the raw standard Python file (blocking, not async), useful and
+    needed for non-async code.
+
+    Read more about it in the
+    [FastAPI docs for Request Files](https://fastapi.tiangolo.com/tutorial/request-files/).
+
+    ## Example
+
+    ```python
+    from typing import Annotated
+
+    from fastapi import FastAPI, File, UploadFile
+
+    app = FastAPI()
+
+
+    @app.post("/files/")
+    async def create_file(file: Annotated[bytes, File()]):
+        return {"file_size": len(file)}
+
+
+    @app.post("/uploadfile/")
+    async def create_upload_file(file: UploadFile):
+        return {"filename": file.filename}
+    ```
+    """
+
+    file: Annotated[
+        BinaryIO,
+        Doc("The standard Python file object (non-async)."),
+    ]
+    filename: Annotated[str | None, Doc("The original file name.")]
+    size: Annotated[int | None, Doc("The size of the file in bytes.")]
+    headers: Annotated[Headers, Doc("The headers of the request.")]
+    content_type: Annotated[
+        str | None, Doc("The content type of the request, from the headers.")
+    ]
+
+    async def write(
+        self,
+        data: Annotated[
+            bytes,
+            Doc(
+                """
+                The bytes to write to the file.
+                """
+            ),
+        ],
+    ) -> None:
+        """
+        Write some bytes to the file.
+
+        You normally wouldn't use this from a file you read in a request.
+
+        To be awaitable, compatible with async, this is run in threadpool.
+        """
+        return await super().write(data)
+
+    async def read(
+        self,
+        size: Annotated[
+            int,
+            Doc(
+                """
+                The number of bytes to read from the file.
+                """
+            ),
+        ] = -1,
+    ) -> bytes:
+        """
+        Read some bytes from the file.
+
+        To be awaitable, compatible with async, this is run in threadpool.
+        """
+        return await super().read(size)
+
+    async def seek(
+        self,
+        offset: Annotated[
+            int,
+            Doc(
+                """
+                The position in bytes to seek to in the file.
+                """
+            ),
+        ],
+    ) -> None:
+        """
+        Move to a position in the file.
+
+        Any next read or write will be done from that position.
+
+        To be awaitable, compatible with async, this is run in threadpool.
+        """
+        return await super().seek(offset)
+
+    async def close(self) -> None:
+        """
+        Close the file.
+
+        To be awaitable, compatible with async, this is run in threadpool.
+        """
+        return await super().close()
+
+    @classmethod
+    def _validate(cls, __input_value: Any, _: Any) -> "UploadFile":
+        if not isinstance(__input_value, StarletteUploadFile):
+            raise ValueError(f"Expected UploadFile, received: {type(__input_value)}")
+        return cast(UploadFile, __input_value)
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler
+    ) -> dict[str, Any]:
+        return {"type": "string", "contentMediaType": "application/octet-stream"}
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source: type[Any], handler: Callable[[Any], Mapping[str, Any]]
+    ) -> Mapping[str, Any]:
+        from ._compat.v2 import with_info_plain_validator_function
+
+        return with_info_plain_validator_function(cls._validate)
+
+
+class DefaultPlaceholder:
+    """
+    You shouldn't use this class directly.
+
+    It's used internally to recognize when a default value has been overwritten, even
+    if the overridden default value was truthy.
+    """
+
+    def __init__(self, value: Any):
+        self.value = value
+
+    def __bool__(self) -> bool:
+        return bool(self.value)
+
+    def __eq__(self, o: object) -> bool:
+        return isinstance(o, DefaultPlaceholder) and o.value == self.value
+
+
+DefaultType = TypeVar("DefaultType")
+
+
+def Default(value: DefaultType) -> DefaultType:
+    """
+    You shouldn't use this function directly.
+
+    It's used internally to recognize when a default value has been overwritten, even
+    if the overridden default value was truthy.
+    """
+    return DefaultPlaceholder(value)  # type: ignore
+
+
+# Sentinel for "parameter not provided" in Param/FieldInfo.
+# Typed as None to satisfy ty
+_Unset = Default(None)

venv/Lib/site-packages/fastapi/encoders.py

@@ -0,0 +1,364 @@
+import dataclasses
+import datetime
+from collections import defaultdict, deque
+from collections.abc import Callable
+from decimal import Decimal
+from enum import Enum
+from ipaddress import (
+    IPv4Address,
+    IPv4Interface,
+    IPv4Network,
+    IPv6Address,
+    IPv6Interface,
+    IPv6Network,
+)
+from pathlib import Path, PurePath
+from re import Pattern
+from types import GeneratorType
+from typing import Annotated, Any
+from uuid import UUID
+
+from annotated_doc import Doc
+from fastapi.exceptions import PydanticV1NotSupportedError
+from fastapi.types import IncEx
+from pydantic import BaseModel
+from pydantic.networks import AnyUrl, NameEmail
+from pydantic.types import SecretBytes, SecretStr
+from pydantic_core import PydanticUndefinedType
+
+from ._compat import (
+    Url,
+    is_pydantic_v1_model_instance,
+)
+
+try:
+    # pydantic.color.Color is deprecated since v2.0b3, but supporting for bwd-compat
+    from pydantic.color import Color  # ty: ignore[deprecated]
+except ImportError:  # pragma: no cover
+
+    class Color:  # type: ignore[no-redef]
+        pass
+
+
+try:
+    # Supporting the new Color format for newer versions of Pydantic
+    from pydantic_extra_types.color import Color as PyExtraColor
+except ImportError:  # pragma: no cover
+
+    class PyExtraColor:  # type: ignore[no-redef]
+        pass
+
+
+# Taken from Pydantic v1 as is
+def isoformat(o: datetime.date | datetime.time) -> str:
+    return o.isoformat()
+
+
+# Adapted from Pydantic v1
+# TODO: pv2 should this return strings instead?
+def decimal_encoder(dec_value: Decimal) -> int | float:
+    """
+    Encodes a Decimal as int if there's no exponent, otherwise float
+
+    This is useful when we use ConstrainedDecimal to represent Numeric(x,0)
+    where an integer (but not int typed) is used. Encoding this as a float
+    results in failed round-tripping between encode and parse.
+    Our Id type is a prime example of this.
+
+    >>> decimal_encoder(Decimal("1.0"))
+    1.0
+
+    >>> decimal_encoder(Decimal("1"))
+    1
+
+    >>> decimal_encoder(Decimal("NaN"))
+    nan
+    """
+    exponent = dec_value.as_tuple().exponent
+    if isinstance(exponent, int) and exponent >= 0:
+        return int(dec_value)
+    else:
+        return float(dec_value)
+
+
+ENCODERS_BY_TYPE: dict[type[Any], Callable[[Any], Any]] = {
+    bytes: lambda o: o.decode(),
+    Color: str,
+    PyExtraColor: str,
+    datetime.date: isoformat,
+    datetime.datetime: isoformat,
+    datetime.time: isoformat,
+    datetime.timedelta: lambda td: td.total_seconds(),
+    Decimal: decimal_encoder,
+    Enum: lambda o: o.value,
+    frozenset: list,
+    deque: list,
+    GeneratorType: list,
+    IPv4Address: str,
+    IPv4Interface: str,
+    IPv4Network: str,
+    IPv6Address: str,
+    IPv6Interface: str,
+    IPv6Network: str,
+    NameEmail: str,
+    Path: str,
+    Pattern: lambda o: o.pattern,
+    SecretBytes: str,
+    SecretStr: str,
+    set: list,
+    UUID: str,
+    Url: str,
+    AnyUrl: str,
+}
+
+
+def generate_encoders_by_class_tuples(
+    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
+    )
+    for type_, encoder in type_encoder_map.items():
+        encoders_by_class_tuples[encoder] += (type_,)
+    return encoders_by_class_tuples
+
+
+encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)
+
+
+def jsonable_encoder(
+    obj: Annotated[
+        Any,
+        Doc(
+            """
+            The input object to convert to JSON.
+            """
+        ),
+    ],
+    include: Annotated[
+        IncEx | None,
+        Doc(
+            """
+            Pydantic's `include` parameter, passed to Pydantic models to set the
+            fields to include.
+            """
+        ),
+    ] = None,
+    exclude: Annotated[
+        IncEx | None,
+        Doc(
+            """
+            Pydantic's `exclude` parameter, passed to Pydantic models to set the
+            fields to exclude.
+            """
+        ),
+    ] = None,
+    by_alias: Annotated[
+        bool,
+        Doc(
+            """
+            Pydantic's `by_alias` parameter, passed to Pydantic models to define if
+            the output should use the alias names (when provided) or the Python
+            attribute names. In an API, if you set an alias, it's probably because you
+            want to use it in the result, so you probably want to leave this set to
+            `True`.
+            """
+        ),
+    ] = True,
+    exclude_unset: Annotated[
+        bool,
+        Doc(
+            """
+            Pydantic's `exclude_unset` parameter, passed to Pydantic models to define
+            if it should exclude from the output the fields that were not explicitly
+            set (and that only had their default values).
+            """
+        ),
+    ] = False,
+    exclude_defaults: Annotated[
+        bool,
+        Doc(
+            """
+            Pydantic's `exclude_defaults` parameter, passed to Pydantic models to define
+            if it should exclude from the output the fields that had the same default
+            value, even when they were explicitly set.
+            """
+        ),
+    ] = False,
+    exclude_none: Annotated[
+        bool,
+        Doc(
+            """
+            Pydantic's `exclude_none` parameter, passed to Pydantic models to define
+            if it should exclude from the output any fields that have a `None` value.
+            """
+        ),
+    ] = False,
+    custom_encoder: Annotated[
+        dict[Any, Callable[[Any], Any]] | None,
+        Doc(
+            """
+            Pydantic's `custom_encoder` parameter, passed to Pydantic models to define
+            a custom encoder.
+            """
+        ),
+    ] = None,
+    sqlalchemy_safe: Annotated[
+        bool,
+        Doc(
+            """
+            Exclude from the output any fields that start with the name `_sa`.
+
+            This is mainly a hack for compatibility with SQLAlchemy objects, they
+            store internal SQLAlchemy-specific state in attributes named with `_sa`,
+            and those objects can't (and shouldn't be) serialized to JSON.
+            """
+        ),
+    ] = True,
+) -> Any:
+    """
+    Convert any object to something that can be encoded in JSON.
+
+    This is used internally by FastAPI to make sure anything you return can be
+    encoded as JSON before it is sent to the client.
+
+    You can also use it yourself, for example to convert objects before saving them
+    in a database that supports only JSON.
+
+    Read more about it in the
+    [FastAPI docs for JSON Compatible Encoder](https://fastapi.tiangolo.com/tutorial/encoder/).
+    """
+    custom_encoder = custom_encoder or {}
+    if custom_encoder:
+        if type(obj) in custom_encoder:
+            return custom_encoder[type(obj)](obj)
+        else:
+            for encoder_type, encoder_instance in custom_encoder.items():
+                if isinstance(obj, encoder_type):
+                    return encoder_instance(obj)
+    if include is not None and not isinstance(include, (set, dict)):
+        include = set(include)  # type: ignore[assignment]  # ty: ignore[invalid-assignment]
+    if exclude is not None and not isinstance(exclude, (set, dict)):
+        exclude = set(exclude)  # type: ignore[assignment]  # ty: ignore[invalid-assignment]
+    if isinstance(obj, BaseModel):
+        obj_dict = obj.model_dump(
+            mode="json",
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+        )
+        return jsonable_encoder(
+            obj_dict,
+            exclude_none=exclude_none,
+            exclude_defaults=exclude_defaults,
+            sqlalchemy_safe=sqlalchemy_safe,
+        )
+    if dataclasses.is_dataclass(obj):
+        assert not isinstance(obj, type)
+        obj_dict = dataclasses.asdict(obj)
+        return jsonable_encoder(
+            obj_dict,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_defaults=exclude_defaults,
+            exclude_none=exclude_none,
+            custom_encoder=custom_encoder,
+            sqlalchemy_safe=sqlalchemy_safe,
+        )
+    if isinstance(obj, Enum):
+        return obj.value
+    if isinstance(obj, PurePath):
+        return str(obj)
+    if isinstance(obj, (str, int, float, type(None))):
+        return obj
+    if isinstance(obj, PydanticUndefinedType):
+        return None
+    if isinstance(obj, dict):
+        encoded_dict = {}
+        allowed_keys = set(obj.keys())
+        if include is not None:
+            allowed_keys &= set(include)
+        if exclude is not None:
+            allowed_keys -= set(exclude)
+        for key, value in obj.items():
+            if (
+                (
+                    not sqlalchemy_safe
+                    or (not isinstance(key, str))
+                    or (not key.startswith("_sa"))
+                )
+                and (value is not None or not exclude_none)
+                and key in allowed_keys
+            ):
+                encoded_key = jsonable_encoder(
+                    key,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+                encoded_value = jsonable_encoder(
+                    value,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+                encoded_dict[encoded_key] = encoded_value
+        return encoded_dict
+    if isinstance(obj, (list, set, frozenset, GeneratorType, tuple, deque)):
+        encoded_list = []
+        for item in obj:
+            encoded_list.append(
+                jsonable_encoder(
+                    item,
+                    include=include,
+                    exclude=exclude,
+                    by_alias=by_alias,
+                    exclude_unset=exclude_unset,
+                    exclude_defaults=exclude_defaults,
+                    exclude_none=exclude_none,
+                    custom_encoder=custom_encoder,
+                    sqlalchemy_safe=sqlalchemy_safe,
+                )
+            )
+        return encoded_list
+
+    if type(obj) in ENCODERS_BY_TYPE:
+        return ENCODERS_BY_TYPE[type(obj)](obj)
+    for encoder, classes_tuple in encoders_by_class_tuples.items():
+        if isinstance(obj, classes_tuple):
+            return encoder(obj)
+    if is_pydantic_v1_model_instance(obj):
+        raise PydanticV1NotSupportedError(
+            "pydantic.v1 models are no longer supported by FastAPI."
+            f" Please update the model {obj!r}."
+        )
+    try:
+        data = dict(obj)
+    except Exception as e:
+        errors: list[Exception] = []
+        errors.append(e)
+        try:
+            data = vars(obj)
+        except Exception as e:
+            errors.append(e)
+            raise ValueError(errors) from e
+    return jsonable_encoder(
+        data,
+        include=include,
+        exclude=exclude,
+        by_alias=by_alias,
+        exclude_unset=exclude_unset,
+        exclude_defaults=exclude_defaults,
+        exclude_none=exclude_none,
+        custom_encoder=custom_encoder,
+        sqlalchemy_safe=sqlalchemy_safe,
+    )

venv/Lib/site-packages/fastapi/exception_handlers.py

@@ -0,0 +1,34 @@
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError
+from fastapi.utils import is_body_allowed_for_status_code
+from fastapi.websockets import WebSocket
+from starlette.exceptions import HTTPException
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.status import WS_1008_POLICY_VIOLATION
+
+
+async def http_exception_handler(request: Request, exc: HTTPException) -> Response:
+    headers = getattr(exc, "headers", None)
+    if not is_body_allowed_for_status_code(exc.status_code):
+        return Response(status_code=exc.status_code, headers=headers)
+    return JSONResponse(
+        {"detail": exc.detail}, status_code=exc.status_code, headers=headers
+    )
+
+
+async def request_validation_exception_handler(
+    request: Request, exc: RequestValidationError
+) -> JSONResponse:
+    return JSONResponse(
+        status_code=422,
+        content={"detail": jsonable_encoder(exc.errors())},
+    )
+
+
+async def websocket_request_validation_exception_handler(
+    websocket: WebSocket, exc: WebSocketRequestValidationError
+) -> None:
+    await websocket.close(
+        code=WS_1008_POLICY_VIOLATION, reason=jsonable_encoder(exc.errors())
+    )

venv/Lib/site-packages/fastapi/exceptions.py

@@ -0,0 +1,256 @@
+from collections.abc import Mapping, Sequence
+from typing import Annotated, Any, TypedDict
+
+from annotated_doc import Doc
+from pydantic import BaseModel, create_model
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.exceptions import WebSocketException as StarletteWebSocketException
+
+
+class EndpointContext(TypedDict, total=False):
+    function: str
+    path: str
+    file: str
+    line: int
+
+
+class HTTPException(StarletteHTTPException):
+    """
+    An HTTP exception you can raise in your own code to show errors to the client.
+
+    This is for client errors, invalid authentication, invalid data, etc. Not for server
+    errors in your code.
+
+    Read more about it in the
+    [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).
+
+    ## Example
+
+    ```python
+    from fastapi import FastAPI, HTTPException
+
+    app = FastAPI()
+
+    items = {"foo": "The Foo Wrestlers"}
+
+
+    @app.get("/items/{item_id}")
+    async def read_item(item_id: str):
+        if item_id not in items:
+            raise HTTPException(status_code=404, detail="Item not found")
+        return {"item": items[item_id]}
+    ```
+    """
+
+    def __init__(
+        self,
+        status_code: Annotated[
+            int,
+            Doc(
+                """
+                HTTP status code to send to the client.
+
+                Read more about it in the
+                [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)
+                """
+            ),
+        ],
+        detail: Annotated[
+            Any,
+            Doc(
+                """
+                Any data to be sent to the client in the `detail` key of the JSON
+                response.
+
+                Read more about it in the
+                [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-httpexception)
+                """
+            ),
+        ] = None,
+        headers: Annotated[
+            Mapping[str, str] | None,
+            Doc(
+                """
+                Any headers to send to the client in the response.
+
+                Read more about it in the
+                [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#add-custom-headers)
+
+                """
+            ),
+        ] = None,
+    ) -> None:
+        super().__init__(status_code=status_code, detail=detail, headers=headers)
+
+
+class WebSocketException(StarletteWebSocketException):
+    """
+    A WebSocket exception you can raise in your own code to show errors to the client.
+
+    This is for client errors, invalid authentication, invalid data, etc. Not for server
+    errors in your code.
+
+    Read more about it in the
+    [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/).
+
+    ## Example
+
+    ```python
+    from typing import Annotated
+
+    from fastapi import (
+        Cookie,
+        FastAPI,
+        WebSocket,
+        WebSocketException,
+        status,
+    )
+
+    app = FastAPI()
+
+    @app.websocket("/items/{item_id}/ws")
+    async def websocket_endpoint(
+        *,
+        websocket: WebSocket,
+        session: Annotated[str | None, Cookie()] = None,
+        item_id: str,
+    ):
+        if session is None:
+            raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
+        await websocket.accept()
+        while True:
+            data = await websocket.receive_text()
+            await websocket.send_text(f"Session cookie is: {session}")
+            await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
+    ```
+    """
+
+    def __init__(
+        self,
+        code: Annotated[
+            int,
+            Doc(
+                """
+                A closing code from the
+                [valid codes defined in the specification](https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1).
+                """
+            ),
+        ],
+        reason: Annotated[
+            str | None,
+            Doc(
+                """
+                The reason to close the WebSocket connection.
+
+                It is UTF-8-encoded data. The interpretation of the reason is up to the
+                application, it is not specified by the WebSocket specification.
+
+                It could contain text that could be human-readable or interpretable
+                by the client code, etc.
+                """
+            ),
+        ] = None,
+    ) -> None:
+        super().__init__(code=code, reason=reason)
+
+
+RequestErrorModel: type[BaseModel] = create_model("Request")
+WebSocketErrorModel: type[BaseModel] = create_model("WebSocket")
+
+
+class FastAPIError(RuntimeError):
+    """
+    A generic, FastAPI-specific error.
+    """
+
+
+class DependencyScopeError(FastAPIError):
+    """
+    A dependency declared that it depends on another dependency with an invalid
+    (narrower) scope.
+    """
+
+
+class ValidationException(Exception):
+    def __init__(
+        self,
+        errors: Sequence[Any],
+        *,
+        endpoint_ctx: EndpointContext | None = None,
+    ) -> None:
+        self._errors = errors
+        self.endpoint_ctx = endpoint_ctx
+
+        ctx = endpoint_ctx or {}
+        self.endpoint_function = ctx.get("function")
+        self.endpoint_path = ctx.get("path")
+        self.endpoint_file = ctx.get("file")
+        self.endpoint_line = ctx.get("line")
+
+    def errors(self) -> Sequence[Any]:
+        return self._errors
+
+    def _format_endpoint_context(self) -> str:
+        if not (self.endpoint_file and self.endpoint_line and self.endpoint_function):
+            if self.endpoint_path:
+                return f"\n  Endpoint: {self.endpoint_path}"
+            return ""
+
+        context = f'\n  File "{self.endpoint_file}", line {self.endpoint_line}, in {self.endpoint_function}'
+        if self.endpoint_path:
+            context += f"\n    {self.endpoint_path}"
+        return context
+
+    def __str__(self) -> str:
+        message = f"{len(self._errors)} validation error{'s' if len(self._errors) != 1 else ''}:\n"
+        for err in self._errors:
+            message += f"  {err}\n"
+        message += self._format_endpoint_context()
+        return message.rstrip()
+
+
+class RequestValidationError(ValidationException):
+    def __init__(
+        self,
+        errors: Sequence[Any],
+        *,
+        body: Any = None,
+        endpoint_ctx: EndpointContext | None = None,
+    ) -> None:
+        super().__init__(errors, endpoint_ctx=endpoint_ctx)
+        self.body = body
+
+
+class WebSocketRequestValidationError(ValidationException):
+    def __init__(
+        self,
+        errors: Sequence[Any],
+        *,
+        endpoint_ctx: EndpointContext | None = None,
+    ) -> None:
+        super().__init__(errors, endpoint_ctx=endpoint_ctx)
+
+
+class ResponseValidationError(ValidationException):
+    def __init__(
+        self,
+        errors: Sequence[Any],
+        *,
+        body: Any = None,
+        endpoint_ctx: EndpointContext | None = None,
+    ) -> None:
+        super().__init__(errors, endpoint_ctx=endpoint_ctx)
+        self.body = body
+
+
+class PydanticV1NotSupportedError(FastAPIError):
+    """
+    A pydantic.v1 model is used, which is no longer supported.
+    """
+
+
+class FastAPIDeprecationWarning(UserWarning):
+    """
+    A custom deprecation warning as DeprecationWarning is ignored
+    Ref: https://sethmlarson.dev/deprecations-via-warnings-dont-work-for-python-libraries
+    """