pred() - Code Metrics - Inspection of "fix: remove the debug messaging part in .ioinfo.ma..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ef7389...74fe88 )

by Satoru

created 2018-07-04 11:48 UTC

pred() A

↳ Parent: find_by_type()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
c	0
b	0
f	0
dl	0
loc	3
rs	10

#
# Copyright (C) 2018 Satoru SATOH <ssato @ redhat.com>
# License: MIT
#
r"""Abstract processor module.

.. versionadded:: 0.9.5

   - Add to abstract processors such like Parsers (loaders and dumpers).
"""
from __future__ import absolute_import

import operator
import pkg_resources

import anyconfig.compat
import anyconfig.utils

from anyconfig.globals import (
    UnknownProcessorTypeError, UnknownFileTypeError
)


class Processor(object):
    """
    Abstract processor class to provide basic implementation of some methods,
    interfaces and members.

    - _type: type indicates data types it can process
    - _priority: Priority to select it if there are others of same type
    - _extensions: File extensions of data type it can process
    """
    _type = None
    _priority = 0   # 0 (lowest priority) .. 99  (highest priority)
    _extensions = []

    @classmethod
    def type(cls):
        """Processors' type
        """
        return cls._type

    @classmethod
    def priority(cls):
        """Processors's priority
        """
        return cls._priority

    @classmethod
    def extensions(cls):
        """File extensions which this process can process
        """
        return cls._extensions


def _load_plugins_itr(pgroup, safe=True):
    """
    .. seealso:: the doc of :func:`load_plugins`
    """
    for res in pkg_resources.iter_entry_points(pgroup):
        try:
            yield res.load()
        except ImportError:
            if safe:
                continue
            raise


def load_plugins(pgroup, safe=True):
    """
    :param pgroup: A string represents plugin type, e.g. anyconfig_backends
    :param safe: Do not raise ImportError during load if True
    :raises: ImportError
    """
    return list(_load_plugins_itr(pgroup, safe=safe))


def find_with_pred(predicate, prs):
    """
    :param predicate: any callable to filter results
    :param prs: A list of :class:`Processor` classes
    :return: Most appropriate processor class or None

    >>> class A(Processor):
    ...    _type = "json"
    ...    _extensions = ['json', 'js']
    >>> class A2(A):
    ...    _priority = 99  # Higher priority than A.
    >>> class B(Processor):
    ...    _type = "yaml"
    ...    _extensions = ['yaml', 'yml']
    >>> prs = [A, A2, B]

    >>> find_with_pred(lambda p: 'js' in p.extensions(), prs)
    <class 'anyconfig.processors.A2'>
    >>> find_with_pred(lambda p: 'yml' in p.extensions(), prs)
    <class 'anyconfig.processors.B'>
    >>> x = find_with_pred(lambda p: 'xyz' in p.extensions(), prs)
    >>> assert x is None

    >>> find_with_pred(lambda p: p.type() == "json", prs)
    <class 'anyconfig.processors.A2'>
    >>> find_with_pred(lambda p: p.type() == "yaml", prs)
    <class 'anyconfig.processors.B'>
    >>> x = find_with_pred(lambda p: p.type() == "x", prs)
    >>> assert x is None
    """
    _prs = sorted((p for p in prs if predicate(p)),
                  key=operator.methodcaller("priority"), reverse=True)
    if _prs:
        return _prs[0]  # Found.

    return None


def find_by_type(ptype, prs):
    """
    :param ptype: Type of the data to process
    :param prs: A list of :class:`Processor` classes
    :return:
        Most appropriate processor class to process files of given data type
        `ptype` or None
    :raises: UnknownProcessorTypeError
    """
    def pred(pcls):
        """Predicate"""
        return pcls.type() == ptype

    processor = find_with_pred(pred, prs)
    if processor is None:
        raise UnknownProcessorTypeError(ptype)

    return processor


def find_by_fileext(fileext, prs):
    """
    :param fileext: File extension
    :param prs: A list of :class:`Processor` classes
    :return:
        Most appropriate processor class to process files with given
        extentsions or None
    :raises: UnknownFileTypeError
    """
    def pred(pcls):
        """Predicate"""
        return fileext in pcls.extensions()

    return find_with_pred(pred, prs)


def find_by_filepath(filepath, prs):
    """
    :param filepath: Path to the file to find out processor to process it
    :param cps_by_ext: A list of processor classes

    :return: Most appropriate processor class to process given file
    :raises: UnknownFileTypeError
    """
    fileext = anyconfig.utils.get_file_extension(filepath)
    processor = find_by_fileext(fileext, prs)
    if processor is None:
        raise UnknownFileTypeError(fileext)

    return processor


def find(ipath, prs, forced_type=None):
    """
    :param ipath: file path
    :param prs: A list of processor classes
    :param forced_type: Forced processor type or processor object itself

    :return: an instance of processor class appropriate to process `ipath` data
    :raises: ValueError, UnknownProcessorTypeError, UnknownFileTypeError
    """
    if (ipath is None or not ipath) and forced_type is None:
        raise ValueError("file path or file type must be given")

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

        return processor()

    elif isinstance(forced_type, Processor):
        return forced_type

    elif type(forced_type) == type(Processor):
        return forced_type()

    processor = find_by_type(forced_type, prs)
    if processor is None:
        raise UnknownProcessorTypeError(forced_type)

    return processor()

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


1			#
2			# Copyright (C) 2018 Satoru SATOH <ssato @ redhat.com>
3			# License: MIT
4			#
5			r"""Abstract processor module.
6
7			.. versionadded:: 0.9.5
8
9			- Add to abstract processors such like Parsers (loaders and dumpers).
10			"""
11			from __future__ import absolute_import
12
13			import operator
14			import pkg_resources
15
16			import anyconfig.compat
17			import anyconfig.utils
18
19			from anyconfig.globals import (
20			UnknownProcessorTypeError, UnknownFileTypeError
21			)
22
23
24			class Processor(object):
25			"""
26			Abstract processor class to provide basic implementation of some methods,
27			interfaces and members.
28
29			- _type: type indicates data types it can process
30			- _priority: Priority to select it if there are others of same type
31			- _extensions: File extensions of data type it can process
32			"""
33			_type = None
34			_priority = 0 # 0 (lowest priority) .. 99 (highest priority)
35			_extensions = []
36
37			@classmethod
38			def type(cls):
39			"""Processors' type
40			"""
41			return cls._type
42
43			@classmethod
44			def priority(cls):
45			"""Processors's priority
46			"""
47			return cls._priority
48
49			@classmethod
50			def extensions(cls):
51			"""File extensions which this process can process
52			"""
53			return cls._extensions
54
55
56			def _load_plugins_itr(pgroup, safe=True):
57			"""
58			.. seealso:: the doc of :func:`load_plugins`
59			"""
60			for res in pkg_resources.iter_entry_points(pgroup):
61			try:
62			yield res.load()
63			except ImportError:
64			if safe:
65			continue
66			raise
67
68
69			def load_plugins(pgroup, safe=True):
70			"""
71			:param pgroup: A string represents plugin type, e.g. anyconfig_backends
72			:param safe: Do not raise ImportError during load if True
73			:raises: ImportError
74			"""
75			return list(_load_plugins_itr(pgroup, safe=safe))
76
77
78			def find_with_pred(predicate, prs):
79			"""
80			:param predicate: any callable to filter results
81			:param prs: A list of :class:`Processor` classes
82			:return: Most appropriate processor class or None
83
84			>>> class A(Processor):
85			... _type = "json"
86			... _extensions = ['json', 'js']
87			>>> class A2(A):
88			... _priority = 99 # Higher priority than A.
89			>>> class B(Processor):
90			... _type = "yaml"
91			... _extensions = ['yaml', 'yml']
92			>>> prs = [A, A2, B]
93
94			>>> find_with_pred(lambda p: 'js' in p.extensions(), prs)
95			<class 'anyconfig.processors.A2'>
96			>>> find_with_pred(lambda p: 'yml' in p.extensions(), prs)
97			<class 'anyconfig.processors.B'>
98			>>> x = find_with_pred(lambda p: 'xyz' in p.extensions(), prs)
99			>>> assert x is None
100
101			>>> find_with_pred(lambda p: p.type() == "json", prs)
102			<class 'anyconfig.processors.A2'>
103			>>> find_with_pred(lambda p: p.type() == "yaml", prs)
104			<class 'anyconfig.processors.B'>
105			>>> x = find_with_pred(lambda p: p.type() == "x", prs)
106			>>> assert x is None
107			"""
108			_prs = sorted((p for p in prs if predicate(p)),
109			key=operator.methodcaller("priority"), reverse=True)
110			if _prs:
111			return _prs[0] # Found.
112
113			return None
114
115
116			def find_by_type(ptype, prs):
117			"""
118			:param ptype: Type of the data to process
119			:param prs: A list of :class:`Processor` classes
120			:return:
121			Most appropriate processor class to process files of given data type
122			`ptype` or None
123			:raises: UnknownProcessorTypeError
124			"""
125			def pred(pcls):
126			"""Predicate"""
127			return pcls.type() == ptype
128
129			processor = find_with_pred(pred, prs)
130			if processor is None:
131			raise UnknownProcessorTypeError(ptype)
132
133			return processor
134
135
136			def find_by_fileext(fileext, prs):
137			"""
138			:param fileext: File extension
139			:param prs: A list of :class:`Processor` classes
140			:return:
141			Most appropriate processor class to process files with given
142			extentsions or None
143			:raises: UnknownFileTypeError
144			"""
145			def pred(pcls):
146			"""Predicate"""
147			return fileext in pcls.extensions()
148
149			return find_with_pred(pred, prs)
150
151
152			def find_by_filepath(filepath, prs):
153			"""
154			:param filepath: Path to the file to find out processor to process it
155			:param cps_by_ext: A list of processor classes
156
157			:return: Most appropriate processor class to process given file
158			:raises: UnknownFileTypeError
159			"""
160			fileext = anyconfig.utils.get_file_extension(filepath)
161			processor = find_by_fileext(fileext, prs)
162			if processor is None:
163			raise UnknownFileTypeError(fileext)
164
165			return processor
166
167
168			def find(ipath, prs, forced_type=None):
169			"""
170			:param ipath: file path
171			:param prs: A list of processor classes
172			:param forced_type: Forced processor type or processor object itself
173
174			:return: an instance of processor class appropriate to process `ipath` data
175			:raises: ValueError, UnknownProcessorTypeError, UnknownFileTypeError
176			"""
177			if (ipath is None or not ipath) and forced_type is None:
178			raise ValueError("file path or file type must be given")
179
180			if forced_type is None:
181			processor = find_by_filepath(ipath, prs)
182			if processor is None:
183			raise UnknownFileTypeError(ipath)
184
185			return processor()
186
187			elif isinstance(forced_type, Processor):
188			return forced_type
189
190			elif type(forced_type) == type(Processor):
191			return forced_type()
192
193			processor = find_by_type(forced_type, prs)
194			if processor is None:
195			raise UnknownProcessorTypeError(forced_type)
196
197			return processor()
198
199			# vim:sw=4:ts=4:et:
200

ssato / python-anyconfig

Push — master ( ef7389...74fe88 )

pred() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like