blob: 9f53b8d59fc2a502f2d1a9511809b73dd751ccc6 (
about) (
plain) (
blame)
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
|
import datetime, os
from src import PollHook, utils
EXPIRATION = 60 # 1 minute
class LockFile(PollHook.PollHook):
def __init__(self, database_location: str):
self._database_location = database_location
self._lock_location = "%s.lock" % database_location
self._next_lock = None
def available(self):
now = utils.datetime_utcnow()
if os.path.exists(self._lock_location):
with open(self._lock_location, "r") as lock_file:
timestamp_str = lock_file.read().strip().split(" ", 1)[0]
timestamp = utils.iso8601_parse(timestamp_str)
if (now-timestamp).total_seconds() < EXPIRATION:
return False
return True
def lock(self):
with open(self._lock_location, "w") as lock_file:
last_lock = utils.datetime_utcnow()
lock_file.write("%s" % utils.iso8601_format(last_lock))
self._next_lock = last_lock+datetime.timedelta(
seconds=EXPIRATION/2)
def next(self):
return max(0, (self._next_lock-utils.datetime_utcnow()).total_seconds())
def call(self):
if self.next() == 0:
self.lock()
def unlock(self):
if os.path.isfile(self._lock_location):
os.remove(self._lock_location)
|
