Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cachecontrol/caches/file_cache.py @ 5:9b1c78e6ba9c draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
comparison
equal deleted inserted replaced
4:79f47841a781 5:9b1c78e6ba9c
1 import hashlib
2 import os
3
4 from lockfile import LockFile
5 from lockfile.mkdirlockfile import MkdirLockFile
6
7 from ..cache import BaseCache
8 from ..controller import CacheController
9
10
11 def _secure_open_write(filename, fmode):
12 # We only want to write to this file, so open it in write only mode
13 flags = os.O_WRONLY
14
15 # os.O_CREAT | os.O_EXCL will fail if the file already exists, so we only
16 # will open *new* files.
17 # We specify this because we want to ensure that the mode we pass is the
18 # mode of the file.
19 flags |= os.O_CREAT | os.O_EXCL
20
21 # Do not follow symlinks to prevent someone from making a symlink that
22 # we follow and insecurely open a cache file.
23 if hasattr(os, "O_NOFOLLOW"):
24 flags |= os.O_NOFOLLOW
25
26 # On Windows we'll mark this file as binary
27 if hasattr(os, "O_BINARY"):
28 flags |= os.O_BINARY
29
30 # Before we open our file, we want to delete any existing file that is
31 # there
32 try:
33 os.remove(filename)
34 except (IOError, OSError):
35 # The file must not exist already, so we can just skip ahead to opening
36 pass
37
38 # Open our file, the use of os.O_CREAT | os.O_EXCL will ensure that if a
39 # race condition happens between the os.remove and this line, that an
40 # error will be raised. Because we utilize a lockfile this should only
41 # happen if someone is attempting to attack us.
42 fd = os.open(filename, flags, fmode)
43 try:
44 return os.fdopen(fd, "wb")
45 except:
46 # An error occurred wrapping our FD in a file object
47 os.close(fd)
48 raise
49
50
51 class FileCache(BaseCache):
52 def __init__(self, directory, forever=False, filemode=0o0600,
53 dirmode=0o0700, use_dir_lock=None, lock_class=None):
54
55 if use_dir_lock is not None and lock_class is not None:
56 raise ValueError("Cannot use use_dir_lock and lock_class together")
57
58 if use_dir_lock:
59 lock_class = MkdirLockFile
60
61 if lock_class is None:
62 lock_class = LockFile
63
64 self.directory = directory
65 self.forever = forever
66 self.filemode = filemode
67 self.dirmode = dirmode
68 self.lock_class = lock_class
69
70
71 @staticmethod
72 def encode(x):
73 return hashlib.sha224(x.encode()).hexdigest()
74
75 def _fn(self, name):
76 # NOTE: This method should not change as some may depend on it.
77 # See: https://github.com/ionrock/cachecontrol/issues/63
78 hashed = self.encode(name)
79 parts = list(hashed[:5]) + [hashed]
80 return os.path.join(self.directory, *parts)
81
82 def get(self, key):
83 name = self._fn(key)
84 if not os.path.exists(name):
85 return None
86
87 with open(name, 'rb') as fh:
88 return fh.read()
89
90 def set(self, key, value):
91 name = self._fn(key)
92
93 # Make sure the directory exists
94 try:
95 os.makedirs(os.path.dirname(name), self.dirmode)
96 except (IOError, OSError):
97 pass
98
99 with self.lock_class(name) as lock:
100 # Write our actual file
101 with _secure_open_write(lock.path, self.filemode) as fh:
102 fh.write(value)
103
104 def delete(self, key):
105 name = self._fn(key)
106 if not self.forever:
107 os.remove(name)
108
109
110 def url_to_file_path(url, filecache):
111 """Return the file cache path based on the URL.
112
113 This does not ensure the file exists!
114 """
115 key = CacheController.cache_url(url)
116 return filecache._fn(key)