Coverage for fastapi / datastructures.py: 100%

1from collections.abc import Callable, Mapping 4 ctx1adbc

2from typing import ( 4 ctx1adbc

3 Annotated, 

4 Any, 

5 BinaryIO, 

6 TypeVar, 

7 cast, 

8) 

9 

10from annotated_doc import Doc 4 ctx1adbc

11from pydantic import GetJsonSchemaHandler 4 ctx1adbc

12from starlette.datastructures import URL as URL # noqa: F401 4 ctx1adbc

13from starlette.datastructures import Address as Address # noqa: F401 4 ctx1adbc

14from starlette.datastructures import FormData as FormData # noqa: F401 4 ctx1adbc

15from starlette.datastructures import Headers as Headers # noqa: F401 4 ctx1adbc

16from starlette.datastructures import QueryParams as QueryParams # noqa: F401 4 ctx1adbc

17from starlette.datastructures import State as State # noqa: F401 4 ctx1adbc

18from starlette.datastructures import UploadFile as StarletteUploadFile 4 ctx1adbc

19 

20 

21class UploadFile(StarletteUploadFile): 4 ctx1adbc

22 """ 

23 A file uploaded in a request. 

24 

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

26 

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

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

29 needed for non-async code. 

30 

31 Read more about it in the 

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

33 

34 ## Example 

35 

36 ```python 

37 from typing import Annotated 

38 

39 from fastapi import FastAPI, File, UploadFile 

40 

41 app = FastAPI() 

42 

43 

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

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

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

47 

48 

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

50 async def create_upload_file(file: UploadFile): 

51 return {"filename": file.filename} 

52 ``` 

53 """ 

54 

55 file: Annotated[ 3 ctx1abc

56 BinaryIO, 

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

58 ] 

59 filename: Annotated[str | None, Doc("The original file name.")] 3 ctx1abc

60 size: Annotated[int | None, Doc("The size of the file in bytes.")] 3 ctx1abc

61 headers: Annotated[Headers, Doc("The headers of the request.")] 3 ctx1abc

62 content_type: Annotated[ 3 ctx1abc

63 str | None, Doc("The content type of the request, from the headers.") 

64 ] 

65 

66 async def write( 4 ctx1adbc

67 self, 

68 data: Annotated[ 

69 bytes, 

70 Doc( 

71 """ 

72 The bytes to write to the file. 

73 """ 

74 ), 

75 ], 

76 ) -> None: 

77 """ 

78 Write some bytes to the file. 

79 

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

81 

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

83 """ 

84 return await super().write(data) 3 ctx1efg

85 

86 async def read( 4 ctx1adbc

87 self, 

88 size: Annotated[ 

89 int, 

90 Doc( 

91 """ 

92 The number of bytes to read from the file. 

93 """ 

94 ), 

95 ] = -1, 

96 ) -> bytes: 

97 """ 

98 Read some bytes from the file. 

99 

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

101 """ 

102 return await super().read(size) 3 ctx1efg

103 

104 async def seek( 4 ctx1adbc

105 self, 

106 offset: Annotated[ 

107 int, 

108 Doc( 

109 """ 

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

111 """ 

112 ), 

113 ], 

114 ) -> None: 

115 """ 

116 Move to a position in the file. 

117 

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

119 

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

121 """ 

122 return await super().seek(offset) 3 ctx1efg

123 

124 async def close(self) -> None: 4 ctx1adbc

125 """ 

126 Close the file. 

127 

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

129 """ 

130 return await super().close() 3 ctx1efg

131 

132 @classmethod 4 ctx1adbc

133 def _validate(cls, __input_value: Any, _: Any) -> "UploadFile": 4 ctx1adbc

134 if not isinstance(__input_value, StarletteUploadFile): 75 ctx2U V h W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' ( ) * + , i - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbgbj hbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCb

135 raise ValueError(f"Expected UploadFile, received: {type(__input_value)}") 3 ctx2V , gb

136 return cast(UploadFile, __input_value) 72 ctx2U h W X Y Z 0 1 2 3 4 5 6 7 8 9 ! # $ % ' ( ) * + i - . / : ; = ? @ [ ] ^ _ ` { | } ~ abbbcbdbebfbj hbibjbkblbmbnbobpbqbrbsbtbubvbwbxbybzbAbBbCb

137 

138 @classmethod 4 ctx1adbc

139 def __get_pydantic_json_schema__( 4 ctx1adbc

140 cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler 

141 ) -> dict[str, Any]: 

142 return {"type": "string", "format": "binary"} 30 ctx1klmnopqrstuvwxyzABCDEFGHIJKLMN

143 

144 @classmethod 4 ctx1adbc

145 def __get_pydantic_core_schema__( 4 ctx1adbc

146 cls, source: type[Any], handler: Callable[[Any], Mapping[str, Any]] 

147 ) -> Mapping[str, Any]: 

148 from ._compat.v2 import with_info_plain_validator_function 37 ctx1ahklmnopqrstdiuvwxyzABCDbcjEFGHIJKLMN

149 

150 return with_info_plain_validator_function(cls._validate) 37 ctx1ahklmnopqrstdiuvwxyzABCDbcjEFGHIJKLMN

151 

152 

153class DefaultPlaceholder: 4 ctx1adbc

154 """ 

155 You shouldn't use this class directly. 

156 

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

158 if the overridden default value was truthy. 

159 """ 

160 

161 def __init__(self, value: Any): 4 ctx1adbc

162 self.value = value 10 ctx1aOPdQRbcST

163 

164 def __bool__(self) -> bool: 4 ctx1adbc

165 return bool(self.value) 3 ctx1OQS

166 

167 def __eq__(self, o: object) -> bool: 4 ctx1adbc

168 return isinstance(o, DefaultPlaceholder) and o.value == self.value 6 ctx2P DbR EbT Fb

169 

170 

171DefaultType = TypeVar("DefaultType") 4 ctx1adbc

172 

173 

174def Default(value: DefaultType) -> DefaultType: 4 ctx1adbc

175 """ 

176 You shouldn't use this function directly. 

177 

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

179 if the overridden default value was truthy. 

180 """ 

181 return DefaultPlaceholder(value) # type: ignore 10 ctx1aOPdQRbcST