Coverage for fastapi/datastructures.py: 100%

1from typing import ( 6 ctx1abcdef

2 Any, 

3 BinaryIO, 

4 Callable, 

5 Dict, 

6 Iterable, 

7 Optional, 

8 Type, 

9 TypeVar, 

10 cast, 

11) 

12 

13from fastapi._compat import ( 6 ctx1abcdef

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 6 ctx1abcdef

21from starlette.datastructures import Address as Address # noqa: F401 6 ctx1abcdef

22from starlette.datastructures import FormData as FormData # noqa: F401 6 ctx1abcdef

23from starlette.datastructures import Headers as Headers # noqa: F401 6 ctx1abcdef

24from starlette.datastructures import QueryParams as QueryParams # noqa: F401 6 ctx1abcdef

25from starlette.datastructures import State as State # noqa: F401 6 ctx1abcdef

26from starlette.datastructures import UploadFile as StarletteUploadFile 6 ctx1abcdef

27from typing_extensions import Annotated, Doc 6 ctx1abcdef

28 

29 

30class UploadFile(StarletteUploadFile): 6 ctx1abcdef

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[ 6 ctx1abcdef

65 BinaryIO, 

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

67 ] 

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

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

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

71 content_type: Annotated[ 6 ctx1abcdef

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

73 ] 

74 

75 async def write( 6 ctx1abcdef

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) 6 ctx1mnopqr

94 

95 async def read( 6 ctx1abcdef

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) 6 ctx1mnopqr

112 

113 async def seek( 6 ctx1abcdef

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) 6 ctx1mnopqr

132 

133 async def close(self) -> None: 6 ctx1abcdef

134 """ 

135 Close the file. 

136 

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

138 """ 

139 return await super().close() 6 ctx1mnopqr

140 

141 @classmethod 6 ctx1abcdef

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

143 yield cls.validate 12 ctx1agbhcidjekfl

144 

145 @classmethod 6 ctx1abcdef

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

147 if not isinstance(v, StarletteUploadFile): 54 ctx1s-gtuvwxyz.hABCDEFG/iHIJKLMN:jOPQRSTU;kVWXYZ01=l234567

148 raise ValueError(f"Expected UploadFile, received: {type(v)}") 6 ctx1-./:;=

149 return v 48 ctx1sgtuvwxyzhABCDEFGiHIJKLMNjOPQRSTUkVWXYZ01l234567

150 

151 @classmethod 6 ctx1abcdef

152 def _validate(cls, __input_value: Any, _: Any) -> "UploadFile": 6 ctx1abcdef

153 if not isinstance(__input_value, StarletteUploadFile): 54 ctx1s?gtuvwxyz@hABCDEFG[iHIJKLMN]jOPQRSTU^kVWXYZ01_l234567

154 raise ValueError(f"Expected UploadFile, received: {type(__input_value)}") 6 ctx1?@[]^_

155 return cast(UploadFile, __input_value) 48 ctx1sgtuvwxyzhABCDEFGiHIJKLMNjOPQRSTUkVWXYZ01l234567

156 

157 if not PYDANTIC_V2: 6 ctx1abcdef

158 

159 @classmethod 6 ctx1abcdef

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

161 field_schema.update({"type": "string", "format": "binary"}) 36 ctx2` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbDbEb

162 

163 @classmethod 6 ctx1abcdef

164 def __get_pydantic_json_schema__( 6 ctx1abcdef

165 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

166 ) -> JsonSchemaValue: 

167 return {"type": "string", "format": "binary"} 36 ctx2` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCbDbEb

168 

169 @classmethod 6 ctx1abcdef

170 def __get_pydantic_core_schema__( 6 ctx1abcdef

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

172 ) -> CoreSchema: 

173 return with_info_plain_validator_function(cls._validate) 18 ctx2a Fbg b Gbh c Hbi d Ibj e Jbk f Kbl

174 

175 

176class DefaultPlaceholder: 6 ctx1abcdef

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): 6 ctx1abcdef

185 self.value = value 18 ctx1a89b!#c$%d'(e)*f+,

186 

187 def __bool__(self) -> bool: 6 ctx1abcdef

188 return bool(self.value) 6 ctx18!$')+

189 

190 def __eq__(self, o: object) -> bool: 6 ctx1abcdef

191 return isinstance(o, DefaultPlaceholder) and o.value == self.value 12 ctx29 Lb# Mb% Nb( Ob* Pb, Qb

192 

193 

194DefaultType = TypeVar("DefaultType") 6 ctx1abcdef

195 

196 

197def Default(value: DefaultType) -> DefaultType: 6 ctx1abcdef

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 18 ctx1a89b!#c$%d'(e)*f+,