Adding lib/create_terraform/consul.py

diff --git a/lib/create_terraform/consul.py b/lib/create_terraform/consul.py
new file mode 100644 (file)
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