find_processor() - Code Metrics - Inspection of "change: rename IOIInfo.parser to IOIInfo.processor..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 52e99c...3e76cf )

by Satoru

created 2018-06-15 11:39 UTC

find_processor() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
dl	0
loc	28
rs	4
c	0
b	0
f	0

#
# Copyright (C) 2018 Satoru SATOH <ssato @ redhat.com>
# License: MIT
#
# pylint: disable=invalid-name
r"""Functions for value objects represent inputs and outputs.

.. versionadded:: 0.9.5

- Add functions to make and process input and output object holding some
  attributes like input and output type (path, stream or pathlib.Path object),
  path, opener, etc.
"""
from __future__ import absolute_import

import anyconfig.utils

from anyconfig.globals import (
    IOInfo, IOI_NONE, IOI_PATH_STR, IOI_PATH_OBJ, IOI_STREAM,
    UnknownFileTypeError, UnknownParserTypeError
)


def guess_io_type(obj):
    """Guess input or output type of ``obj``.

    :param obj: a path string, a pathlib.Path or a file / file-like object
    :return: IOInfo type defined in anyconfig.globals.IOI_TYPES

    >>> apath = "/path/to/a_conf.ext"
    >>> assert guess_io_type(apath) == IOI_PATH_STR

    >>> from anyconfig.compat import pathlib
    >>> if pathlib is not None:
    ...     assert guess_io_type(pathlib.Path(apath)) == IOI_PATH_OBJ
    >>> assert guess_io_type(open(__file__)) == IOI_STREAM
    """
    if obj is None:
        return IOI_NONE
    elif anyconfig.utils.is_path(obj):
        return IOI_PATH_STR
    elif anyconfig.utils.is_path_obj(obj):
        return IOI_PATH_OBJ

    return IOI_STREAM


def inspect_io_obj(obj):
    """
    :param obj: a path string, a pathlib.Path or a file / file-like object

    :return: A tuple of (objtype, objpath, objopener)
    :raises: UnknownFileTypeError
    """
    itype = guess_io_type(obj)

    if itype == IOI_PATH_STR:
        ipath = anyconfig.utils.normpath(obj)
        opener = open
    elif itype == IOI_PATH_OBJ:
        ipath = anyconfig.utils.normpath(obj.as_posix())
        opener = obj.open
    elif itype == IOI_STREAM:
        ipath = anyconfig.utils.get_path_from_stream(obj)
        opener = anyconfig.utils.noop
    elif itype == IOI_NONE:
        ipath = None
        opener = anyconfig.utils.noop
    else:
        raise UnknownFileTypeError("%r" % obj)

    return (itype, ipath, opener)


def find_by_fileext(fileext, cps_by_ext):
    """
    :param fileext: File extension
    :param cps_by_ext: A list of pairs (file_extension, [processor_class])

    :return: Most appropriate processor class to process given file

    >>> from anyconfig.backends import _PARSERS_BY_EXT as cps
    >>> find_by_fileext("json", cps)
    <class 'anyconfig.backend.json.Parser'>
    >>> find_by_fileext("ext_should_not_be_found", cps) is None
    True
    """
    return next((psrs[-1] for ext, psrs in cps_by_ext if ext == fileext),
                None)


def find_by_filepath(filepath, cps_by_ext):
    """
    :param filepath: Path to the file to find out processor to process it
    :param cps_by_ext: A list of pairs (file_extension, [processor_class])

    :return: Most appropriate processor class to process given file

    >>> from anyconfig.backends import _PARSERS_BY_EXT as cps
    >>> find_by_filepath("/a/b/c/x.json", cps)
    <class 'anyconfig.backend.json.Parser'>
    >>> find_by_filepath("/path/to/a.ext_should_not_be_found", cps) is None
    True
    """
    fileext = anyconfig.utils.get_file_extension(filepath)
    return find_by_fileext(fileext, cps_by_ext)


def find_by_type(cptype, cps_by_type):
    """
    :param cptype: Config file's type
    :param cps_by_type: A list of pairs, (processor_type, [processor_class])

    :return: Most appropriate processor class to process given type or None

    >>> from anyconfig.backends import _PARSERS_BY_TYPE as cps
    >>> find_by_type("json", cps)
    <class 'anyconfig.backend.json.Parser'>
    >>> find_by_type("missing_type", cps) is None
    True
    """
    return next((psrs[-1] or None for t, psrs in cps_by_type if t == cptype),
                None)


def find_processor(ipath, cps_by_ext, cps_by_type, forced_type=None):
    """
    :param ipath: file path
    :param cps_by_ext: A list of pairs (file_extension, [processor_cls])
    :param cps_by_type: A list of pairs (processor_type, [processor_cls])
    :param forced_type: Forced processor type or processor object

    :return: Instance of processor class appropriate for the input `ipath`
    :raises: ValueError, UnknownParserTypeError, UnknownFileTypeError
    """
    if (ipath is None or not ipath) and forced_type is None:
        raise ValueError("ipath or forced_type must be some value")

    if isinstance(forced_type, anyconfig.backend.base.Parser):
        return forced_type

    if forced_type is None:
        processor = find_by_filepath(ipath, cps_by_ext)
        if processor is None:
            raise UnknownFileTypeError(ipath)

        return processor()

    processor = find_by_type(forced_type, cps_by_type)
    if processor is None:
        raise UnknownParserTypeError(forced_type)

    return processor()


def make(obj, cps_by_ext, cps_by_type, forced_type=None):
    """
    :param obj: a path string, a pathlib.Path or a file / file-like object
    :param cps_by_ext: A list of pairs (file_extension, [processor_cls])
    :param cps_by_type: A list of pairs (processor_type, [processor_cls])
    :param forced_type: Forced configuration processor type

    :return:
        Namedtuple object represents a kind of input object such as a file /
        file-like object, path string or pathlib.Path object

    :raises: ValueError, UnknownParserTypeError, UnknownFileTypeError
    """
    if anyconfig.utils.is_ioinfo(obj):
        return obj

    if (obj is None or not obj) and forced_type is None:
        raise ValueError("obj or forced_type must be some value")

    (itype, ipath, opener) = inspect_io_obj(obj)
    psr = find_processor(ipath, cps_by_ext, cps_by_type,
                         forced_type=forced_type)

    return IOInfo(src=obj, type=itype, path=ipath, processor=psr,
                  opener=opener)

# vim:sw=4:ts=4:et:


1			#
2			# Copyright (C) 2018 Satoru SATOH <ssato @ redhat.com>
3			# License: MIT
4			#
5			# pylint: disable=invalid-name
6			r"""Functions for value objects represent inputs and outputs.
7
8			.. versionadded:: 0.9.5
9
10			- Add functions to make and process input and output object holding some
11			attributes like input and output type (path, stream or pathlib.Path object),
12			path, opener, etc.
13			"""
14			from __future__ import absolute_import
15
16			import anyconfig.utils
17
18			from anyconfig.globals import (
19			IOInfo, IOI_NONE, IOI_PATH_STR, IOI_PATH_OBJ, IOI_STREAM,
20			UnknownFileTypeError, UnknownParserTypeError
21			)
22
23
24			def guess_io_type(obj):
25			"""Guess input or output type of ``obj``.
26
27			:param obj: a path string, a pathlib.Path or a file / file-like object
28			:return: IOInfo type defined in anyconfig.globals.IOI_TYPES
29
30			>>> apath = "/path/to/a_conf.ext"
31			>>> assert guess_io_type(apath) == IOI_PATH_STR
32
33			>>> from anyconfig.compat import pathlib
34			>>> if pathlib is not None:
35			... assert guess_io_type(pathlib.Path(apath)) == IOI_PATH_OBJ
36			>>> assert guess_io_type(open(__file__)) == IOI_STREAM
37			"""
38			if obj is None:
39			return IOI_NONE
40			elif anyconfig.utils.is_path(obj):
41			return IOI_PATH_STR
42			elif anyconfig.utils.is_path_obj(obj):
43			return IOI_PATH_OBJ
44
45			return IOI_STREAM
46
47
48			def inspect_io_obj(obj):
49			"""
50			:param obj: a path string, a pathlib.Path or a file / file-like object
51
52			:return: A tuple of (objtype, objpath, objopener)
53			:raises: UnknownFileTypeError
54			"""
55			itype = guess_io_type(obj)
56
57			if itype == IOI_PATH_STR:
58			ipath = anyconfig.utils.normpath(obj)
59			opener = open
60			elif itype == IOI_PATH_OBJ:
61			ipath = anyconfig.utils.normpath(obj.as_posix())
62			opener = obj.open
63			elif itype == IOI_STREAM:
64			ipath = anyconfig.utils.get_path_from_stream(obj)
65			opener = anyconfig.utils.noop
66			elif itype == IOI_NONE:
67			ipath = None
68			opener = anyconfig.utils.noop
69			else:
70			raise UnknownFileTypeError("%r" % obj)
71
72			return (itype, ipath, opener)
73
74
75			def find_by_fileext(fileext, cps_by_ext):
76			"""
77			:param fileext: File extension
78			:param cps_by_ext: A list of pairs (file_extension, [processor_class])
79
80			:return: Most appropriate processor class to process given file
81
82			>>> from anyconfig.backends import _PARSERS_BY_EXT as cps
83			>>> find_by_fileext("json", cps)
84			<class 'anyconfig.backend.json.Parser'>
85			>>> find_by_fileext("ext_should_not_be_found", cps) is None
86			True
87			"""
88			return next((psrs[-1] for ext, psrs in cps_by_ext if ext == fileext),
89			None)
90
91
92			def find_by_filepath(filepath, cps_by_ext):
93			"""
94			:param filepath: Path to the file to find out processor to process it
95			:param cps_by_ext: A list of pairs (file_extension, [processor_class])
96
97			:return: Most appropriate processor class to process given file
98
99			>>> from anyconfig.backends import _PARSERS_BY_EXT as cps
100			>>> find_by_filepath("/a/b/c/x.json", cps)
101			<class 'anyconfig.backend.json.Parser'>
102			>>> find_by_filepath("/path/to/a.ext_should_not_be_found", cps) is None
103			True
104			"""
105			fileext = anyconfig.utils.get_file_extension(filepath)
106			return find_by_fileext(fileext, cps_by_ext)
107
108
109			def find_by_type(cptype, cps_by_type):
110			"""
111			:param cptype: Config file's type
112			:param cps_by_type: A list of pairs, (processor_type, [processor_class])
113
114			:return: Most appropriate processor class to process given type or None
115
116			>>> from anyconfig.backends import _PARSERS_BY_TYPE as cps
117			>>> find_by_type("json", cps)
118			<class 'anyconfig.backend.json.Parser'>
119			>>> find_by_type("missing_type", cps) is None
120			True
121			"""
122			return next((psrs[-1] or None for t, psrs in cps_by_type if t == cptype),
123			None)
124
125
126			def find_processor(ipath, cps_by_ext, cps_by_type, forced_type=None):
127			"""
128			:param ipath: file path
129			:param cps_by_ext: A list of pairs (file_extension, [processor_cls])
130			:param cps_by_type: A list of pairs (processor_type, [processor_cls])
131			:param forced_type: Forced processor type or processor object
132
133			:return: Instance of processor class appropriate for the input `ipath`
134			:raises: ValueError, UnknownParserTypeError, UnknownFileTypeError
135			"""
136			if (ipath is None or not ipath) and forced_type is None:
137			raise ValueError("ipath or forced_type must be some value")
138
139			if isinstance(forced_type, anyconfig.backend.base.Parser):
140			return forced_type
141
142			if forced_type is None:
143			processor = find_by_filepath(ipath, cps_by_ext)
144			if processor is None:
145			raise UnknownFileTypeError(ipath)
146
147			return processor()
148
149			processor = find_by_type(forced_type, cps_by_type)
150			if processor is None:
151			raise UnknownParserTypeError(forced_type)
152
153			return processor()
154
155
156			def make(obj, cps_by_ext, cps_by_type, forced_type=None):
157			"""
158			:param obj: a path string, a pathlib.Path or a file / file-like object
159			:param cps_by_ext: A list of pairs (file_extension, [processor_cls])
160			:param cps_by_type: A list of pairs (processor_type, [processor_cls])
161			:param forced_type: Forced configuration processor type
162
163			:return:
164			Namedtuple object represents a kind of input object such as a file /
165			file-like object, path string or pathlib.Path object
166
167			:raises: ValueError, UnknownParserTypeError, UnknownFileTypeError
168			"""
169			if anyconfig.utils.is_ioinfo(obj):
170			return obj
171
172			if (obj is None or not obj) and forced_type is None:
173			raise ValueError("obj or forced_type must be some value")
174
175			(itype, ipath, opener) = inspect_io_obj(obj)
176			psr = find_processor(ipath, cps_by_ext, cps_by_type,
177			forced_type=forced_type)
178
179			return IOInfo(src=obj, type=itype, path=ipath, processor=psr,
180			opener=opener)
181
182			# vim:sw=4:ts=4:et:
183

ssato / python-anyconfig

Push — master ( 52e99c...3e76cf )

find_processor() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like