Coverage for aiocoap/util/cli.py: 87%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

47 statements  

1# This file is part of the Python aiocoap library project. 

2# 

3# Copyright (c) 2012-2014 Maciej Wasilak <http://sixpinetrees.blogspot.com/>, 

4# 2013-2014 Christian Amsüss <c.amsuess@energyharvesting.at> 

5# 

6# aiocoap is free software, this file is published under the MIT license as 

7# described in the accompanying LICENSE file. 

8 

9"""Helpers for creating server-style applications in aiocoap 

10 

11Note that these are not particular to aiocoap, but are used at different places 

12in aiocoap and thus shared here.""" 

13 

14import argparse 

15import sys 

16import logging 

17import asyncio 

18import signal 

19 

20from ..util.asyncio import py38args 

21 

22class ActionNoYes(argparse.Action): 

23 """Simple action that automatically manages --{,no-}something style options""" 

24 # adapted from Omnifarious's code on 

25 # https://stackoverflow.com/questions/9234258/in-python-argparse-is-it-possible-to-have-paired-no-something-something-arg#9236426 

26 def __init__(self, option_strings, dest, default=True, required=False, help=None): 

27 assert len(option_strings) == 1, "ActionNoYes takes only one option name" 

28 assert option_strings[0].startswith('--'), "ActionNoYes options must start with --" 

29 super().__init__(['--' + option_strings[0][2:], '--no-' + option_strings[0][2:]], dest, nargs=0, const=None, default=default, required=required, help=help) 

30 def __call__(self, parser, namespace, values, option_string=None): 

31 if option_string.startswith('--no-'): 

32 setattr(namespace, self.dest, False) 

33 else: 

34 setattr(namespace, self.dest, True) 

35 

36class AsyncCLIDaemon: 

37 """Helper for creating daemon-style CLI prorgrams. 

38 

39 Note that this currently doesn't create a Daemon in the sense of doing a 

40 daemon-fork; that could be added on demand, though. 

41 

42 Subclass this and implement the :meth:`start` method as an async 

43 function; it will be passed all the constructor's arguments. 

44 

45 When all setup is complete and the program is operational, return from the 

46 start method. 

47 

48 Implement the :meth:`shutdown` coroutine and to do cleanup; what actually 

49 runs your program will, if possible, call that and await its return. 

50 

51 Two usage patterns for this are supported: 

52 

53 * Outside of an async context, run run ``MyClass.sync_main()``, typically 

54 in the program's ``if __name__ == "__main__":`` section. 

55 

56 In this mode, the loop that is started is configured to safely shut down 

57 the loop when SIGINT is received. 

58 

59 * To run a subclass of this in an existing loop, start it with 

60 ``MyClass(...)`` (possibly passing in the loop to run it on if not already 

61 in an async context), and then awaiting its ``.initializing`` future. To 

62 stop it, await its ``.shutdown()`` method. 

63 

64 Note that with this usage pattern, the :meth:`.stop()` method has no 

65 effect; servers that ``.stop()`` themselves need to signal their desire 

66 to be shut down through other channels (but that is an atypical case). 

67 """ 

68 

69 def __init__(self, *args, **kwargs): 

70 loop = kwargs.pop('loop', None) 

71 if loop is None: 

72 loop = asyncio.get_running_loop() 

73 self.__exitcode = loop.create_future() 

74 self.initializing = loop.create_task( 

75 self.start(*args, **kwargs), 

76 **py38args(name="Initialization of %r" % (self,)) 

77 ) 

78 

79 def stop(self, exitcode): 

80 """Stop the operation (and exit sync_main) at the next convenience.""" 

81 self.__exitcode.set_result(exitcode) 

82 

83 @classmethod 

84 async def _async_main(cls, *args, **kwargs): 

85 """Run the application in an AsyncIO main loop, shutting down cleanly 

86 on keyboard interrupt. 

87 

88 This is not exposed publicly as it messes with the loop, and we only do 

89 that with loops created in sync_main. 

90 """ 

91 main = cls(*args, **kwargs) 

92 

93 try: 

94 asyncio.get_running_loop().add_signal_handler( 

95 signal.SIGTERM, 

96 lambda: main.__exitcode.set_result(143), 

97 ) 

98 except NotImplementedError: 

99 # Impossible on win32 -- just won't make that clean of a shutdown. 

100 pass 

101 

102 try: 

103 await main.initializing 

104 # This is the time when we'd signal setup completion by the parent 

105 # exiting in case of a daemon setup, or to any other process 

106 # management. 

107 logging.info("Application ready.") 

108 # Common options are 143 or 0 

109 # (<https://github.com/go-task/task/issues/75#issuecomment-339466142> and 

110 # <https://unix.stackexchange.com/questions/10231/when-does-the-system-send-a-sigterm-to-a-process>) 

111 exitcode = await main.__exitcode 

112 except KeyboardInterrupt: 

113 logging.info("Keyboard interupt received, shutting down") 

114 sys.exit(3) 

115 else: 

116 sys.exit(exitcode) 

117 finally: 

118 if main.initializing.done() and main.initializing.exception(): 

119 # The exception if initializing is what we are just watching 

120 # fly by. No need to trigger it again, and running shutdown 

121 # would be even weirder. 

122 pass 

123 else: 

124 # May be done, then it's a no-op, or we might have received a 

125 # signal during startup in which case we better fetch the 

126 # result and shut down cleanly again 

127 await main.initializing 

128 

129 # And no matter whether that happened during initialization 

130 # (which now has finished) or due to a regular signal... 

131 await main.shutdown() 

132 

133 @classmethod 

134 def sync_main(cls, *args, **kwargs): 

135 """Run the application in an AsyncIO main loop, shutting down cleanly 

136 on keyboard interrupt.""" 

137 asyncio.run(cls._async_main(*args, **kwargs))