Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/datastructures/file_storage.py: 35%

1from __future__ import annotations 

2 

3import collections.abc as cabc 

4import mimetypes 

5import os 

6import typing as t 

7from io import BytesIO 

8from os import fsdecode 

9from os import fspath 

10 

11from .._header_property import header_property 

12from .._internal import _plain_int 

13from ..http import parse_options_header 

14from ..utils import cached_property 

15from .headers import Headers 

16from .structures import MultiDict 

17 

18 

19class FileStorage: 

20 """A parsed form data file part in :attr:`.Request.files`.""" 

21 

22 def __init__( 

23 self, 

24 stream: t.IO[bytes] | None = None, 

25 filename: str | None = None, 

26 name: str | None = None, 

27 content_type: str | None = None, 

28 content_length: int | None = None, 

29 headers: Headers | None = None, 

30 ): 

31 self.name = name 

32 """The name of the field in the form data, corresponding to a key in 

33 :attr:`.Request.files`. 

34 """ 

35 

36 self.stream = stream or BytesIO() 

37 """The wrapped file object. Attribute access on ``FileStorage`` is 

38 forwarded to this. For example, ``fs.read()`` is equivalent to 

39 ``fs.stream.read()``. 

40 """ 

41 

42 self.filename = _guess_filename(self.stream, filename) 

43 """The name associated with the file data in the form. This is only a 

44 name, not a path on the filesystem of the client or server. 

45 """ 

46 

47 if headers is None: 

48 headers = Headers() 

49 

50 self.headers = headers 

51 """Any headers associated with the file data in the form. 

52 

53 .. versionadded:: 0.6 

54 """ 

55 

56 if content_type is not None: 

57 headers["Content-Type"] = content_type 

58 

59 if content_length is not None: 

60 headers["Content-Length"] = str(content_length) 

61 

62 content_type = header_property[str | None]( 

63 "Content-Type", 

64 read_only=True, 

65 doc="""The ``Content-Type`` header. The type of data in the file, with 

66 optional parameters for additional detail. 

67 

68 A ``str``, or ``None`` if not set. 

69 

70 :attr:`mimetype` and :attr:`mimetype_params` allow working with the two 

71 parts of the value separately. 

72 """, 

73 ) 

74 

75 content_length = header_property[int | None]( 

76 "Content-Length", 

77 load_func=_plain_int, 

78 read_only=True, 

79 doc="""The ``Content-Length`` header. The size of the file in bytes. 

80 

81 An ``int``, or ``None`` if not set. 

82 """, 

83 ) 

84 

85 _parsed_content_type: tuple[str, dict[str, str]] | None = None 

86 

87 def _parse_content_type(self) -> None: 

88 if self._parsed_content_type is None: 

89 self._parsed_content_type = parse_options_header(self.content_type) 

90 

91 @cached_property 

92 def mimetype(self) -> str: 

93 """The value from :attr:`content_type`, lowercase. For example, 

94 ``text/HTML; charset=utf-8`` becomes ``text/html``. 

95 """ 

96 self._parse_content_type() 

97 return self._parsed_content_type[0].lower() # type: ignore[index] 

98 

99 @cached_property 

100 def mimetype_params(self) -> cabc.Mapping[str, str]: 

101 """The parameters from :attr:`content_type``. For example, 

102 ``text/html; charset=utf-8`` becomes ``{"charset": "utf-8"}``. 

103 """ 

104 self._parse_content_type() 

105 return self._parsed_content_type[1] # type: ignore[index] 

106 

107 def save( 

108 self, dst: str | os.PathLike[str] | t.IO[bytes], buffer_size: int = 0 

109 ) -> None: 

110 """Save the file to a path or file object. 

111 

112 For secure file saving also have a look at :func:`secure_filename`. 

113 

114 :param dst: The location to write to. A path, or a file object 

115 in ``wb`` mode. 

116 :param buffer_size: Passed as the ``length`` parameter of 

117 :func:`shutil.copyfileobj`. 

118 

119 .. versionchanged:: 1.0 

120 Supports :mod:`pathlib`. 

121 """ 

122 from shutil import copyfileobj 

123 

124 close_dst = False 

125 

126 if hasattr(dst, "__fspath__"): 

127 dst = fspath(dst) 

128 

129 if isinstance(dst, str): 

130 dst = open(dst, "wb") 

131 close_dst = True 

132 

133 try: 

134 copyfileobj(self.stream, dst, buffer_size) 

135 finally: 

136 if close_dst: 

137 dst.close() 

138 

139 def close(self) -> None: 

140 """Close the underlying file if possible.""" 

141 try: 

142 self.stream.close() 

143 except Exception: 

144 pass 

145 

146 def __bool__(self) -> bool: 

147 return bool(self.filename) 

148 

149 def __getattr__(self, name: str) -> t.Any: 

150 return getattr(self.stream, name) 

151 

152 def __iter__(self) -> cabc.Iterator[bytes]: 

153 return iter(self.stream) 

154 

155 def __repr__(self) -> str: 

156 return f"<{type(self).__name__}: {self.filename!r} ({self.content_type!r})>" 

157 

158 

159class FileMultiDict(MultiDict[str, FileStorage]): 

160 """A :class:`.MultiDict` for managing form data file values. Used by 

161 :class:`.EnvironBuilder` for tests. 

162 

163 .. versionadded:: 0.5 

164 """ 

165 

166 def add_file( 

167 self, 

168 name: str, 

169 file: str | os.PathLike[str] | t.IO[bytes] | FileStorage, 

170 filename: str | None = None, 

171 content_type: str | None = None, 

172 ) -> None: 

173 """Add a file to the given key. Can be passed a filename or IO object, 

174 which will construct a :class:`.FileStorage` object. 

175 

176 :param name: The key to add the file to. 

177 :param file: The file to add. Constructs a :class:`FileStorage` object 

178 if the value is not one. 

179 :param filename: The filename to set for the field. Defaults to ``file`` 

180 if it's a filename or ``file.name`` if it's an IO object. 

181 :param content_type: The content type to set for the field. Defaults to 

182 guessing based on the filename, falling back to 

183 ``application/octet-stream``. 

184 

185 .. versionchanged:: 3.2 

186 The filename is detected from an IO object. 

187 """ 

188 if isinstance(file, FileStorage): 

189 self.add(name, file) 

190 return 

191 

192 if isinstance(file, (str, os.PathLike)): 

193 if filename is None: 

194 filename = os.fspath(file) 

195 

196 file_obj: t.IO[bytes] = open(file, "rb") 

197 else: 

198 file_obj = file # type: ignore[assignment] 

199 filename = _guess_filename(file_obj, filename) 

200 

201 if filename is not None and content_type is None: 

202 content_type = ( 

203 mimetypes.guess_type(filename)[0] or "application/octet-stream" 

204 ) 

205 

206 self.add(name, FileStorage(file_obj, filename, name, content_type)) 

207 

208 def close(self) -> None: 

209 """Call :meth:`~FileStorage.close` on every open file. 

210 

211 .. versionadded:: 3.2 

212 """ 

213 for values in self.listvalues(): 

214 for value in values: 

215 if not value.closed: 

216 value.close() 

217 

218 def clear(self) -> None: 

219 """Call :meth:`close`, then remove all items. 

220 

221 .. versionadded:: 3.2 

222 """ 

223 self.close() 

224 super().clear() 

225 

226 

227def _guess_filename(stream: t.IO[t.Any], filename: str | None) -> str | None: 

228 if filename is not None: 

229 return fsdecode(filename) 

230 

231 filename = getattr(stream, "name", None) 

232 

233 if filename is not None: 

234 filename = fsdecode(filename) 

235 

236 # Python names special streams like `<stderr>`, ignore these. 

237 if filename.startswith("<") and filename.endswith(">"): 

238 filename = None 

239 

240 return filename