Python configobj 模块，flatten_errors() 实例源码

def report_config_errors(config, result):
    """
    Interpret configobj results and report configuration errors to the user.
    """
    # from http://www.voidspace.org.uk/python/configobj.html#example-usage
    print 'Invalid config file:'
    for entry in configobj.flatten_errors(config, result):
        # each entry is a tuple
        section_list, key, error = entry
        if key is not None:
            section_list.append(key)
        else:
            section_list.append('(missing section)')
        section_string = ', '.join(section_list)
        if error == False:
            error = 'Invalid value (or section).'
        print '{}: {}'.format(section_string, error)

def _validate_conf(self):
        """ Check if the configuration file is correct """
        validator = Validator()
        results = self.config.validate(validator)
        if not results:
            for entry in configobj.flatten_errors(self.config, results):
                # each entry is a tuple
                section_list, key, error = entry
                if key is not None:
                    section_list.append(key)
                else:
                    section_list.append('[missing section]')
                section_string = ', '.join(section_list)
                if not error:
                    error = 'Missing value or section.'
                print(section_string, ' = ', error)
            sys.exit(1)

def _validate(config, filename=None):
    """Validate a ``ConfigObj`` object

    In:
      - ``config`` -- the ``ConfigObj`` object, created from the configuration file
      - ``filename`` -- the path to the configuration file

    Return:
      - yield the error messages
    """
    errors = configobj.flatten_errors(config, config.validate(Validator(), preserve_errors=True))

    for sections, name, error in errors:
        yield 'file "%s", section "[%s]", parameter "%s": %s' % (
            filename or '<undefined>',
            ' / '.join(sections),
            name, error
        )

def start(dynamo_region, dynamo_host, table_root, log_config, verbosity, environment, server_config):
    """
    Starts an APScheduler job to periodically reload HAProxy config as well as run the API to register/deregister
    new services, target groups and backends.
    :param fqdn: The fully qualified domain name of Flyby - requests coming here will go to the API endpoints
    :param dynamo_region: The AWS region of the DynamoDB tables Flyby stores and reads config in
    :param dynamo_host: The hostname and port to use for DynamoDB connections. Useful for local testing with
    moto or DynamoDB Local.
    :param table_root: The root that will be used for table names in DynamboDB. This will be prefixed to all tables
    created.
    :param log_config: Location of python yaml config file.
    :param verbosity: Logging verbosity, defaults to INFO.
    :return:
    """
    logging.getLogger().setLevel(level=getattr(logging, verbosity))
    dynamo_manager = DynamoTableManagement()
    config = ConfigObj(infile=server_config, configspec='flyby/configspec.ini', stringify=True)
    res = config.validate(Validator(), preserve_errors=True)
    if res is not True:
        for section, key, msg in flatten_errors(config, res):
            click.echo("{}: {} in {}".format(key, msg, section))
        raise click.ClickException('bad server config')

    # Create the DynamoDB tables if missing, update the DynamoDB read/write capacity if required
    dynamo_manager.update_capacity(dynamo_host, dynamo_region, table_root, logger, config['dynamodb'])
    if log_config:
        with open(log_config, 'r') as conf:
            logging.config.dictConfig(yaml.load(conf))
    scheduler = BackgroundScheduler(timezone=utc)
    scheduler.add_job(update, 'interval', seconds=10, next_run_time=datetime.now(), args=(config['haproxy'],))
    scheduler.start()
    if environment == "development":
        app.run(host='0.0.0.0')
    else:
        serve(app, listen='*:5000')

def _validate(self, config):
        """
        Validate the config against the specification using a default
        validator.  The validated config values are returned if success,
        otherwise, the ``ConfigError`` raised with details.
        """
        validator = Validator()
        try:
            # NOTE:
            # Use the "copy" mode, which will copy both the default values
            # and all the comments.
            results = config.validate(validator, preserve_errors=True,
                                      copy=True)
        except ConfigObjError as e:
            raise ConfigError(e)
        if results is not True:
            error_msg = ""
            for (section_list, key, res) in flatten_errors(config, results):
                if key is not None:
                    if res is False:
                        msg = 'key "%s" in section "%s" is missing.'
                        msg = msg % (key, ", ".join(section_list))
                    else:
                        msg = 'key "%s" in section "%s" failed validation: %s'
                        msg = msg % (key, ", ".join(section_list), res)
                else:
                    msg = 'section "%s" is missing' % ".".join(section_list)
                error_msg += msg + "\n"
            raise ConfigError(error_msg)
        return config

def _register_errors(self, config: dict, validated):
        for [section_list, key, error] in flatten_errors(config, validated):
            if key is not None:
                error = 'Missing' if error is False else str(error)
                self.errors[key] = error

def print_validation_errors(config, results):
    click.secho('Configuration validation failed:', fg='red', bold=True)
    for (section_list, key, _) in configobj.flatten_errors(config, results):
        if key is not None:
            click.secho('- the "%s" key in the section "%s" failed validation'
                        % (key, ', '.join(section_list)), fg='red')
        else:
            click.secho('- the following section was missing:%s ' %
                        ', '.join(section_list), fg='red')

def validate(self):

        #def walk_nested_dict(d):
            #for key1, value1 in d.items():
                #if isinstance(value1, dict):
                    #for key2, value2 in walk_nested_dict(value1):
                        #yield [key1, key2], value2
                #else:
                    #yield [key1,], value1

        for key1, value1 in self.entrydict.items():
            if not isinstance(value1, dict): # shouldn't happen
                if key1.find('Password') == -1:
                    self.settings[key1] = value1.getvalue()
                else:
                    self.settings[key1] = myutils.password_obfuscate(value1.getvalue())
            else:
                for key2, value2 in value1.items():
                    if not isinstance(value2, dict):
                        if key2.find('Password') == -1:
                            self.settings[key1][key2] = value2.getvalue()
                        else:
                            self.settings[key1][key2] = myutils.password_obfuscate(value2.getvalue())
                    else:
                        for key3, value3 in value2.items():
                            if not isinstance(value3, dict):
                                if key3.find('Password') == -1:
                                    self.settings[key1][key2][key3] = value3.getvalue()
                                else:
                                    self.settings[key1][key2][key3] = myutils.password_obfuscate(value3.getvalue())
                            else:
                                pass # shouldn't happen

        errortext=["Some of your input contains errors. "
                    "Detailed error output below.",]

        val = Validator()
        val.functions['log_filename_check'] = myutils.validate_log_filename
        val.functions['image_filename_check'] = myutils.validate_image_filename
        valresult=self.settings.validate(val, preserve_errors=True)
        if valresult != True:
            for section_list, key, error in flatten_errors(self.settings, 
                                                                valresult):
                if key is not None:
                    section_list.append(key)
                else:
                    section_list.append('[missing section]')
                section_string = ','.join(section_list)
                if error == False:
                    error = 'Missing value or section.'
                errortext.append('%s: %s' % (section_string, error))
            tkMessageBox.showerror("Erroneous input. Please try again.", 
                        '\n\n'.join(errortext), parent=self.dialog.interior())
            self.settings = self.read_settings()
            return False
        else:
            return True