aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/fastapi/datastructures.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/fastapi/datastructures.py')
-rw-r--r--.venv/lib/python3.12/site-packages/fastapi/datastructures.py204
1 files changed, 204 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/fastapi/datastructures.py b/.venv/lib/python3.12/site-packages/fastapi/datastructures.py
new file mode 100644
index 00000000..cf8406b0
--- /dev/null
+++ b/.venv/lib/python3.12/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