pypz.plugins.misc.health_check module

class pypz.plugins.misc.health_check.HttpHealthCheckPlugin(name: str = None, *args, **kwargs)

Bases: ServicePlugin

This service plugin starts a small HTTP server with a single endpoint, which remains active in the lifetime of the operator. It allows external entities to realize HTTP based health checks.

__logger: ContextLogger | None

Context logger, which is the Operator’s logger if Operator context existing, otherwise it defaults back to the DefaultContextLogger. Note that Plugin without Operator context makes only sense in test cases, hence the default log level is set to DEBUG.

_abc_impl = <_abc._abc_data object>

_on_error(source: Any, exception: Exception) → None

This method can be implemented to react to error events during execution. The error itself may come from arbitrary sources.

_on_interrupt(system_signal: int = None) → None

This method can be implemented to react to interrupt signals like SIGINT, SIGTERM etc. The specs implementation can then execute interrupt logic e.g., early termination of loops.

Parameters:

system_signal – id of the system signal that causes interrupt

_on_service_shutdown() → bool

This method shall implement the logic of stopping the service and clean up the residuals. E.g., a mounting service could unmount or a background thread could be stopped. VERY IMPORTANT NOTE - you must always check, if your service start method has been called, because it can be that it is never called, if there was an exception raised from other entity. However, the shutdown will be called anyway.

Returns:

True if logic finished, False if it needs more iteration

_on_service_start() → bool

This method shall implement the starting logic of the service. You can consider services as example like mounting service, which only mounts a folder to the system or a background service, which starts a background thread.

Returns:

True if logic finished, False if it needs more iteration

_start_http_server() → None

_stop_http_server() → None

port

Convenience class to represent an optional parameter Usage:

class TestImpl(Instance):
    required_param = RequiredParameter(str)
    optional_param = OptionalParameter(str)
    def __init__(self):
        self.required_param = None
        self.optional_param = "defaultValue"

Parameters:

• parameter_type – (str, int, float, set, list, dict, type(None))

• alt_name – alternative name for the parameter, if specified it acts as reference to the parameter

• on_update – callback to react on value update