logo

oasis-root

Compiled tree of Oasis Linux based on own branch at <https://hacktivis.me/git/oasis/> git clone https://anongit.hacktivis.me/git/oasis-root.git

tempfile.py (28379B)


  1. """Temporary files.
  2. This module provides generic, low- and high-level interfaces for
  3. creating temporary files and directories. All of the interfaces
  4. provided by this module can be used without fear of race conditions
  5. except for 'mktemp'. 'mktemp' is subject to race conditions and
  6. should not be used; it is provided for backward compatibility only.
  7. The default path names are returned as str. If you supply bytes as
  8. input, all return values will be in bytes. Ex:
  9. >>> tempfile.mkstemp()
  10. (4, '/tmp/tmptpu9nin8')
  11. >>> tempfile.mkdtemp(suffix=b'')
  12. b'/tmp/tmppbi8f0hy'
  13. This module also provides some data items to the user:
  14. TMP_MAX - maximum number of names that will be tried before
  15. giving up.
  16. tempdir - If this is set to a string before the first use of
  17. any routine from this module, it will be considered as
  18. another candidate location to store temporary files.
  19. """
  20. __all__ = [
  21. "NamedTemporaryFile", "TemporaryFile", # high level safe interfaces
  22. "SpooledTemporaryFile", "TemporaryDirectory",
  23. "mkstemp", "mkdtemp", # low level safe interfaces
  24. "mktemp", # deprecated unsafe interface
  25. "TMP_MAX", "gettempprefix", # constants
  26. "tempdir", "gettempdir",
  27. "gettempprefixb", "gettempdirb",
  28. ]
  29. # Imports.
  30. import functools as _functools
  31. import warnings as _warnings
  32. import io as _io
  33. import os as _os
  34. import shutil as _shutil
  35. import errno as _errno
  36. from random import Random as _Random
  37. import sys as _sys
  38. import types as _types
  39. import weakref as _weakref
  40. import _thread
  41. _allocate_lock = _thread.allocate_lock
  42. _text_openflags = _os.O_RDWR | _os.O_CREAT | _os.O_EXCL
  43. if hasattr(_os, 'O_NOFOLLOW'):
  44. _text_openflags |= _os.O_NOFOLLOW
  45. _bin_openflags = _text_openflags
  46. if hasattr(_os, 'O_BINARY'):
  47. _bin_openflags |= _os.O_BINARY
  48. if hasattr(_os, 'TMP_MAX'):
  49. TMP_MAX = _os.TMP_MAX
  50. else:
  51. TMP_MAX = 10000
  52. # This variable _was_ unused for legacy reasons, see issue 10354.
  53. # But as of 3.5 we actually use it at runtime so changing it would
  54. # have a possibly desirable side effect... But we do not want to support
  55. # that as an API. It is undocumented on purpose. Do not depend on this.
  56. template = "tmp"
  57. # Internal routines.
  58. _once_lock = _allocate_lock()
  59. def _exists(fn):
  60. try:
  61. _os.lstat(fn)
  62. except OSError:
  63. return False
  64. else:
  65. return True
  66. def _infer_return_type(*args):
  67. """Look at the type of all args and divine their implied return type."""
  68. return_type = None
  69. for arg in args:
  70. if arg is None:
  71. continue
  72. if isinstance(arg, bytes):
  73. if return_type is str:
  74. raise TypeError("Can't mix bytes and non-bytes in "
  75. "path components.")
  76. return_type = bytes
  77. else:
  78. if return_type is bytes:
  79. raise TypeError("Can't mix bytes and non-bytes in "
  80. "path components.")
  81. return_type = str
  82. if return_type is None:
  83. if tempdir is None or isinstance(tempdir, str):
  84. return str # tempfile APIs return a str by default.
  85. else:
  86. # we could check for bytes but it'll fail later on anyway
  87. return bytes
  88. return return_type
  89. def _sanitize_params(prefix, suffix, dir):
  90. """Common parameter processing for most APIs in this module."""
  91. output_type = _infer_return_type(prefix, suffix, dir)
  92. if suffix is None:
  93. suffix = output_type()
  94. if prefix is None:
  95. if output_type is str:
  96. prefix = template
  97. else:
  98. prefix = _os.fsencode(template)
  99. if dir is None:
  100. if output_type is str:
  101. dir = gettempdir()
  102. else:
  103. dir = gettempdirb()
  104. return prefix, suffix, dir, output_type
  105. class _RandomNameSequence:
  106. """An instance of _RandomNameSequence generates an endless
  107. sequence of unpredictable strings which can safely be incorporated
  108. into file names. Each string is eight characters long. Multiple
  109. threads can safely use the same instance at the same time.
  110. _RandomNameSequence is an iterator."""
  111. characters = "abcdefghijklmnopqrstuvwxyz0123456789_"
  112. @property
  113. def rng(self):
  114. cur_pid = _os.getpid()
  115. if cur_pid != getattr(self, '_rng_pid', None):
  116. self._rng = _Random()
  117. self._rng_pid = cur_pid
  118. return self._rng
  119. def __iter__(self):
  120. return self
  121. def __next__(self):
  122. return ''.join(self.rng.choices(self.characters, k=8))
  123. def _candidate_tempdir_list():
  124. """Generate a list of candidate temporary directories which
  125. _get_default_tempdir will try."""
  126. dirlist = []
  127. # First, try the environment.
  128. for envname in 'TMPDIR', 'TEMP', 'TMP':
  129. dirname = _os.getenv(envname)
  130. if dirname: dirlist.append(dirname)
  131. # Failing that, try OS-specific locations.
  132. if _os.name == 'nt':
  133. dirlist.extend([ _os.path.expanduser(r'~\AppData\Local\Temp'),
  134. _os.path.expandvars(r'%SYSTEMROOT%\Temp'),
  135. r'c:\temp', r'c:\tmp', r'\temp', r'\tmp' ])
  136. else:
  137. dirlist.extend([ '/tmp', '/var/tmp', '/usr/tmp' ])
  138. # As a last resort, the current directory.
  139. try:
  140. dirlist.append(_os.getcwd())
  141. except (AttributeError, OSError):
  142. dirlist.append(_os.curdir)
  143. return dirlist
  144. def _get_default_tempdir():
  145. """Calculate the default directory to use for temporary files.
  146. This routine should be called exactly once.
  147. We determine whether or not a candidate temp dir is usable by
  148. trying to create and write to a file in that directory. If this
  149. is successful, the test file is deleted. To prevent denial of
  150. service, the name of the test file must be randomized."""
  151. namer = _RandomNameSequence()
  152. dirlist = _candidate_tempdir_list()
  153. for dir in dirlist:
  154. if dir != _os.curdir:
  155. dir = _os.path.abspath(dir)
  156. # Try only a few names per directory.
  157. for seq in range(100):
  158. name = next(namer)
  159. filename = _os.path.join(dir, name)
  160. try:
  161. fd = _os.open(filename, _bin_openflags, 0o600)
  162. try:
  163. try:
  164. with _io.open(fd, 'wb', closefd=False) as fp:
  165. fp.write(b'blat')
  166. finally:
  167. _os.close(fd)
  168. finally:
  169. _os.unlink(filename)
  170. return dir
  171. except FileExistsError:
  172. pass
  173. except PermissionError:
  174. # This exception is thrown when a directory with the chosen name
  175. # already exists on windows.
  176. if (_os.name == 'nt' and _os.path.isdir(dir) and
  177. _os.access(dir, _os.W_OK)):
  178. continue
  179. break # no point trying more names in this directory
  180. except OSError:
  181. break # no point trying more names in this directory
  182. raise FileNotFoundError(_errno.ENOENT,
  183. "No usable temporary directory found in %s" %
  184. dirlist)
  185. _name_sequence = None
  186. def _get_candidate_names():
  187. """Common setup sequence for all user-callable interfaces."""
  188. global _name_sequence
  189. if _name_sequence is None:
  190. _once_lock.acquire()
  191. try:
  192. if _name_sequence is None:
  193. _name_sequence = _RandomNameSequence()
  194. finally:
  195. _once_lock.release()
  196. return _name_sequence
  197. def _mkstemp_inner(dir, pre, suf, flags, output_type):
  198. """Code common to mkstemp, TemporaryFile, and NamedTemporaryFile."""
  199. names = _get_candidate_names()
  200. if output_type is bytes:
  201. names = map(_os.fsencode, names)
  202. for seq in range(TMP_MAX):
  203. name = next(names)
  204. file = _os.path.join(dir, pre + name + suf)
  205. _sys.audit("tempfile.mkstemp", file)
  206. try:
  207. fd = _os.open(file, flags, 0o600)
  208. except FileExistsError:
  209. continue # try again
  210. except PermissionError:
  211. # This exception is thrown when a directory with the chosen name
  212. # already exists on windows.
  213. if (_os.name == 'nt' and _os.path.isdir(dir) and
  214. _os.access(dir, _os.W_OK)):
  215. continue
  216. else:
  217. raise
  218. return (fd, _os.path.abspath(file))
  219. raise FileExistsError(_errno.EEXIST,
  220. "No usable temporary file name found")
  221. # User visible interfaces.
  222. def gettempprefix():
  223. """The default prefix for temporary directories as string."""
  224. return _os.fsdecode(template)
  225. def gettempprefixb():
  226. """The default prefix for temporary directories as bytes."""
  227. return _os.fsencode(template)
  228. tempdir = None
  229. def _gettempdir():
  230. """Private accessor for tempfile.tempdir."""
  231. global tempdir
  232. if tempdir is None:
  233. _once_lock.acquire()
  234. try:
  235. if tempdir is None:
  236. tempdir = _get_default_tempdir()
  237. finally:
  238. _once_lock.release()
  239. return tempdir
  240. def gettempdir():
  241. """Returns tempfile.tempdir as str."""
  242. return _os.fsdecode(_gettempdir())
  243. def gettempdirb():
  244. """Returns tempfile.tempdir as bytes."""
  245. return _os.fsencode(_gettempdir())
  246. def mkstemp(suffix=None, prefix=None, dir=None, text=False):
  247. """User-callable function to create and return a unique temporary
  248. file. The return value is a pair (fd, name) where fd is the
  249. file descriptor returned by os.open, and name is the filename.
  250. If 'suffix' is not None, the file name will end with that suffix,
  251. otherwise there will be no suffix.
  252. If 'prefix' is not None, the file name will begin with that prefix,
  253. otherwise a default prefix is used.
  254. If 'dir' is not None, the file will be created in that directory,
  255. otherwise a default directory is used.
  256. If 'text' is specified and true, the file is opened in text
  257. mode. Else (the default) the file is opened in binary mode.
  258. If any of 'suffix', 'prefix' and 'dir' are not None, they must be the
  259. same type. If they are bytes, the returned name will be bytes; str
  260. otherwise.
  261. The file is readable and writable only by the creating user ID.
  262. If the operating system uses permission bits to indicate whether a
  263. file is executable, the file is executable by no one. The file
  264. descriptor is not inherited by children of this process.
  265. Caller is responsible for deleting the file when done with it.
  266. """
  267. prefix, suffix, dir, output_type = _sanitize_params(prefix, suffix, dir)
  268. if text:
  269. flags = _text_openflags
  270. else:
  271. flags = _bin_openflags
  272. return _mkstemp_inner(dir, prefix, suffix, flags, output_type)
  273. def mkdtemp(suffix=None, prefix=None, dir=None):
  274. """User-callable function to create and return a unique temporary
  275. directory. The return value is the pathname of the directory.
  276. Arguments are as for mkstemp, except that the 'text' argument is
  277. not accepted.
  278. The directory is readable, writable, and searchable only by the
  279. creating user.
  280. Caller is responsible for deleting the directory when done with it.
  281. """
  282. prefix, suffix, dir, output_type = _sanitize_params(prefix, suffix, dir)
  283. names = _get_candidate_names()
  284. if output_type is bytes:
  285. names = map(_os.fsencode, names)
  286. for seq in range(TMP_MAX):
  287. name = next(names)
  288. file = _os.path.join(dir, prefix + name + suffix)
  289. _sys.audit("tempfile.mkdtemp", file)
  290. try:
  291. _os.mkdir(file, 0o700)
  292. except FileExistsError:
  293. continue # try again
  294. except PermissionError:
  295. # This exception is thrown when a directory with the chosen name
  296. # already exists on windows.
  297. if (_os.name == 'nt' and _os.path.isdir(dir) and
  298. _os.access(dir, _os.W_OK)):
  299. continue
  300. else:
  301. raise
  302. return file
  303. raise FileExistsError(_errno.EEXIST,
  304. "No usable temporary directory name found")
  305. def mktemp(suffix="", prefix=template, dir=None):
  306. """User-callable function to return a unique temporary file name. The
  307. file is not created.
  308. Arguments are similar to mkstemp, except that the 'text' argument is
  309. not accepted, and suffix=None, prefix=None and bytes file names are not
  310. supported.
  311. THIS FUNCTION IS UNSAFE AND SHOULD NOT BE USED. The file name may
  312. refer to a file that did not exist at some point, but by the time
  313. you get around to creating it, someone else may have beaten you to
  314. the punch.
  315. """
  316. ## from warnings import warn as _warn
  317. ## _warn("mktemp is a potential security risk to your program",
  318. ## RuntimeWarning, stacklevel=2)
  319. if dir is None:
  320. dir = gettempdir()
  321. names = _get_candidate_names()
  322. for seq in range(TMP_MAX):
  323. name = next(names)
  324. file = _os.path.join(dir, prefix + name + suffix)
  325. if not _exists(file):
  326. return file
  327. raise FileExistsError(_errno.EEXIST,
  328. "No usable temporary filename found")
  329. class _TemporaryFileCloser:
  330. """A separate object allowing proper closing of a temporary file's
  331. underlying file object, without adding a __del__ method to the
  332. temporary file."""
  333. file = None # Set here since __del__ checks it
  334. close_called = False
  335. def __init__(self, file, name, delete=True):
  336. self.file = file
  337. self.name = name
  338. self.delete = delete
  339. # NT provides delete-on-close as a primitive, so we don't need
  340. # the wrapper to do anything special. We still use it so that
  341. # file.name is useful (i.e. not "(fdopen)") with NamedTemporaryFile.
  342. if _os.name != 'nt':
  343. # Cache the unlinker so we don't get spurious errors at
  344. # shutdown when the module-level "os" is None'd out. Note
  345. # that this must be referenced as self.unlink, because the
  346. # name TemporaryFileWrapper may also get None'd out before
  347. # __del__ is called.
  348. def close(self, unlink=_os.unlink):
  349. if not self.close_called and self.file is not None:
  350. self.close_called = True
  351. try:
  352. self.file.close()
  353. finally:
  354. if self.delete:
  355. unlink(self.name)
  356. # Need to ensure the file is deleted on __del__
  357. def __del__(self):
  358. self.close()
  359. else:
  360. def close(self):
  361. if not self.close_called:
  362. self.close_called = True
  363. self.file.close()
  364. class _TemporaryFileWrapper:
  365. """Temporary file wrapper
  366. This class provides a wrapper around files opened for
  367. temporary use. In particular, it seeks to automatically
  368. remove the file when it is no longer needed.
  369. """
  370. def __init__(self, file, name, delete=True):
  371. self.file = file
  372. self.name = name
  373. self.delete = delete
  374. self._closer = _TemporaryFileCloser(file, name, delete)
  375. def __getattr__(self, name):
  376. # Attribute lookups are delegated to the underlying file
  377. # and cached for non-numeric results
  378. # (i.e. methods are cached, closed and friends are not)
  379. file = self.__dict__['file']
  380. a = getattr(file, name)
  381. if hasattr(a, '__call__'):
  382. func = a
  383. @_functools.wraps(func)
  384. def func_wrapper(*args, **kwargs):
  385. return func(*args, **kwargs)
  386. # Avoid closing the file as long as the wrapper is alive,
  387. # see issue #18879.
  388. func_wrapper._closer = self._closer
  389. a = func_wrapper
  390. if not isinstance(a, int):
  391. setattr(self, name, a)
  392. return a
  393. # The underlying __enter__ method returns the wrong object
  394. # (self.file) so override it to return the wrapper
  395. def __enter__(self):
  396. self.file.__enter__()
  397. return self
  398. # Need to trap __exit__ as well to ensure the file gets
  399. # deleted when used in a with statement
  400. def __exit__(self, exc, value, tb):
  401. result = self.file.__exit__(exc, value, tb)
  402. self.close()
  403. return result
  404. def close(self):
  405. """
  406. Close the temporary file, possibly deleting it.
  407. """
  408. self._closer.close()
  409. # iter() doesn't use __getattr__ to find the __iter__ method
  410. def __iter__(self):
  411. # Don't return iter(self.file), but yield from it to avoid closing
  412. # file as long as it's being used as iterator (see issue #23700). We
  413. # can't use 'yield from' here because iter(file) returns the file
  414. # object itself, which has a close method, and thus the file would get
  415. # closed when the generator is finalized, due to PEP380 semantics.
  416. for line in self.file:
  417. yield line
  418. def NamedTemporaryFile(mode='w+b', buffering=-1, encoding=None,
  419. newline=None, suffix=None, prefix=None,
  420. dir=None, delete=True, *, errors=None):
  421. """Create and return a temporary file.
  422. Arguments:
  423. 'prefix', 'suffix', 'dir' -- as for mkstemp.
  424. 'mode' -- the mode argument to io.open (default "w+b").
  425. 'buffering' -- the buffer size argument to io.open (default -1).
  426. 'encoding' -- the encoding argument to io.open (default None)
  427. 'newline' -- the newline argument to io.open (default None)
  428. 'delete' -- whether the file is deleted on close (default True).
  429. 'errors' -- the errors argument to io.open (default None)
  430. The file is created as mkstemp() would do it.
  431. Returns an object with a file-like interface; the name of the file
  432. is accessible as its 'name' attribute. The file will be automatically
  433. deleted when it is closed unless the 'delete' argument is set to False.
  434. """
  435. prefix, suffix, dir, output_type = _sanitize_params(prefix, suffix, dir)
  436. flags = _bin_openflags
  437. # Setting O_TEMPORARY in the flags causes the OS to delete
  438. # the file when it is closed. This is only supported by Windows.
  439. if _os.name == 'nt' and delete:
  440. flags |= _os.O_TEMPORARY
  441. if "b" not in mode:
  442. encoding = _io.text_encoding(encoding)
  443. (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
  444. try:
  445. file = _io.open(fd, mode, buffering=buffering,
  446. newline=newline, encoding=encoding, errors=errors)
  447. return _TemporaryFileWrapper(file, name, delete)
  448. except BaseException:
  449. _os.unlink(name)
  450. _os.close(fd)
  451. raise
  452. if _os.name != 'posix' or _sys.platform == 'cygwin':
  453. # On non-POSIX and Cygwin systems, assume that we cannot unlink a file
  454. # while it is open.
  455. TemporaryFile = NamedTemporaryFile
  456. else:
  457. # Is the O_TMPFILE flag available and does it work?
  458. # The flag is set to False if os.open(dir, os.O_TMPFILE) raises an
  459. # IsADirectoryError exception
  460. _O_TMPFILE_WORKS = hasattr(_os, 'O_TMPFILE')
  461. def TemporaryFile(mode='w+b', buffering=-1, encoding=None,
  462. newline=None, suffix=None, prefix=None,
  463. dir=None, *, errors=None):
  464. """Create and return a temporary file.
  465. Arguments:
  466. 'prefix', 'suffix', 'dir' -- as for mkstemp.
  467. 'mode' -- the mode argument to io.open (default "w+b").
  468. 'buffering' -- the buffer size argument to io.open (default -1).
  469. 'encoding' -- the encoding argument to io.open (default None)
  470. 'newline' -- the newline argument to io.open (default None)
  471. 'errors' -- the errors argument to io.open (default None)
  472. The file is created as mkstemp() would do it.
  473. Returns an object with a file-like interface. The file has no
  474. name, and will cease to exist when it is closed.
  475. """
  476. global _O_TMPFILE_WORKS
  477. if "b" not in mode:
  478. encoding = _io.text_encoding(encoding)
  479. prefix, suffix, dir, output_type = _sanitize_params(prefix, suffix, dir)
  480. flags = _bin_openflags
  481. if _O_TMPFILE_WORKS:
  482. try:
  483. flags2 = (flags | _os.O_TMPFILE) & ~_os.O_CREAT
  484. fd = _os.open(dir, flags2, 0o600)
  485. except IsADirectoryError:
  486. # Linux kernel older than 3.11 ignores the O_TMPFILE flag:
  487. # O_TMPFILE is read as O_DIRECTORY. Trying to open a directory
  488. # with O_RDWR|O_DIRECTORY fails with IsADirectoryError, a
  489. # directory cannot be open to write. Set flag to False to not
  490. # try again.
  491. _O_TMPFILE_WORKS = False
  492. except OSError:
  493. # The filesystem of the directory does not support O_TMPFILE.
  494. # For example, OSError(95, 'Operation not supported').
  495. #
  496. # On Linux kernel older than 3.11, trying to open a regular
  497. # file (or a symbolic link to a regular file) with O_TMPFILE
  498. # fails with NotADirectoryError, because O_TMPFILE is read as
  499. # O_DIRECTORY.
  500. pass
  501. else:
  502. try:
  503. return _io.open(fd, mode, buffering=buffering,
  504. newline=newline, encoding=encoding,
  505. errors=errors)
  506. except:
  507. _os.close(fd)
  508. raise
  509. # Fallback to _mkstemp_inner().
  510. (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
  511. try:
  512. _os.unlink(name)
  513. return _io.open(fd, mode, buffering=buffering,
  514. newline=newline, encoding=encoding, errors=errors)
  515. except:
  516. _os.close(fd)
  517. raise
  518. class SpooledTemporaryFile:
  519. """Temporary file wrapper, specialized to switch from BytesIO
  520. or StringIO to a real file when it exceeds a certain size or
  521. when a fileno is needed.
  522. """
  523. _rolled = False
  524. def __init__(self, max_size=0, mode='w+b', buffering=-1,
  525. encoding=None, newline=None,
  526. suffix=None, prefix=None, dir=None, *, errors=None):
  527. if 'b' in mode:
  528. self._file = _io.BytesIO()
  529. else:
  530. encoding = _io.text_encoding(encoding)
  531. self._file = _io.TextIOWrapper(_io.BytesIO(),
  532. encoding=encoding, errors=errors,
  533. newline=newline)
  534. self._max_size = max_size
  535. self._rolled = False
  536. self._TemporaryFileArgs = {'mode': mode, 'buffering': buffering,
  537. 'suffix': suffix, 'prefix': prefix,
  538. 'encoding': encoding, 'newline': newline,
  539. 'dir': dir, 'errors': errors}
  540. __class_getitem__ = classmethod(_types.GenericAlias)
  541. def _check(self, file):
  542. if self._rolled: return
  543. max_size = self._max_size
  544. if max_size and file.tell() > max_size:
  545. self.rollover()
  546. def rollover(self):
  547. if self._rolled: return
  548. file = self._file
  549. newfile = self._file = TemporaryFile(**self._TemporaryFileArgs)
  550. del self._TemporaryFileArgs
  551. pos = file.tell()
  552. if hasattr(newfile, 'buffer'):
  553. newfile.buffer.write(file.detach().getvalue())
  554. else:
  555. newfile.write(file.getvalue())
  556. newfile.seek(pos, 0)
  557. self._rolled = True
  558. # The method caching trick from NamedTemporaryFile
  559. # won't work here, because _file may change from a
  560. # BytesIO/StringIO instance to a real file. So we list
  561. # all the methods directly.
  562. # Context management protocol
  563. def __enter__(self):
  564. if self._file.closed:
  565. raise ValueError("Cannot enter context with closed file")
  566. return self
  567. def __exit__(self, exc, value, tb):
  568. self._file.close()
  569. # file protocol
  570. def __iter__(self):
  571. return self._file.__iter__()
  572. def close(self):
  573. self._file.close()
  574. @property
  575. def closed(self):
  576. return self._file.closed
  577. @property
  578. def encoding(self):
  579. return self._file.encoding
  580. @property
  581. def errors(self):
  582. return self._file.errors
  583. def fileno(self):
  584. self.rollover()
  585. return self._file.fileno()
  586. def flush(self):
  587. self._file.flush()
  588. def isatty(self):
  589. return self._file.isatty()
  590. @property
  591. def mode(self):
  592. try:
  593. return self._file.mode
  594. except AttributeError:
  595. return self._TemporaryFileArgs['mode']
  596. @property
  597. def name(self):
  598. try:
  599. return self._file.name
  600. except AttributeError:
  601. return None
  602. @property
  603. def newlines(self):
  604. return self._file.newlines
  605. def read(self, *args):
  606. return self._file.read(*args)
  607. def readline(self, *args):
  608. return self._file.readline(*args)
  609. def readlines(self, *args):
  610. return self._file.readlines(*args)
  611. def seek(self, *args):
  612. return self._file.seek(*args)
  613. def tell(self):
  614. return self._file.tell()
  615. def truncate(self, size=None):
  616. if size is None:
  617. self._file.truncate()
  618. else:
  619. if size > self._max_size:
  620. self.rollover()
  621. self._file.truncate(size)
  622. def write(self, s):
  623. file = self._file
  624. rv = file.write(s)
  625. self._check(file)
  626. return rv
  627. def writelines(self, iterable):
  628. file = self._file
  629. rv = file.writelines(iterable)
  630. self._check(file)
  631. return rv
  632. class TemporaryDirectory:
  633. """Create and return a temporary directory. This has the same
  634. behavior as mkdtemp but can be used as a context manager. For
  635. example:
  636. with TemporaryDirectory() as tmpdir:
  637. ...
  638. Upon exiting the context, the directory and everything contained
  639. in it are removed.
  640. """
  641. def __init__(self, suffix=None, prefix=None, dir=None,
  642. ignore_cleanup_errors=False):
  643. self.name = mkdtemp(suffix, prefix, dir)
  644. self._ignore_cleanup_errors = ignore_cleanup_errors
  645. self._finalizer = _weakref.finalize(
  646. self, self._cleanup, self.name,
  647. warn_message="Implicitly cleaning up {!r}".format(self),
  648. ignore_errors=self._ignore_cleanup_errors)
  649. @classmethod
  650. def _rmtree(cls, name, ignore_errors=False):
  651. def onerror(func, path, exc_info):
  652. if issubclass(exc_info[0], PermissionError):
  653. def resetperms(path):
  654. try:
  655. _os.chflags(path, 0)
  656. except AttributeError:
  657. pass
  658. _os.chmod(path, 0o700)
  659. try:
  660. if path != name:
  661. resetperms(_os.path.dirname(path))
  662. resetperms(path)
  663. try:
  664. _os.unlink(path)
  665. # PermissionError is raised on FreeBSD for directories
  666. except (IsADirectoryError, PermissionError):
  667. cls._rmtree(path, ignore_errors=ignore_errors)
  668. except FileNotFoundError:
  669. pass
  670. elif issubclass(exc_info[0], FileNotFoundError):
  671. pass
  672. else:
  673. if not ignore_errors:
  674. raise
  675. _shutil.rmtree(name, onerror=onerror)
  676. @classmethod
  677. def _cleanup(cls, name, warn_message, ignore_errors=False):
  678. cls._rmtree(name, ignore_errors=ignore_errors)
  679. _warnings.warn(warn_message, ResourceWarning)
  680. def __repr__(self):
  681. return "<{} {!r}>".format(self.__class__.__name__, self.name)
  682. def __enter__(self):
  683. return self.name
  684. def __exit__(self, exc, value, tb):
  685. self.cleanup()
  686. def cleanup(self):
  687. if self._finalizer.detach() or _os.path.exists(self.name):
  688. self._rmtree(self.name, ignore_errors=self._ignore_cleanup_errors)
  689. __class_getitem__ = classmethod(_types.GenericAlias)