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, 2024 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 

34 

35if sys.platform != "win32": 

36 import fcntl 

37 

38 

39class SingleInstanceException(BaseException): 

40 pass 

41 

42 

43class SingleInstance(object): 

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

45 

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

47 SingleInstance() class. 

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

49 `SingleInstanceException`. 

50 

51 >>> import tendo 

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

53 

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

55 amounts of time. 

56 

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

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

59 """ 

60 

61 def __init__(self, appname: str): 

62 self.initialized = False 

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

64 ":", "" 

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

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

67 

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

69 if sys.platform == "win32": 

70 try: 

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

72 # execution was interrupted) 

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

74 os.unlink(self.lockfile) 

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

76 except OSError: 

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

78 if e.errno == 13: 

79 __hermes__.logger.error( 

80 "Another instance is already running, quitting." 

81 ) 

82 raise SingleInstanceException() 

83 print(e.errno) 

84 raise 

85 else: # non Windows 

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

87 self.fp.flush() 

88 try: 

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

90 except IOError: 

91 __hermes__.logger.error( 

92 "Another instance is already running, quitting." 

93 ) 

94 raise SingleInstanceException() 

95 self.initialized = True 

96 

97 def __del__(self): 

98 if not self.initialized: 

99 return 

100 try: 

101 if sys.platform == "win32": 

102 if hasattr(self, "fd"): 

103 os.close(self.fd) 

104 os.unlink(self.lockfile) 

105 else: 

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

107 self.fp.close() 

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

109 os.unlink(self.lockfile) 

110 except Exception as e: 

111 if __hermes__.logger: 

112 __hermes__.logger.warning(e) 

113 else: 

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

115 sys.exit(-1)