| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186 |
- 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)
|