Coverage for lib / utils / singleton.py: 54%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3 

4# Hermes : Change Data Capture (CDC) tool from any source(s) to any target 

5# Copyright (C) 2023 INSA Strasbourg 

6# 

7# This file is part of Hermes. 

8# 

9# Hermes is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# Hermes is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with Hermes. If not, see <https://www.gnu.org/licenses/>. 

21 

22 

23# File mostly based on singleton.py from tendo package under 

24# PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2: 

25# - https://github.com/pycontribs/tendo 

26# - https://raw.githubusercontent.com/pycontribs/tendo/master/tendo/singleton.py 

27# - https://pypi.org/project/tendo/ 

28 

29 

30import os 

31import sys 

32import tempfile 

33 

34if sys.platform != "win32": 

35 import fcntl 

36 

37 

38class SingleInstanceException(BaseException): 

39 pass 

40 

41 

42class SingleInstance(object): 

43 """Class that can be instantiated only once per machine. 

44 

45 If you want to prevent your script from running in parallel just instantiate 

46 SingleInstance() class. 

47 If is there another instance already running it will throw a 

48 `SingleInstanceException`. 

49 

50 >>> import tendo 

51 ... me = SingleInstance("appname") 

52 

53 This option is very useful if you have scripts executed by crontab at small 

54 amounts of time. 

55 

56 Remember that this works by creating a lock file with a filename based on the 

57 current work dir path of the script file and the specified appname. 

58 """ 

59 

60 def __init__(self, appname: str): 

61 self.initialized = False 

62 basename = (os.getcwd() + "/" + appname).replace("/", "-").replace( 

63 ":", "" 

64 ).replace("\\", "-") + ".lock" 

65 self.lockfile = os.path.normpath(tempfile.gettempdir() + "/" + basename) 

66 

67 __hermes__.logger.debug("SingleInstance lockfile: " + self.lockfile) 

68 if sys.platform == "win32": 

69 try: 

70 # file already exists, we try to remove (in case previous 

71 # execution was interrupted) 

72 if os.path.exists(self.lockfile): 

73 os.unlink(self.lockfile) 

74 self.fd = os.open(self.lockfile, os.O_CREAT | os.O_EXCL | os.O_RDWR) 

75 except OSError: 

76 type, e, tb = sys.exc_info() 

77 if e.errno == 13: 

78 __hermes__.logger.error( 

79 "Another instance is already running, quitting." 

80 ) 

81 raise SingleInstanceException() 

82 print(e.errno) 

83 raise 

84 else: # non Windows 

85 self.fp = open(self.lockfile, "w") 

86 self.fp.flush() 

87 try: 

88 fcntl.lockf(self.fp, fcntl.LOCK_EX | fcntl.LOCK_NB) 

89 except IOError: 

90 __hermes__.logger.error( 

91 "Another instance is already running, quitting." 

92 ) 

93 raise SingleInstanceException() 

94 self.initialized = True 

95 

96 def __del__(self): 

97 if not self.initialized: 

98 return 

99 try: 

100 if sys.platform == "win32": 

101 if hasattr(self, "fd"): 

102 os.close(self.fd) 

103 os.unlink(self.lockfile) 

104 else: 

105 fcntl.lockf(self.fp, fcntl.LOCK_UN) 

106 self.fp.close() 

107 if os.path.isfile(self.lockfile): 

108 os.unlink(self.lockfile) 

109 except Exception as e: 

110 if __hermes__.logger: 

111 __hermes__.logger.warning(e) 

112 else: 

113 print("Unloggable error: %s" % e) 

114 sys.exit(-1)