Update 2025-04-13_16:49:18

venv/lib/python3.11/site-packages/fastapi/datastructures.py

@@ -0,0 +1,204 @@
+from typing import (
+    Any,
+    BinaryIO,
+    Callable,
+    Dict,
+    Iterable,
+    Optional,
+    Type,
+    TypeVar,
+    cast,
+)
+
+from fastapi._compat import (
+    PYDANTIC_V2,
+    CoreSchema,
+    GetJsonSchemaHandler,
+    JsonSchemaValue,
+    with_info_plain_validator_function,
+)
+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
+from typing_extensions import Annotated, Doc
+
+
+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[Optional[str], Doc("The original file name.")]
+    size: Annotated[Optional[int], Doc("The size of the file in bytes.")]
+    headers: Annotated[Headers, Doc("The headers of the request.")]
+    content_type: Annotated[
+        Optional[str], 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 __get_validators__(cls: Type["UploadFile"]) -> Iterable[Callable[..., Any]]:
+        yield cls.validate
+
+    @classmethod
+    def validate(cls: Type["UploadFile"], v: Any) -> Any:
+        if not isinstance(v, StarletteUploadFile):
+            raise ValueError(f"Expected UploadFile, received: {type(v)}")
+        return v
+
+    @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)
+
+    if not PYDANTIC_V2:
+
+        @classmethod
+        def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:
+            field_schema.update({"type": "string", "format": "binary"})
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+    ) -> JsonSchemaValue:
+        return {"type": "string", "format": "binary"}
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source: Type[Any], handler: Callable[[Any], CoreSchema]
+    ) -> CoreSchema:
+        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