From: Frank Brehm Date: Fri, 24 May 2024 12:56:30 +0000 (+0200) Subject: Adding lib/create_terraform/consul.py X-Git-Tag: 1.9.1^2~8 X-Git-Url: https://git.uhu-banane.de/?a=commitdiff_plain;h=26481aa52d247e398a88c20bfce810cac3622783;p=pixelpark%2Fcreate-terraform.git Adding lib/create_terraform/consul.py --- diff --git a/lib/create_terraform/consul.py b/lib/create_terraform/consul.py new file mode 100644 index 0000000..1b3718c --- /dev/null +++ b/lib/create_terraform/consul.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +@author: Frank Brehm +@contact: frank.brehm@pixelpark.com +@copyright: © 2024 by Frank Brehm, Berlin +@summary: A handler module for managing Consul key/value pairs +""" +from __future__ import absolute_import, print_function + +# Standard module +import logging +import os + +# Third party modules +from fb_tools.common import to_bool +from fb_tools.handling_obj import HandlingObject + +import requests +from requests.exceptions import RequestException + +# Own modules +from . import DEFAULT_CONSUL_API_KV_ROOTPATH +from . import DEFAULT_CONSUL_API_TIMEOUT +from . import DEFAULT_CONSUL_PORT +from . import DEFAULT_CONSUL_SERVER +from . import LIBRARY_NAME +from . import MAX_PORT_NUMBER +from . import __version__ as GLOBAL_VERSION + +from .errors import ConsulHandlerError + +from .xlate import XLATOR + +__version__ = '0.1.0' +LOG = logging.getLogger(__name__) + +LOGLEVEL_REQUESTS_SET = False + +_ = XLATOR.gettext + + +# ============================================================================= +class ConsulHandler(HandlingObject): + """ + Class for a object handling with key/value pairs on a consul server. + """ + + default_consul_server = DEFAULT_CONSUL_SERVER + default_kv_rootpath = DEFAULT_CONSUL_API_KV_ROOTPATH + default_port = DEFAULT_CONSUL_PORT + default_timeout = DEFAULT_CONSUL_API_TIMEOUT + default_use_https = True + + loglevel_requests_set = LOGLEVEL_REQUESTS_SET + + # ------------------------------------------------------------------------- + def __init__( + self, version=__version__, consul_server=None, port=None, + use_https=None, timeout=None, path_prefix=None, + *args, **kwargs): + """Initialize a ConsulHandler object.""" + self._consul_server = self.default_consul_server + self._port = self.default_port + self._use_https = self.default_use_https + self._path_prefix = self.default_kv_rootpath + self._timeout = self.default_timeout + self._user_agent = '{}/{}'.format(LIBRARY_NAME, GLOBAL_VERSION) + self._mocked = False + self.mocking_paths = [] + + super(ConsulHandler, self).__init__(version=version, *args, **kwargs) + + if consul_server is not None: + self.consul_server = consul_server + if port is not None: + self.port = port + if use_https is not None: + self.use_https = use_https + if timeout is not None: + self.timeout = timeout + if path_prefix is not None: + self.path_prefix = path_prefix + + if not self.loglevel_requests_set: + msg = _('Setting loglevel of the {m} module to {ll}.').format( + m='requests', ll='WARNING') + LOG.debug(msg) + logging.getLogger('requests').setLevel(logging.WARNING) + self.loglevel_requests_set = True + + if 'initialized' in kwargs: + self.initialized = kwargs['initialized'] + + # ----------------------------------------------------------- + @property + def consul_server(self): + """The hostname or address of the Consul server.""" + return self._consul_server + + @consul_server.setter + def consul_server(self, value): + if value is None: + self._consul_server = None + return + + val = str(value).strip().lower() + if val == '': + self._consul_server = None + else: + self._consul_server = val + + # ----------------------------------------------------------- + @property + def port(self): + """The TCP port number of the Consul API.""" + return self._port + + @port.setter + def port(self, value): + if value is None: + self._port = self.default_port + return + val = int(value) + err_msg = _( + 'Invalid port number {port!r} for the Consul API, must be greater than zero ' + 'and less than or equal to {max}.').format(port=value, max=(MAX_PORT_NUMBER + 1)) + if val <= 0 or val >= MAX_PORT_NUMBER: + raise ValueError(err_msg) + self._port = val + + # ----------------------------------------------------------- + @property + def use_https(self): + """Return, whether to use HTTPS to communicate with the Consul API.""" + if self.mocked: + return False + return self._use_https + + @use_https.setter + def use_https(self, value): + self._use_https = to_bool(value) + + # ----------------------------------------------------------- + @property + def mocked(self): + """Flag, that a mocked URI should be used.""" + return self._mocked + + @mocked.setter + def mocked(self, value): + self._mocked = to_bool(value) + + # ----------------------------------------------------------- + @property + def path_prefix(self): + """The hostname or address of the PowerDNS master server.""" + return self._path_prefix + + @path_prefix.setter + def path_prefix(self, value): + if value is None: + self._path_prefix = None + return + + val = str(value).strip() + if val == '': + self._path_prefix = None + else: + if not os.path.isabs(val): + msg = _('The path prefix {!r} must be an absolute path.').format(value) + raise ValueError(msg) + self._path_prefix = val + + # ----------------------------------------------------------- + @property + def timeout(self): + """The timeout in seconds for requesting the Consul API.""" + return self._timeout + + @timeout.setter + def timeout(self, value): + if value is None: + self._timeout = self.default_timeout + return + val = int(value) + err_msg = _( + 'Invalid timeout {!r} for requesting the Consul API, must be greater than zero and ' + 'less or equal to 3600.') + if val <= 0 or val > 3600: + msg = err_msg.format(value) + raise ValueError(msg) + self._timeout = val + + # ----------------------------------------------------------- + @property + def user_agent(self): + """The name of the user agent used in API calls.""" + return self._user_agent + + @user_agent.setter + def user_agent(self, value): + if value is None or str(value).strip() == '': + raise ConsulHandlerError(_('Invalid user agent {!r} given.').format(value)) + self._user_agent = str(value).strip() + + # ------------------------------------------------------------------------- + def as_dict(self, short=True): + """ + Transform the elements of the object into a dict. + + @param short: don't include local properties in resulting dict. + @type short: bool + + @return: structure as dict + @rtype: dict + """ + res = super(ConsulHandler, self).as_dict(short=short) + + res['consul_server'] = self.consul_server + res['default_consul_server'] = self.default_consul_server + res['default_kv_rootpath'] = self.default_kv_rootpath + res['default_port'] = self.default_port + res['default_timeout'] = self.default_timeout + res['default_use_https'] = self.default_use_https + res['path_prefix'] = self.path_prefix + res['port'] = self.port + res['mocked'] = self.mocked + res['timeout'] = self.timeout + res['use_https'] = self.use_https + res['user_agent'] = self.user_agent + + return res + + # ------------------------------------------------------------------------- + def _build_url(self, key): + + if not os.path.isabs(path): + msg = _('The path {!r} must be an absolute path.').format(path) + raise ValueError(msg) + + url = 'http://{}'.format(self.consul_server) + if self.mocked: + url = 'mock://{}'.format(self.consul_server) + elif self.use_https: + url = 'https://{}'.format(self.consul_server) + if self.port != 443: + url += ':{}'.format(self.port) + else: + if self.port != 80: + url += ':{}'.format(self.port) + + url += self.path_prefix + '/' + key + + return url + + + + + +# vim: ts=4 et list