Coverage for fastapi/datastructures.py: 100%

1from typing import ( 5 ctx1abcde

2 Any, 

3 BinaryIO, 

4 Callable, 

5 Dict, 

6 Iterable, 

7 Optional, 

8 Type, 

9 TypeVar, 

10 cast, 

11) 

12 

13from fastapi._compat import ( 5 ctx1abcde

14 PYDANTIC_V2, 

15 CoreSchema, 

16 GetJsonSchemaHandler, 

17 JsonSchemaValue, 

18 with_info_plain_validator_function, 

19) 

20from starlette.datastructures import URL as URL # noqa: F401 5 ctx1abcde

21from starlette.datastructures import Address as Address # noqa: F401 5 ctx1abcde

22from starlette.datastructures import FormData as FormData # noqa: F401 5 ctx1abcde

23from starlette.datastructures import Headers as Headers # noqa: F401 5 ctx1abcde

24from starlette.datastructures import QueryParams as QueryParams # noqa: F401 5 ctx1abcde

25from starlette.datastructures import State as State # noqa: F401 5 ctx1abcde

26from starlette.datastructures import UploadFile as StarletteUploadFile 5 ctx1abcde

27from typing_extensions import Annotated, Doc 5 ctx1abcde

28 

29 

30class UploadFile(StarletteUploadFile): 5 ctx1abcde

31 """ 

32 A file uploaded in a request. 

33 

34 Define it as a *path operation function* (or dependency) parameter. 

35 

36 If you are using a regular `def` function, you can use the `upload_file.file` 

37 attribute to access the raw standard Python file (blocking, not async), useful and 

38 needed for non-async code. 

39 

40 Read more about it in the 

41 [FastAPI docs for Request Files](https://fastapi.tiangolo.com/tutorial/request-files/). 

42 

43 ## Example 

44 

45 ```python 

46 from typing import Annotated 

47 

48 from fastapi import FastAPI, File, UploadFile 

49 

50 app = FastAPI() 

51 

52 

53 @app.post("/files/") 

54 async def create_file(file: Annotated[bytes, File()]): 

55 return {"file_size": len(file)} 

56 

57 

58 @app.post("/uploadfile/") 

59 async def create_upload_file(file: UploadFile): 

60 return {"filename": file.filename} 

61 ``` 

62 """ 

63 

64 file: Annotated[ 5 ctx1abcde

65 BinaryIO, 

66 Doc("The standard Python file object (non-async)."), 

67 ] 

68 filename: Annotated[Optional[str], Doc("The original file name.")] 5 ctx1abcde

69 size: Annotated[Optional[int], Doc("The size of the file in bytes.")] 5 ctx1abcde

70 headers: Annotated[Headers, Doc("The headers of the request.")] 5 ctx1abcde

71 content_type: Annotated[ 5 ctx1abcde

72 Optional[str], Doc("The content type of the request, from the headers.") 

73 ] 

74 

75 async def write( 5 ctx1abcde

76 self, 

77 data: Annotated[ 

78 bytes, 

79 Doc( 

80 """ 

81 The bytes to write to the file. 

82 """ 

83 ), 

84 ], 

85 ) -> None: 

86 """ 

87 Write some bytes to the file. 

88 

89 You normally wouldn't use this from a file you read in a request. 

90 

91 To be awaitable, compatible with async, this is run in threadpool. 

92 """ 

93 return await super().write(data) 5 ctx1abcde

94 

95 async def read( 5 ctx1abcde

96 self, 

97 size: Annotated[ 

98 int, 

99 Doc( 

100 """ 

101 The number of bytes to read from the file. 

102 """ 

103 ), 

104 ] = -1, 

105 ) -> bytes: 

106 """ 

107 Read some bytes from the file. 

108 

109 To be awaitable, compatible with async, this is run in threadpool. 

110 """ 

111 return await super().read(size) 5 ctx1abcde

112 

113 async def seek( 5 ctx1abcde

114 self, 

115 offset: Annotated[ 

116 int, 

117 Doc( 

118 """ 

119 The position in bytes to seek to in the file. 

120 """ 

121 ), 

122 ], 

123 ) -> None: 

124 """ 

125 Move to a position in the file. 

126 

127 Any next read or write will be done from that position. 

128 

129 To be awaitable, compatible with async, this is run in threadpool. 

130 """ 

131 return await super().seek(offset) 5 ctx1abcde

132 

133 async def close(self) -> None: 5 ctx1abcde

134 """ 

135 Close the file. 

136 

137 To be awaitable, compatible with async, this is run in threadpool. 

138 """ 

139 return await super().close() 5 ctx1abcde

140 

141 @classmethod 5 ctx1abcde

142 def __get_validators__(cls: Type["UploadFile"]) -> Iterable[Callable[..., Any]]: 5 ctx1abcde

143 yield cls.validate 5 ctx1abcde

144 

145 @classmethod 5 ctx1abcde

146 def validate(cls: Type["UploadFile"], v: Any) -> Any: 5 ctx1abcde

147 if not isinstance(v, StarletteUploadFile): 5 ctx1abcde

148 raise ValueError(f"Expected UploadFile, received: {type(v)}") 5 ctx1abcde

149 return v 5 ctx1abcde

150 

151 @classmethod 5 ctx1abcde

152 def _validate(cls, __input_value: Any, _: Any) -> "UploadFile": 5 ctx1abcde

153 if not isinstance(__input_value, StarletteUploadFile): 5 ctx1abcde

154 raise ValueError(f"Expected UploadFile, received: {type(__input_value)}") 5 ctx1abcde

155 return cast(UploadFile, __input_value) 5 ctx1abcde

156 

157 if not PYDANTIC_V2: 5 ctx1abcde

158 

159 @classmethod 5 ctx1abcde

160 def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None: 5 ctx1abcde

161 field_schema.update({"type": "string", "format": "binary"}) 5 ctx1abcde

162 

163 @classmethod 5 ctx1abcde

164 def __get_pydantic_json_schema__( 5 ctx1abcde

165 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

166 ) -> JsonSchemaValue: 

167 return {"type": "string", "format": "binary"} 5 ctx1abcde

168 

169 @classmethod 5 ctx1abcde

170 def __get_pydantic_core_schema__( 5 ctx1abcde

171 cls, source: Type[Any], handler: Callable[[Any], CoreSchema] 

172 ) -> CoreSchema: 

173 return with_info_plain_validator_function(cls._validate) 5 ctx1abcde

174 

175 

176class DefaultPlaceholder: 5 ctx1abcde

177 """ 

178 You shouldn't use this class directly. 

179 

180 It's used internally to recognize when a default value has been overwritten, even 

181 if the overridden default value was truthy. 

182 """ 

183 

184 def __init__(self, value: Any): 5 ctx1abcde

185 self.value = value 5 ctx1abcde

186 

187 def __bool__(self) -> bool: 5 ctx1abcde

188 return bool(self.value) 5 ctx1abcde

189 

190 def __eq__(self, o: object) -> bool: 5 ctx1abcde

191 return isinstance(o, DefaultPlaceholder) and o.value == self.value 5 ctx1abcde

192 

193 

194DefaultType = TypeVar("DefaultType") 5 ctx1abcde

195 

196 

197def Default(value: DefaultType) -> DefaultType: 5 ctx1abcde

198 """ 

199 You shouldn't use this function directly. 

200 

201 It's used internally to recognize when a default value has been overwritten, even 

202 if the overridden default value was truthy. 

203 """ 

204 return DefaultPlaceholder(value) # type: ignore 5 ctx1abcde