Coverage for fastapi/datastructures.py: 100%

1from typing import ( 7 ctx1abcdefg

2 Any, 

3 BinaryIO, 

4 Callable, 

5 Dict, 

6 Iterable, 

7 Optional, 

8 Type, 

9 TypeVar, 

10 cast, 

11) 

12 

13from annotated_doc import Doc 7 ctx1abcdefg

14from fastapi._compat import ( 7 ctx1abcdefg

15 CoreSchema, 

16 GetJsonSchemaHandler, 

17 JsonSchemaValue, 

18) 

19from starlette.datastructures import URL as URL # noqa: F401 7 ctx1abcdefg

20from starlette.datastructures import Address as Address # noqa: F401 7 ctx1abcdefg

21from starlette.datastructures import FormData as FormData # noqa: F401 7 ctx1abcdefg

22from starlette.datastructures import Headers as Headers # noqa: F401 7 ctx1abcdefg

23from starlette.datastructures import QueryParams as QueryParams # noqa: F401 7 ctx1abcdefg

24from starlette.datastructures import State as State # noqa: F401 7 ctx1abcdefg

25from starlette.datastructures import UploadFile as StarletteUploadFile 7 ctx1abcdefg

26from typing_extensions import Annotated 7 ctx1abcdefg

27 

28 

29class UploadFile(StarletteUploadFile): 7 ctx1abcdefg

30 """ 

31 A file uploaded in a request. 

32 

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

34 

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

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

37 needed for non-async code. 

38 

39 Read more about it in the 

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

41 

42 ## Example 

43 

44 ```python 

45 from typing import Annotated 

46 

47 from fastapi import FastAPI, File, UploadFile 

48 

49 app = FastAPI() 

50 

51 

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

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

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

55 

56 

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

58 async def create_upload_file(file: UploadFile): 

59 return {"filename": file.filename} 

60 ``` 

61 """ 

62 

63 file: Annotated[ 6 ctx1abcdef

64 BinaryIO, 

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

66 ] 

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

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

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

70 content_type: Annotated[ 6 ctx1abcdef

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

72 ] 

73 

74 async def write( 7 ctx1abcdefg

75 self, 

76 data: Annotated[ 

77 bytes, 

78 Doc( 

79 """ 

80 The bytes to write to the file. 

81 """ 

82 ), 

83 ], 

84 ) -> None: 

85 """ 

86 Write some bytes to the file. 

87 

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

89 

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

91 """ 

92 return await super().write(data) 7 ctx1nopqrst

93 

94 async def read( 7 ctx1abcdefg

95 self, 

96 size: Annotated[ 

97 int, 

98 Doc( 

99 """ 

100 The number of bytes to read from the file. 

101 """ 

102 ), 

103 ] = -1, 

104 ) -> bytes: 

105 """ 

106 Read some bytes from the file. 

107 

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

109 """ 

110 return await super().read(size) 7 ctx1nopqrst

111 

112 async def seek( 7 ctx1abcdefg

113 self, 

114 offset: Annotated[ 

115 int, 

116 Doc( 

117 """ 

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

119 """ 

120 ), 

121 ], 

122 ) -> None: 

123 """ 

124 Move to a position in the file. 

125 

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

127 

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

129 """ 

130 return await super().seek(offset) 7 ctx1nopqrst

131 

132 async def close(self) -> None: 7 ctx1abcdefg

133 """ 

134 Close the file. 

135 

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

137 """ 

138 return await super().close() 7 ctx1nopqrst

139 

140 @classmethod 7 ctx1abcdefg

141 def __get_validators__(cls: Type["UploadFile"]) -> Iterable[Callable[..., Any]]: 7 ctx1abcdefg

142 yield cls.validate 12 ctx1ahbicjdkelfm

143 

144 @classmethod 7 ctx1abcdefg

145 def validate(cls: Type["UploadFile"], v: Any) -> Any: 7 ctx1abcdefg

146 if not isinstance(v, StarletteUploadFile): 55 ctx2u Ebh v w x y z A B Fbi C D E F G H I Gbj J K L M N O P Hbk Q R S T U V W Ibl X Y Z 0 1 2 3 Jbm 4 5 6 7 8 9 Kb

147 raise ValueError(f"Expected UploadFile, received: {type(v)}") 7 ctx2EbFbGbHbIbJbKb

148 return v 48 ctx1uhvwxyzABiCDEFGHIjJKLMNOPkQRSTUVWlXYZ0123m456789

149 

150 @classmethod 7 ctx1abcdefg

151 def _validate(cls, __input_value: Any, _: Any) -> "UploadFile": 7 ctx1abcdefg

152 if not isinstance(__input_value, StarletteUploadFile): 63 ctx2u Lbh v w x y z A B Mbi C D E F G H I Nbj J K L M N O P Obk Q R S T U V W Pbl X Y Z 0 1 2 3 Qbm 4 5 6 7 8 9 RbSb! TbUbVbWbXbYb

153 raise ValueError(f"Expected UploadFile, received: {type(__input_value)}") 7 ctx2LbMbNbObPbQbSb

154 return cast(UploadFile, __input_value) 56 ctx2u h v w x y z A B i C D E F G H I j J K L M N O P k Q R S T U V W l X Y Z 0 1 2 3 m 4 5 6 7 8 9 Rb! TbUbVbWbXbYb

155 

156 # TODO: remove when deprecating Pydantic v1 

157 @classmethod 7 ctx1abcdefg

158 def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None: 7 ctx1abcdefg

159 field_schema.update({"type": "string", "format": "binary"}) 36 ctx2# $ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjb

160 

161 @classmethod 7 ctx1abcdefg

162 def __get_pydantic_json_schema__( 7 ctx1abcdefg

163 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

164 ) -> JsonSchemaValue: 

165 return {"type": "string", "format": "binary"} 42 ctx2# $ % ' ( ) * + , - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbhbibjbkblbmbnbobpb

166 

167 @classmethod 7 ctx1abcdefg

168 def __get_pydantic_core_schema__( 7 ctx1abcdefg

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

170 ) -> CoreSchema: 

171 from ._compat.v2 import with_info_plain_validator_function 56 ctx2a h # $ % ' ( ) b i * + , - . / c j : ; = ? @ [ d k ] ^ _ ` { | e l } ~ abbbcbdbf m ebfbgbhbibjbg ! kblbmbnbobpb

172 

173 return with_info_plain_validator_function(cls._validate) 56 ctx2a h # $ % ' ( ) b i * + , - . / c j : ; = ? @ [ d k ] ^ _ ` { | e l } ~ abbbcbdbf m ebfbgbhbibjbg ! kblbmbnbobpb

174 

175 

176class DefaultPlaceholder: 7 ctx1abcdefg

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): 7 ctx1abcdefg

185 self.value = value 21 ctx2a qbrbb sbtbc ubvbd wbxbe ybzbf AbBbg CbDb

186 

187 def __bool__(self) -> bool: 7 ctx1abcdefg

188 return bool(self.value) 7 ctx2qbsbubwbybAbCb

189 

190 def __eq__(self, o: object) -> bool: 7 ctx1abcdefg

191 return isinstance(o, DefaultPlaceholder) and o.value == self.value 14 ctx2rbZbtb0bvb1bxb2bzb3bBb4bDb5b

192 

193 

194DefaultType = TypeVar("DefaultType") 7 ctx1abcdefg

195 

196 

197def Default(value: DefaultType) -> DefaultType: 7 ctx1abcdefg

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 21 ctx2a qbrbb sbtbc ubvbd wbxbe ybzbf AbBbg CbDb