1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
"""
fs.tempfs
=========
Make a temporary file system that exists in a folder provided by the OS. All files contained in a TempFS are removed when the `close` method is called (or when the TempFS is cleaned up by Python).
"""
import os
import time
import tempfile
from fs.osfs import OSFS
from fs.errors import *
from fs import _thread_synchronize_default
class TempFS(OSFS):
"""Create a Filesystem in a tempory directory (with tempfile.mkdtemp),
and removes it when the TempFS object is cleaned up."""
def __init__(self, identifier=None, temp_dir=None, dir_mode=0700, thread_synchronize=_thread_synchronize_default):
"""Creates a temporary Filesystem
identifier -- A string that is included in the name of the temporary directory,
default uses "TempFS"
"""
self._temp_dir = tempfile.mkdtemp(identifier or "TempFS",dir=temp_dir)
self._cleaned = False
OSFS.__init__(self, self._temp_dir, dir_mode=dir_mode, thread_synchronize=thread_synchronize)
def __str__(self):
return '<TempFS: %s>' % self._temp_dir
__repr__ = __str__
def __unicode__(self):
return u'<TempFS: %s>' % self._temp_dir
def close(self):
"""Removes the temporary directory.
This will be called automatically when the object is cleaned up by
Python, although it is advisable to call it manually.
Note that once this method has been called, the FS object may
no longer be used.
"""
# Depending on how resources are freed by the OS, there could
# be some transient errors when freeing a TempFS soon after it
# was used. If they occur, do a small sleep and try again.
try:
self._close()
except (ResourceLockedError,ResourceInvalidError):
time.sleep(0.5)
self._close()
@convert_os_errors
def _close(self):
"""Actual implementation of close().
This is a separate method so it can be re-tried in the face of
transient errors.
"""
os_remove = convert_os_errors(os.remove)
os_rmdir = convert_os_errors(os.rmdir)
if not self._cleaned and self.exists("/"):
self._lock.acquire()
try:
# shutil.rmtree doesn't handle long paths on win32,
# so we walk the tree by hand.
entries = os.walk(self.root_path,topdown=False)
for (dir,dirnames,filenames) in entries:
for filename in filenames:
try:
os_remove(os.path.join(dir,filename))
except ResourceNotFoundError:
pass
for dirname in dirnames:
try:
os_rmdir(os.path.join(dir,dirname))
except ResourceNotFoundError:
pass
os.rmdir(self.root_path)
self._cleaned = True
finally:
self._lock.release()
super(TempFS,self).close()
|
