-#!/bin/env python3
# -*- coding: utf-8 -*-
+"""
+@author: Frank Brehm
+@contact: frank.brehm@pixelpark.com
+@copyright: © 2022 by Frank Brehm, Berlin
+@summary: A base module for all DPX application classes
+"""
+from __future__ import absolute_import
+
+# Standard modules
+import logging
+import argparse
+
+# Third party modules
+from fb_tools.common import to_bool
+from fb_tools.cfg_app import FbConfigApplication
+from fb_tools.errors import FbAppError
+from fb_tools.multi_config import BaseMultiConfig
+
+# Own modules
+from .. import __version__ as GLOBAL_VERSION
+from .. import DEFAULT_CONFIG_DIR
+
+from ..config.ldap import MAX_TIMEOUT
+
+from ..xlate import XLATOR
+
+LOG = logging.getLogger(__name__)
+
+_ = XLATOR.gettext
+ngettext = XLATOR.ngettext
+
+__version__ = '0.2.0'
+
+
+# =============================================================================
+class DPXAppError(FbAppError):
+ """ Base exception class for all exceptions in all LDAP using application classes."""
+ pass
+
+
+# =============================================================================
+class TimeoutOptionAction(argparse.Action):
+
+ # -------------------------------------------------------------------------
+ def __init__(self, option_strings, *args, **kwargs):
+
+ super(TimeoutOptionAction, self).__init__(
+ option_strings=option_strings, *args, **kwargs)
+
+ # -------------------------------------------------------------------------
+ def __call__(self, parser, namespace, given_timeout, option_string=None):
+
+ try:
+ timeout = int(given_timeout)
+ if timeout <= 0 or timeout > MAX_TIMEOUT:
+ msg = _(
+ "a timeout must be greater than zero and less "
+ "or equal to {}.").format(MAX_TIMEOUT)
+ raise ValueError(msg)
+ except (ValueError, TypeError) as e:
+ msg = _("Wrong timeout {!r}:").format(given_timeout)
+ msg += ' ' + str(e)
+ raise argparse.ArgumentError(self, msg)
+
+ setattr(namespace, self.dest, timeout)
+
+# =============================================================================
+class BaseDPXApplication(FbConfigApplication):
+ """
+ Base class for all DPX application classes.
+ """
+
+ default_prompt_timeout = 10
+ max_prompt_timeout = 600
+
+ # -------------------------------------------------------------------------
+ def __init__(
+ self, appname=None, verbose=0, version=GLOBAL_VERSION, base_dir=None,
+ cfg_class=BaseMultiConfig, initialized=False, usage=None, description=None,
+ argparse_epilog=None, argparse_prefix_chars='-', env_prefix=None,
+ config_dir=DEFAULT_CONFIG_DIR):
+
+ self._yes = False
+ self._prompt_timeout = self.default_prompt_timeout
+
+ super(BaseDPXApplication, self).__init__(
+ appname=appname, verbose=verbose, version=version, base_dir=base_dir,
+ description=description, cfg_class=cfg_class, initialized=False,
+ argparse_epilog=argparse_epilog, argparse_prefix_chars=argparse_prefix_chars,
+ env_prefix=env_prefix, config_dir=config_dir
+ )
+
+ # -----------------------------------------------------------
+ @property
+ def yes(self):
+ """Assume 'yes' as an answer to all questions."""
+ return self._yes
+
+ @yes.setter
+ def yes(self, value):
+ self._yes = to_bool(value)
+
+ # -----------------------------------------------------------
+ @property
+ def prompt_timeout(self):
+ """The timeout in seconds for waiting for an answer on a prompt."""
+ return getattr(self, '_prompt_timeout', self.default_prompt_timeout)
+
+ @prompt_timeout.setter
+ def prompt_timeout(self, value):
+ v = int(value)
+ if v < 0 or v > self.max_prompt_timeout:
+ LOG.warning(_("Wrong prompt timeout {v!r}, must be 0 >= 0 and <= {max}.").format(
+ v=value, max=self.max_prompt_timeout))
+ else:
+ self._prompt_timeout = v
+
+ # -------------------------------------------------------------------------
+ def as_dict(self, short=True):
+ """
+ Transforms 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(BaseDPXApplication, self).as_dict(short=short)
+
+ res['prompt_timeout'] = self.prompt_timeout
+ res['yes'] = self.yes
+
+ return res
+
+ # -------------------------------------------------------------------------
+ def init_arg_parser(self):
+ """
+ Public available method to initiate the argument parser.
+ """
+
+ super(BaseDPXApplication, self).init_arg_parser()
+
+ self.arg_parser.add_argument(
+ '-Y', '--yes', '--assume-yes', action="store_true", dest="yes", help=argparse.SUPPRESS)
+
+ # -------------------------------------------------------------------------
+ def post_init(self):
+ """
+ Method to execute before calling run(). Here could be done some
+ finishing actions after reading in commandline parameters,
+ configuration a.s.o.
+
+ This method could be overwritten by descendant classes, these
+ methhods should allways include a call to post_init() of the
+ parent class.
+
+ """
+
+ self.initialized = False
+
+ super(BaseDPXApplication, self).post_init()
+
+ self.yes = self.args.yes
-__version__ = '0.1.0'
# vim: ts=4 et list
# from ldap3.core.exceptions import LDAPInvalidDnError, LDAPInvalidValueError
# from ldap3.core.exceptions import LDAPException, LDAPBindError
-from fb_tools.common import pp, to_bool, is_sequence
-from fb_tools.cfg_app import FbConfigApplication
-from fb_tools.errors import FbAppError
+from fb_tools.common import pp, is_sequence
from fb_tools.mailaddress import MailAddress
from fb_tools.collections import FrozenCIStringSet, CIStringSet
# from ..argparse_actions import PortOptionAction
+from . import DPXAppError, TimeoutOptionAction, BaseDPXApplication
+
# from ..config.ldap import LdapConfigError
from ..config.ldap import LdapConnectionInfo, LdapConfiguration
# rom ..config.ldap import DEFAULT_PORT_LDAP, DEFAULT_PORT_LDAPS
-from ..config.ldap import DEFAULT_TIMEOUT, MAX_TIMEOUT
+from ..config.ldap import DEFAULT_TIMEOUT
__version__ = '0.4.2'
LOG = logging.getLogger(__name__)
# =============================================================================
-class LdapAppError(FbAppError):
+class LdapAppError(DPXAppError):
""" Base exception class for all exceptions in all LDAP using application classes."""
pass
# =============================================================================
-class TimeoutOptionAction(argparse.Action):
-
- # -------------------------------------------------------------------------
- def __init__(self, option_strings, *args, **kwargs):
-
- super(TimeoutOptionAction, self).__init__(
- option_strings=option_strings, *args, **kwargs)
-
- # -------------------------------------------------------------------------
- def __call__(self, parser, namespace, given_timeout, option_string=None):
-
- try:
- timeout = int(given_timeout)
- if timeout <= 0 or timeout > MAX_TIMEOUT:
- msg = _(
- "a timeout must be greater than zero and less "
- "or equal to {}.").format(MAX_TIMEOUT)
- raise ValueError(msg)
- except (ValueError, TypeError) as e:
- msg = _("Wrong timeout {!r}:").format(given_timeout)
- msg += ' ' + str(e)
- raise argparse.ArgumentError(self, msg)
-
- setattr(namespace, self.dest, timeout)
-
-
-# =============================================================================
-class BaseLdapApplication(FbConfigApplication):
+class BaseLdapApplication(BaseDPXApplication):
"""
Base class for all application classes using LDAP.
"""
self._password_file = None
self.ldap_instances = []
- self._yes = False
self.ldap_server = {}
self.ldap_connection = {}
self._password_file = path
- # -----------------------------------------------------------
- @property
- def yes(self):
- """Assume 'yes' as an answer to all questions."""
- return self._yes
-
- @yes.setter
- def yes(self, value):
- self._yes = to_bool(value)
-
# -------------------------------------------------------------------------
def as_dict(self, short=True):
"""
res['show_cmdline_ldap_timeout'] = self.show_cmdline_ldap_timeout
res['use_default_ldap_connection'] = self.use_default_ldap_connection
res['use_multiple_ldap_connections'] = self.use_multiple_ldap_connections
- res['yes'] = self.yes
return res
group_title = _('Options fo LDAP connections')
ldap_group = self.arg_parser.add_argument_group(group_title)
- ldap_group.add_argument(
- '-y', '--yes', '--assume-yes', action="store_true", dest="yes", help=argparse.SUPPRESS)
-
if self.use_default_ldap_connection:
ldap_host = LdapConfiguration.default_ldap_server
self._init_default_connection()
- self.yes = self.args.yes
-
if self.use_default_ldap_connection:
self.ldap_instances = ['default']
return
projects / pixelpark / pp-admin-tools.git / commitdiff