k2hr3_osnl package

Submodules

k2hr3_osnl.cfg module

Parses a config file and stores configurations.

class k2hr3_osnl.cfg.K2hr3Conf(path: Path)

Bases: ConfigOpts

Parses and stores configurations.

This class is a wrapper of oslo_config.cfg.ConfigOpts class. https://github.com/openstack/oslo.config/blob/master/oslo_config/cfg.py

Simple usage:

>>> from k2hr3_osnl.exceptions import K2hr3ConfError
>>> from k2hr3_osnl.cfg import K2hr3Conf
>>> from pathlib import Path
... try:
...         conf = K2hr3Conf(Path('etc/k2hr3_osnl.conf'))
...         print(conf.oslo_messaging_notifications.event_type)
... except K2hr3ConfError as error:
...         print('{}'.format(error))
...
^port\.delete\.end$

k2hr3_osnl.endpoint module

An endpoint for the oslo_messaging notification message listener.

class k2hr3_osnl.endpoint.K2hr3NotificationEndpoint(conf: K2hr3Conf)

Bases: object

An endpoint called by a OpenStack dispatcher.

when a filtered notification message arrives, the endpoint calls.

Simple usage:

>>> from k2hr3_osnl.cfg import K2hr3Conf
>>> from k2hr3_osnl.exceptions import K2hr3Error
>>> from k2hr3_osnl.endpoint import K2hr3NotificationEndpoint
>>> import k2hr3_osnl
>>> from pathlib import Path
>>> try:
...     conf = K2hr3Conf(Path('etc/k2hr3_osnl.conf'))
...     endpoints = [K2hr3NotificationEndpoint(conf)]
...     k2hr3_osnl.listen(endpoints)
... except K2hr3Error as error:
...     print(error)

property conf: K2hr3Conf: Returns the K2hr3Conf object.

info(context: dict[str, object], publisher_id: str, event_type: str, payload: dict[str, object], metadata: dict[str, object])

Notification endpoint in info priority.

Notification messages that match the filter’s rules will be passed to the endpoint’s methods. The oslo_messaging’s callback function dispatcher calls when messages in ‘info’ priority have arrived.

Reference: - https://docs.openstack.org/oslo.messaging/latest/reference/notification_listener.html - https://github.com/openstack/oslo.messaging/blob/master/oslo_messaging/notify/dispatcher.py#L74

Note:: This function catches all exceptions to avoid an infinite loop. If this function hasn’t handled unexpected exceptions, the caller(dispatcher) would have caught them and returned the NotificationResult.REQUEUE to the message queue server which can cause infinite loop. To avoid the posibility of inifinite loop, we catches standard exception in this function.

Parameters:

context (dict) – Context of a notification for NotificationFilter.
publisher_id (str) – Publisher_id of a notification for NotificationFilter
event_type (str) – Event_type of a notification for NotificationFilter
payload (dict) – Payload of a notification for NotificationFilter.
metadata (dict) – Metadata of a notification for NotificationFilter.

Returns:

NotificationResult.HANDLED or NotificationResult.REQUEUE

k2hr3_osnl.exceptions module

Exception classes for the oslo_messaging notification message listener.

exception k2hr3_osnl.exceptions.K2hr3ConfError(msg: str = '')

Bases: K2hr3Error

Raise when failed to instantiate a k2hr3Conf class.

exception k2hr3_osnl.exceptions.K2hr3Error

Bases: Exception

A base class of various exceptions from k2hr3_osnl package classes.

exception k2hr3_osnl.exceptions.K2hr3NotificationEndpointError(msg: str = '')

Bases: K2hr3Error

Raised when failed to instantiate a K2hr3NotificationEndpoint class.

k2hr3_osnl.httpresponse module

Response fromthe k2hr3 api.

k2hr3_osnl.useragent module

Send HTTP requests to the k2hr3 api.

Module contents

K2hr3 OpenStack Notification message Listener.

class k2hr3_osnl.K2hr3Conf(path: Path)

Bases: ConfigOpts

Parses and stores configurations.

This class is a wrapper of oslo_config.cfg.ConfigOpts class. https://github.com/openstack/oslo.config/blob/master/oslo_config/cfg.py

Simple usage:

>>> from k2hr3_osnl.exceptions import K2hr3ConfError
>>> from k2hr3_osnl.cfg import K2hr3Conf
>>> from pathlib import Path
... try:
...         conf = K2hr3Conf(Path('etc/k2hr3_osnl.conf'))
...         print(conf.oslo_messaging_notifications.event_type)
... except K2hr3ConfError as error:
...         print('{}'.format(error))
...
^port\.delete\.end$

exception k2hr3_osnl.K2hr3ConfError(msg: str = '')

Bases: K2hr3Error

Raise when failed to instantiate a k2hr3Conf class.

class k2hr3_osnl.K2hr3NotificationEndpoint(conf: K2hr3Conf)

Bases: object

An endpoint called by a OpenStack dispatcher.

when a filtered notification message arrives, the endpoint calls.

Simple usage:

>>> from k2hr3_osnl.cfg import K2hr3Conf
>>> from k2hr3_osnl.exceptions import K2hr3Error
>>> from k2hr3_osnl.endpoint import K2hr3NotificationEndpoint
>>> import k2hr3_osnl
>>> from pathlib import Path
>>> try:
...     conf = K2hr3Conf(Path('etc/k2hr3_osnl.conf'))
...     endpoints = [K2hr3NotificationEndpoint(conf)]
...     k2hr3_osnl.listen(endpoints)
... except K2hr3Error as error:
...     print(error)

property conf: K2hr3Conf: Returns the K2hr3Conf object.

info(context: dict[str, object], publisher_id: str, event_type: str, payload: dict[str, object], metadata: dict[str, object])

Notification endpoint in info priority.

Notification messages that match the filter’s rules will be passed to the endpoint’s methods. The oslo_messaging’s callback function dispatcher calls when messages in ‘info’ priority have arrived.

Reference: - https://docs.openstack.org/oslo.messaging/latest/reference/notification_listener.html - https://github.com/openstack/oslo.messaging/blob/master/oslo_messaging/notify/dispatcher.py#L74

Note:: This function catches all exceptions to avoid an infinite loop. If this function hasn’t handled unexpected exceptions, the caller(dispatcher) would have caught them and returned the NotificationResult.REQUEUE to the message queue server which can cause infinite loop. To avoid the posibility of inifinite loop, we catches standard exception in this function.

Parameters:

context (dict) – Context of a notification for NotificationFilter.
publisher_id (str) – Publisher_id of a notification for NotificationFilter
event_type (str) – Event_type of a notification for NotificationFilter
payload (dict) – Payload of a notification for NotificationFilter.
metadata (dict) – Metadata of a notification for NotificationFilter.

Returns:

NotificationResult.HANDLED or NotificationResult.REQUEUE

exception k2hr3_osnl.K2hr3NotificationEndpointError(msg: str = '')

Bases: K2hr3Error

Raised when failed to instantiate a K2hr3NotificationEndpoint class.

k2hr3_osnl.listen(endpoints: list[K2hr3NotificationEndpoint]) → int

Run a oslo_messaging notification listener for k2hr3.

This function is a library endpoint to start a oslo_messaging notification listener for k2hr3.

Parameters:: endpoints (list of K2hr3NotificationEndpoint) – endpoint to be called by dispatcher when notification messages arrive.
Returns:: 0 if success, otherwise 1.
Return type:: int

k2hr3_osnl.main() → int

Run a oslo_messaging notification listener for k2hr3.

You can configure the listener by the config file. Simple usage:

$ k2hr3_osnl -c etc/k2hr3_osnl.config

k2hr3_osnl.version() → str

Return a version of k2hr3_osnl package.

Returns:: version
Return type:: str