Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def write_dict(filename, contents):
try:
fp = open(filename, 'w')
except IOError:
logging.error('Unable to write to file %s' % filename)
return
portalocker.lock(fp, portalocker.LOCK_EX)
fp.write('# coding: utf8\n{\n')
for key in sorted(contents):
fp.write('%s: %s,\n' % (utf8_repr(key), utf8_repr(contents[key])))
fp.write('}\n')
portalocker.unlock(fp)
fp.close()
def _unlock(self, response):
if response and response.session_file and response.session_locked:
try:
portalocker.unlock(response.session_file)
response.session_locked = False
except: ### this should never happen but happens in Windows
pass
def _close_shelve_and_unlock(self):
try:
if self.storage:
self.storage.close()
finally:
if self.locker and self.locked:
portalocker.unlock(self.locker)
self.locker.close()
self.locked = False
ret = None
portalocker.lock(self.master,portalocker.LOCK_EX)
try:
(start, stop) = cPickle.load(self.master)
except:
(start, stop) = (0, 1)
if startup or self.now - start > 59.99:
ret = self.now
if not stop:
# this happens if previous cron job longer than 1 minute
logger.warning('WEB2PY CRON: Stale cron.master detected')
logger.debug('WEB2PY CRON: Acquiring lock')
self.master.seek(0)
cPickle.dump((self.now,0),self.master)
finally:
portalocker.unlock(self.master)
if not ret:
# do this so no need to release
self.master.close()
return ret
def unlock(self):
if self.locked:
self._f.flush()
portalocker.unlock(self._f)
self.locked = False
def _unlock(self, response):
if response and response.session_file and response.session_locked:
try:
portalocker.unlock(response.session_file)
response.session_locked = False
except: # this should never happen but happens in Windows
pass
def unlock():
for f in locked_files:
portalocker.unlock(f)
locked_files.remove(f)
logger.error('corrupted cache file %s, will try rebuild it'
% (self.shelve_name))
storage = None
if not storage and os.path.exists(self.shelve_name):
os.unlink(self.shelve_name)
storage = shelve.open(self.shelve_name)
if not CacheAbstract.cache_stats_name in storage.keys():
storage[CacheAbstract.cache_stats_name] = {
'hit_total': 0, 'misses': 0}
storage.sync()
except Exception, e:
if storage:
storage.close()
storage = None
if locked:
portalocker.unlock(locker)
locker.close()
locked = False
raise RuntimeError(
'unable to create/re-create cache file %s' % self.shelve_name)
self.locker = locker
self.locked = locked
self.storage = storage
return storage
def _unlock(self, response):
if response and response.session_file and response.session_locked:
try:
portalocker.unlock(response.session_file)
response.session_locked = False
except: ### this should never happen but happens in Windows
pass
ret = None
portalocker.lock(self.master,portalocker.LOCK_EX)
try:
(start, stop) = cPickle.load(self.master)
except:
(start, stop) = (0, 1)
if startup or self.now - start > 59.99:
ret = self.now
if not stop:
# this happens if previous cron job longer than 1 minute
logger.warning('WEB2PY CRON: Stale cron.master detected')
logger.debug('WEB2PY CRON: Acquiring lock')
self.master.seek(0)
cPickle.dump((self.now,0),self.master)
finally:
portalocker.unlock(self.master)
if not ret:
# do this so no need to release
self.master.close()
return ret