k2hr3_osnl package

Submodules

k2hr3_osnl.cfg module

Parses a config file and stores configurations.

class k2hr3_osnl.cfg.K2hr3Conf(path: Path)[source]

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)[source]

Bases: object

An endpoint called by a OpenStack dispatcher when a filtered notification message arrives.

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])[source]

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:

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 = '')[source]

Bases: K2hr3Error

Raised when failed to instantiate a k2hr3Conf class.

exception k2hr3_osnl.exceptions.K2hr3Error[source]

Bases: Exception

A base class of various exceptions from k2hr3_osnl package classes.

exception k2hr3_osnl.exceptions.K2hr3NotificationEndpointError(msg: str = '')[source]

Bases: K2hr3Error

Raised when failed to instantiate a K2hr3NotificationEndpoint class.

k2hr3_osnl.httpresponse module

Sends http requests to the k2hr3 api. Classes in this module are not public.

k2hr3_osnl.useragent module

Sends http requests to the k2hr3 api. Classes in this module are not public.

Module contents

K2hr3 OpenStack Notification message Listener.

class k2hr3_osnl.K2hr3Conf(path: Path)[source]

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 = '')[source]

Bases: K2hr3Error

Raised when failed to instantiate a k2hr3Conf class.

class k2hr3_osnl.K2hr3NotificationEndpoint(conf: K2hr3Conf)[source]

Bases: object

An endpoint called by a OpenStack dispatcher when a filtered notification message arrives.

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])[source]

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:

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 = '')[source]

Bases: K2hr3Error

Raised when failed to instantiate a K2hr3NotificationEndpoint class.

k2hr3_osnl.listen(endpoints: List[K2hr3NotificationEndpoint]) int[source]

Runs 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[source]

Runs 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

Returns:

0 if success, otherwise 1.

Return type:

int

k2hr3_osnl.version() str[source]

Returns a version of k2hr3_osnl package.

Returns:

version

Return type:

str