cengal.parallel_execution.coroutines.coro_standard_services.shutdown_on_keyboard_interrupt.versions.v_0.shutdown_on_keyboard_interrupt

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

 1#!/usr/bin/env python
 2# coding=utf-8
 3
 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
 5# 
 6# Licensed under the Apache License, Version 2.0 (the "License");
 7# you may not use this file except in compliance with the License.
 8# You may obtain a copy of the License at
 9# 
10#     http://www.apache.org/licenses/LICENSE-2.0
11# 
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17
18"""
19Module Docstring
20Docstrings: http://www.python.org/dev/peps/pep-0257/
21"""
22
23__author__ = "ButenkoMS <gtalk@butenkoms.space>"
24__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
25__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
26__license__ = "Apache License, Version 2.0"
27__version__ = "4.4.1"
28__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
29__email__ = "gtalk@butenkoms.space"
30# __status__ = "Prototype"
31__status__ = "Development"
32# __status__ = "Production"
33
34
35__all__ = ['ShutdownOnKeyboardInterrupt']
36
37
38from cengal.parallel_execution.coroutines.coro_scheduler import *
39import signal
40from typing import Tuple
41
42
43class ShutdownOnKeyboardInterrupt(TypedService[None]):
44    def __init__(self, loop: CoroSchedulerType):
45        super().__init__(loop)
46        self.handler_set: bool = False
47        self.previous_handler = None
48        self.keyboard_interrupt_emited: bool = False
49
50    def single_task_registration_or_immediate_processing(
51            self, *args, **kwargs) -> Tuple[bool, None, None]:
52        if not self.handler_set:
53            self.previous_handler = signal.signal(signal.SIGINT, self.keyboard_interrupt_handler)
54            self.handler_set = True
55        
56        return True, None, None
57
58    def full_processing_iteration(self):
59        if self.keyboard_interrupt_emited:
60            signal.signal(signal.SIGINT, self.previous_handler)
61            self.handler_set = False
62            if callable(self.previous_handler):
63                def on_cs_destroy_handler():
64                    self.previous_handler(signal.SIGINT, None)
65                
66                self._loop.on_destroyed_handlers.add(on_cs_destroy_handler)
67
68            raise CoroSchedulerDestroyRequestedException
69        
70        self.make_dead()
71
72    def in_work(self) -> bool:
73        result: bool = self.keyboard_interrupt_emited
74        return self.thrifty_in_work(result)
75    
76    def keyboard_interrupt_handler(self, sig, frame):
77        self.keyboard_interrupt_emited = True
78        self.make_live()
79    
80    def destroy(self):
81        if self.handler_set:
82            signal.signal(signal.SIGINT, self.previous_handler)
83            self.handler_set = False

class ShutdownOnKeyboardInterrupt(cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.TypedService[NoneType]): View Source

44class ShutdownOnKeyboardInterrupt(TypedService[None]):
45    def __init__(self, loop: CoroSchedulerType):
46        super().__init__(loop)
47        self.handler_set: bool = False
48        self.previous_handler = None
49        self.keyboard_interrupt_emited: bool = False
50
51    def single_task_registration_or_immediate_processing(
52            self, *args, **kwargs) -> Tuple[bool, None, None]:
53        if not self.handler_set:
54            self.previous_handler = signal.signal(signal.SIGINT, self.keyboard_interrupt_handler)
55            self.handler_set = True
56        
57        return True, None, None
58
59    def full_processing_iteration(self):
60        if self.keyboard_interrupt_emited:
61            signal.signal(signal.SIGINT, self.previous_handler)
62            self.handler_set = False
63            if callable(self.previous_handler):
64                def on_cs_destroy_handler():
65                    self.previous_handler(signal.SIGINT, None)
66                
67                self._loop.on_destroyed_handlers.add(on_cs_destroy_handler)
68
69            raise CoroSchedulerDestroyRequestedException
70        
71        self.make_dead()
72
73    def in_work(self) -> bool:
74        result: bool = self.keyboard_interrupt_emited
75        return self.thrifty_in_work(result)
76    
77    def keyboard_interrupt_handler(self, sig, frame):
78        self.keyboard_interrupt_emited = True
79        self.make_live()
80    
81    def destroy(self):
82        if self.handler_set:
83            signal.signal(signal.SIGINT, self.previous_handler)
84            self.handler_set = False

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def __getitem__(self, key: KT) -> VT: ... # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

ShutdownOnKeyboardInterrupt( loop: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable]) View Source

45    def __init__(self, loop: CoroSchedulerType):
46        super().__init__(loop)
47        self.handler_set: bool = False
48        self.previous_handler = None
49        self.keyboard_interrupt_emited: bool = False

handler_set: bool

previous_handler

keyboard_interrupt_emited: bool

def single_task_registration_or_immediate_processing(self, *args, **kwargs) -> Tuple[bool, NoneType, NoneType]: View Source

51    def single_task_registration_or_immediate_processing(
52            self, *args, **kwargs) -> Tuple[bool, None, None]:
53        if not self.handler_set:
54            self.previous_handler = signal.signal(signal.SIGINT, self.keyboard_interrupt_handler)
55            self.handler_set = True
56        
57        return True, None, None

def full_processing_iteration(self): View Source

59    def full_processing_iteration(self):
60        if self.keyboard_interrupt_emited:
61            signal.signal(signal.SIGINT, self.previous_handler)
62            self.handler_set = False
63            if callable(self.previous_handler):
64                def on_cs_destroy_handler():
65                    self.previous_handler(signal.SIGINT, None)
66                
67                self._loop.on_destroyed_handlers.add(on_cs_destroy_handler)
68
69            raise CoroSchedulerDestroyRequestedException
70        
71        self.make_dead()

def in_work(self) -> bool: View Source

73    def in_work(self) -> bool:
74        result: bool = self.keyboard_interrupt_emited
75        return self.thrifty_in_work(result)

Will be executed twice per iteration: once before and once after the full_processing_iteration() execution

Raises: NotImplementedError: _description_

Returns: bool: _description_

def keyboard_interrupt_handler(self, sig, frame): View Source

77    def keyboard_interrupt_handler(self, sig, frame):
78        self.keyboard_interrupt_emited = True
79        self.make_live()

def destroy(self): View Source

81    def destroy(self):
82        if self.handler_set:
83            signal.signal(signal.SIGINT, self.previous_handler)
84            self.handler_set = False

Inherited Members

cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.Service

current_caller_coro_info

iteration

make_response

register_response

put_task

resolve_request

try_resolve_request

in_forground_work

thrifty_in_work

time_left_before_next_event

is_low_latency

make_live

make_dead

service_id_impl

service_id