Python ruamel.yaml 模块，Loader() 实例源码

def load_config_spec(config_spec, config_sections, repl_vars, language):
    config_split = config_spec.strip().split(':')
    config_path = config_split[0]
    if len(config_split) > 1:
        config_sections = config_split[1].split('|')
    with open(config_path) as config_file:
        all_config_data = yaml.load(config_file, Loader=yaml.Loader)

    # Make a list of the appropriate configuration sections (just the ones
    # we are actually using) from the YAML file.
    segments = [all_config_data[i] for i in config_sections]
    segments.append(all_config_data.get(language, {}))

    # Merge all of the segments of data into a single config dictionary.
    config = merge(*segments)

    # Perform final replacements.
    return replace_vars(config, repl_vars)

def _ordered_load(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict):
    """
    Ordered yaml loader
    Use this instead ot yaml.loader/yaml.saveloader to get an Ordereddict

    :param stream: stream to read from
    :param Loader: yaml-loader to use
    :object_pairs_hook: ...

    :return: OrderedDict structure
    """

    # usage example: ordered_load(stream, yaml.SafeLoader)
    class OrderedLoader(Loader):
        pass
    def construct_mapping(loader, node):
        loader.flatten_mapping(node)
        return object_pairs_hook(loader.construct_pairs(node))
    OrderedLoader.add_constructor(
        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
        construct_mapping)
    return yaml.load(stream, OrderedLoader)

def yaml_load_roundtrip(filename):
    """
    Load contents of a yaml file into an dict structure for editing (using Roundtrip Loader)

    :param filename: name of the yaml file to load
    :return: data structure loaded from file
    """

    if not EDITING_ENABLED:
        return None

    y = None
    try:
        with open(filename+YAML_FILE, 'r') as stream:
            sdata = stream.read()
        sdata = sdata.replace('\n', '\n\n')
        y = yaml.load(sdata, yaml.RoundTripLoader)
    except Exception as e:
        logger.error("yaml_load_roundtrip: YAML-file load error: '%s'" % (e))
        y = {} 
    return y

def import_cwl(self, cwl_path):
        """
        Load content of cwl into the :class:`cwlgen.CommandLineTool` object.

        :param cwl_path: Path of the CWL tool to be loaded.
        :type cwl_path: STRING
        :return: CWL tool content in cwlgen model.
        :rtype: :class:`cwlgen.CommandLineTool`
        """
        with open(cwl_path) as yaml_file:
            cwl_dict = ryaml.load(yaml_file, Loader=ryaml.Loader)
        tool = self._init_tool(cwl_dict)
        for key, element in cwl_dict.items():
            try:
                getattr(self, '_load_{}'.format(key))(tool, element)
            except AttributeError:
                logger.warning(key + " content is not processed (yet).")
        return tool

def read_user_config(flags):
    """Read the user config from disk and return it.

    Args:
        flags (argparse.Namespace): The flags from sys.argv.

    Returns:
        dict: The user config.
    """
    # Load the user configuration if it exists and save a dictionary.
    user_config = {}
    user_config_file = os.path.realpath(os.path.expanduser(flags.user_config))
    if os.path.isfile(user_config_file):
        with io.open(user_config_file) as ucf:
            user_config = yaml.load(ucf.read(), Loader=yaml.Loader) or {}

    # Sanity check: Is there a configuration? If not, abort.
    if not user_config:
        setup_logging(INFO)
        logger.critical('No user configuration found.')
        logger.warn('This is probably your first time running Artman.')
        logger.warn('Run `configure-artman` to get yourself set up.')
        sys.exit(64)

    # Done; return the user config.
    return user_config

def execute(self, gapic_code_dir, grpc_code_dir, proto_code_dir, gapic_api_yaml):
        with open(gapic_api_yaml[0]) as f:
            gapic_config = yaml.load(f, Loader=yaml.Loader)
        package_name = gapic_config.get('language_settings').get('csharp').get('package_name')
        package_root = '{0}/{1}'.format(gapic_code_dir, package_name)
        prod_dir = '{0}/{1}'.format(package_root, package_name)
        # Copy proto/grpc .cs files into prod directory
        self.exec_command(['sh', '-c', 'cp {0}/*.cs {1}'.format(proto_code_dir, prod_dir)])
        self.exec_command(['sh', '-c', 'cp {0}/*.cs {1}'.format(grpc_code_dir, prod_dir)])

def execute(self, src_proto_path, import_proto_path, common_protos_yaml,
                organization_name):
        self._organization_name = organization_name

        with io.open(common_protos_yaml) as file_:
            common_protos_data = yaml.load(file_, Loader=yaml.Loader)

        # Treat google.protobuf, google.iam as a common proto package, even
        # though they are not included in the common-protos we generate.
        #
        # TODO (geigerj): remove 'google.iam' when it is included in the common
        # protos package.
        common_protos = ['google.protobuf', 'google.iam']
        for package in common_protos_data['packages']:
            common_protos.append('google.' + package['name'].replace('/', '.'))

        tmpdir = os.path.join(
            tempfile.gettempdir(), 'artman-python', str(int(time.time())))
        new_proto_dir = os.path.join(tmpdir, 'proto')
        new_src_path = set()
        new_import_path = [new_proto_dir]

        self._copy_and_transform_directories(
            src_proto_path, new_proto_dir, common_protos, paths=new_src_path)
        self._copy_and_transform_directories(
            import_proto_path, new_proto_dir, common_protos)

        # Update src_proto_path, import_proto_path
        return list(new_src_path), new_import_path

def load_config(config_path, loader=yaml.Loader, verify_version=True):
    if not os.path.exists(config_path):
        system_log.error(_("config.yml not found in {config_path}").format(config_path))
        return False
    with codecs.open(config_path, encoding="utf-8") as stream:
        config = yaml.load(stream, loader)
    if verify_version:
        config = config_version_verify(config, config_path)
    return config

def load_config(config_path, loader=yaml.Loader):
    if config_path is None:
        return {}
    if not os.path.exists(config_path):
        system_log.error(_(u"config.yml not found in {config_path}").format(config_path))
        return False
    with codecs.open(config_path, encoding="utf-8") as stream:
        config = yaml.load(stream, loader)
    return config

def load_mod_config(config_path, loader=yaml.Loader):
    mod_config = load_config(config_path, loader)
    if mod_config is None or "mod" not in mod_config:
        import os
        os.remove(config_path)
        config_path = get_mod_config_path()
        return load_mod_config(config_path, loader)
    else:
        return mod_config

def __init__(self, stream, version=None, **kwargs):
        """Initialise Loader."""

        try:
            self._root = os.path.dirname(stream.name)
        except AttributeError:
            self._root = os.path.curdir

        yaml.Loader.__init__(self, stream, version=None, **kwargs)

def construct_include(self, node):
        """Include file referenced at node."""

        filename = os.path.join(self._root, self.construct_scalar(node))
        filename = os.path.abspath(filename)
        extension = os.path.splitext(filename)[1].lstrip('.')

        with open(filename, 'r') as f:
            if extension in ('yaml', 'yml'):
                return yaml.load(f, Loader=self)
            else:
                return ''.join(f.readlines())

def prepare_input(argv=None):
    """
    Get, parse and prepare input file.
    """
    p = ArgumentParser(description='InsiliChem Ommprotocol: '
                       'easy to deploy MD protocols for OpenMM')
    p.add_argument('input', metavar='INPUT FILE', type=extant_file,
                   help='YAML input file')
    p.add_argument('--version', action='version', version='%(prog)s v{}'.format(__version__))
    p.add_argument('-c', '--check', action='store_true',
                   help='Validate input file only')
    args = p.parse_args(argv if argv else sys.argv[1:])

    # Load config file
    with open(args.input) as f:
        cfg = yaml.load(f, Loader=YamlLoader)
    # Paths and dirs
    cfg['_path'] = os.path.abspath(args.input)
    cfg['system_options'] = prepare_system_options(cfg)
    cfg['outputpath'] = sanitize_path_for_file(cfg.get('outputpath', '.'), args.input)

    if not args.check:
        with ignored_exceptions(OSError):
            os.makedirs(cfg['outputpath'])

    handler = prepare_handler(cfg)

    return handler, cfg, args

def _create_config(self, api_name, api_version, api_full_name, output_dir,
                       package_dependencies_yaml, package_defaults_yaml, proto_deps,
                       proto_test_deps, language, local_paths, src_proto_path, package_type,
                       gapic_api_yaml, release_level=None,packaging='single-artifact',
                       generated_package_version=None):
        googleapis_dir = local_paths['googleapis']
        googleapis_path = os.path.commonprefix(
            [os.path.relpath(p, googleapis_dir) for p in src_proto_path])

        with open(package_dependencies_yaml) as dep_file:
            package_dependencies = yaml.load(dep_file, Loader=yaml.Loader)
        with open(package_defaults_yaml) as dep_file:
            package_defaults = yaml.load(dep_file, Loader=yaml.Loader)

        if release_level is not None:
            package_defaults['release_level'][language] = (
                release_level)
        # Apply package version and development status overrides if specified
        # in the artman config
        if generated_package_version is not None:
            release_version_type = package_defaults['release_level'][language]
            if release_version_type != 'ga':
                package_defaults['generated_package_version'][language] = (
                    generated_package_version)
            else:
                package_defaults['generated_ga_package_version'][language] = (
                    generated_package_version)

        gapic_config_name = ''
        if len(gapic_api_yaml) > 0:
            gapic_config_name = os.path.basename(gapic_api_yaml[0])

        dependency_type = 'local'
        if packaging == 'single-artifact':
            dependency_type = 'release'

        config = {
            'short_name': api_name,
            'major_version': api_version,
            'proto_path': googleapis_path,
            'package_name': {
                'default': api_full_name,
            },
            'proto_deps': proto_deps,
            'package_type': package_type,
            'dependency_type': dependency_type,
            'gapic_config_name': gapic_config_name,
        }

        if proto_test_deps:
            config['proto_test_deps'] = proto_test_deps

        config.update(package_dependencies)
        config.update(package_defaults)

        return config

    # Separated so that this can be mocked for testing