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