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.

__context: Instance

Reference to the context aka parent object. Derived automatically in the override implementation of __setattr__.

__depends_on: set

Set of other instances that is this instance depending on. Note however that the type of the dependencies are checked dynamically in runtime, since dependencies can only be defined on the same type of instance.

__expected_parameters: dict[str, ExpectedParameter]

Map of expected parameters defined as descriptor of the class. Key is the name of the parameter, value is the parameter descriptor. Used to check, if an expected (described) parameter value shall be set upon instance parameter setting.

__full_name: str

Full name of the instance which uniquely identifies it up to its topmost context. For example, if an instance A has a parent context B, which has a parent context C, then the full name of A is ‘C.B.A’. This value is calculated and the result is stored after the first calculation to avoid recalculation every time.

__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.

__nested_instance_type: Type[NestedInstanceType] | None

Stores the specified type of the nested instances. This is required to be able to discover those instances

__nested_instances: dict[str, NestedInstanceType]

This dictionary holds all the instances that are nested in the context of this instance object

__parameters: InstanceParameters

The interpreted instance parameters i.e., cascading and templates are interpreted

__simple_name: str

Name of the instance, which is represented by the object created from the implementation class. If not provided and there is a parent context, then the parent context will use the name of the variable.

__spec_classes: set

Set of specs classes that are in the class hierarchy i.e., which specs classes are contributing to the implementation of this class

__spec_name: str

The name of the spec constructed of module and qualified class name. Notice that we separate the module name and the class name so that we can identify them at loading by name.

_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