Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:rtype: bool
"""
if self._endtime is not None and time.time() > self._endtime:
return False
return True
def __repr__(self):
"""Readable executor representation."""
return '<{module}.{executor}: "{command}">'.format(
module=self.__class__.__module__,
executor=self.__class__.__name__,
command=self.command[:30]
)
class StartCheckExecutor(Executor):
""" Base class for executors with a pre- and after-start checks. """
def pre_start_check(self):
"""
Method fired before the start of executor.
Should be overridden in order to return boolean value if some
process is already started.
:rtype: bool
"""
raise NotImplementedError
def start(self):
"""
Start executor with additional checks.
# mirakuru is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with mirakuru. If not, see .
"""TCP executor definition."""
import socket
from typing import Union, List, Tuple, Any
from mirakuru.base import Executor
class TCPExecutor(Executor):
"""
TCP-listening process executor.
Used to start (and wait to actually be running) processes that can accept
TCP connections.
"""
def __init__(self,
command: Union[str, List[str], Tuple[str, ...]],
host: str,
port: int,
**kwargs: Any) -> None:
"""
Initialize TCPExecutor executor.
:param (str, list) command: command to be run by the subprocess
# mirakuru is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with mirakuru. If not, see .
"""Pid executor definition."""
import os.path
from typing import Union, List, Tuple, Any
from mirakuru.base import Executor
class PidExecutor(Executor):
"""
File existence checking process executor.
Used to start processes that create pid files (or any other for that
matter). Starts the given process and waits for the given file to be
created.
"""
def __init__(self,
command: Union[str, List[str], Tuple[str, ...]],
filename: str,
**kwargs: Any) -> None:
"""
Initialize the PidExecutor executor.
If the filename is empty, a ValueError is thrown.