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