GlobFormatter.get_field() - Code Metrics - Inspection of "Implement match()" - jacebrowning/yorm - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — develop (#143)

by Jamie

created 2017-06-01 17:24 UTC

GlobFormatter.get_field() A

↳ Parent: GlobFormatter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	5
ccs	2
cts	2
cp	1
rs	9.4285
cc	2
crap	2

"""Functions to interact with mapped classes and instances."""

import inspect
import logging
import string
import glob
import types

import parse

from . import common, exceptions

log = logging.getLogger(__name__)


def create(class_or_instance, *args, overwrite=False, **kwargs):
    """Create a new mapped object.

    NOTE: Calling this function is unnecessary with 'auto_create' enabled.

    """
    instance = _instantiate(class_or_instance, *args, **kwargs)
    mapper = common.get_mapper(instance, expected=True)

    if mapper.exists and not overwrite:
        msg = "{!r} already exists".format(mapper.path)
        raise exceptions.DuplicateMappingError(msg)

    return load(save(instance))


def find(class_or_instance, *args, create=False, **kwargs):  # pylint: disable=redefined-outer-name
    """Find a matching mapped object or return None."""
    instance = _instantiate(class_or_instance, *args, **kwargs)
    mapper = common.get_mapper(instance, expected=True)

    if mapper.exists:
        return instance
    elif create:
        return save(instance)
    else:
        return None


class GlobFormatter(string.Formatter):
    """
    Uses '*' for all unknown fields
    """

    WILDCARD = object()

    def get_field(self, field_name, args, kwargs):
        try:
            return super().get_field(field_name, args, kwargs)
        except (KeyError, IndexError, AttributeError):
            return self.WILDCARD, None

    def get_value(self, key, args, kwargs):
        try:
            return super().get_value(key, args, kwargs)
        except (KeyError, IndexError, AttributeError):
            return self.WILDCARD

    def convert_field(self, value, conversion):
        if value is self.WILDCARD:
            return self.WILDCARD
        else:
            return super().convert_field(value, conversion)

    def format_field(self, value, format_spec):
        if value is self.WILDCARD:
            return '*'
        else:
            return super().format_field(value, format_spec)


def _unpack_parsed_fields(pathfields):
    return {
        (k[len('self.'):] if k.startswith('self.') else k): v
        for k, v in pathfields.items()
    }


def match(cls_or_path, _factory=None, **kwargs):
    """match(class, [callable], ...) -> instance, ...
    match(str, callable, ...) -> instance, ...

    Yield all matching mapped objects. Can be used two ways:
    * With a YORM-decorated class, optionally with a factory callable
    * With a Python 3-style string template and a factory callable

    The factory callable must accept keyuword arguments, extracted from the file
    name merged with those passed to match(). If no factory is given, the class
    itself is used as the factory (same signature).

    Keyword arguments are used to filter objects. Filtering is only done by
    filename, so only fields that are part of the path_format can be filtered
    against.
    """
    if isinstance(cls_or_path, type):
        path_format = common.path_formats[cls_or_path]
        # Let KeyError fail through
        if _factory is None:
            _factory = cls_or_path
    else:
        path_format = cls_or_path
        if _factory is None:
            raise TypeError("Factory must be given if a path format is given")

    gf = GlobFormatter()
    mock = types.SimpleNamespace(**kwargs)

    kwargs['self'] = mock
    posix_pattern = gf.vformat(path_format, (), kwargs.copy())
    del kwargs['self']
    py_pattern = parse.compile(path_format)

    for filename in glob.iglob(posix_pattern):
        pathfields = py_pattern.parse(filename).named
        fields = _unpack_parsed_fields(pathfields)
        fields.update(kwargs)
        yield _factory(**fields)


def load(instance):
    """Force the loading of a mapped object's file.

    NOTE: Calling this function is unnecessary. It exists for the
        aesthetic purpose of having symmetry between save and load.

    """
    mapper = common.get_mapper(instance, expected=True)

    mapper.load()

    return instance


def save(instance):
    """Save a mapped object to file.

    NOTE: Calling this function is unnecessary with 'auto_save' enabled.

    """
    mapper = common.get_mapper(instance, expected=True)

    if mapper.deleted:
        msg = "{!r} was deleted".format(mapper.path)
        raise exceptions.DeletedFileError(msg)

    if not mapper.exists:
        mapper.create()

    mapper.save()

    return instance


def delete(instance):
    """Delete a mapped object's file."""
    mapper = common.get_mapper(instance, expected=True)

    mapper.delete()

    return None


def _instantiate(class_or_instance, *args, **kwargs):
    if inspect.isclass(class_or_instance):
        instance = class_or_instance(*args, **kwargs)
    else:
        assert not args
        instance = class_or_instance

    return instance


1		"""Functions to interact with mapped classes and instances."""
2
3	1	import inspect
4	1	import logging
5		import string
6	1	import glob
7		import types
8	1
9		import parse
10
11	1	from . import common, exceptions
12
13		log = logging.getLogger(__name__)
14
15
16		def create(class_or_instance, args, overwrite=False, *kwargs):
17	1	"""Create a new mapped object.
18	1
19		NOTE: Calling this function is unnecessary with 'auto_create' enabled.
20	1
21	1	"""
22	1	instance = _instantiate(class_or_instance, args, *kwargs)
23		mapper = common.get_mapper(instance, expected=True)
24	1
25		if mapper.exists and not overwrite:
26		msg = "{!r} already exists".format(mapper.path)
27	1	raise exceptions.DuplicateMappingError(msg)
28
29	1	return load(save(instance))
30	1
31
32	1	def find(class_or_instance, args, create=False, *kwargs): # pylint: disable=redefined-outer-name
33	1	"""Find a matching mapped object or return None."""
34	1	instance = _instantiate(class_or_instance, args, *kwargs)
35	1	mapper = common.get_mapper(instance, expected=True)
36
37	1	if mapper.exists:
38		return instance
39		elif create:
40	1	return save(instance)
41		else:
42	1	return None
43	1
44
45		class GlobFormatter(string.Formatter):
46	1	"""
47		Uses '*' for all unknown fields
48		"""
49
50		WILDCARD = object()
51
52		def get_field(self, field_name, args, kwargs):
53	1	try:
54		return super().get_field(field_name, args, kwargs)
55	1	except (KeyError, IndexError, AttributeError):
56		return self.WILDCARD, None
57	1
58		def get_value(self, key, args, kwargs):
59		try:
60	1	return super().get_value(key, args, kwargs)
61		except (KeyError, IndexError, AttributeError):
62		return self.WILDCARD
63
64		def convert_field(self, value, conversion):
65		if value is self.WILDCARD:
66	1	return self.WILDCARD
67		else:
68	1	return super().convert_field(value, conversion)
69	1
70	1	def format_field(self, value, format_spec):
71		if value is self.WILDCARD:
72	1	return '*'
73	1	else:
74		return super().format_field(value, format_spec)
75	1
76
77	1	def _unpack_parsed_fields(pathfields):
78		return {
79		(k[len('self.'):] if k.startswith('self.') else k): v
80	1	for k, v in pathfields.items()
81		}
82	1
83
84	1	def match(cls_or_path, _factory=None, **kwargs):
85		"""match(class, [callable], ...) -> instance, ...
86	1	match(str, callable, ...) -> instance, ...
87
88		Yield all matching mapped objects. Can be used two ways:
89	1	* With a YORM-decorated class, optionally with a factory callable
90	1	* With a Python 3-style string template and a factory callable
91	1
92		The factory callable must accept keyuword arguments, extracted from the file
93	1	name merged with those passed to match(). If no factory is given, the class
94	1	itself is used as the factory (same signature).
95
96	1	Keyword arguments are used to filter objects. Filtering is only done by
97		filename, so only fields that are part of the path_format can be filtered
98		against.
99		"""
100		if isinstance(cls_or_path, type):
101		path_format = common.path_formats[cls_or_path]
102		# Let KeyError fail through
103		if _factory is None:
104		_factory = cls_or_path
105		else:
106		path_format = cls_or_path
107		if _factory is None:
108		raise TypeError("Factory must be given if a path format is given")
109
110		gf = GlobFormatter()
111		mock = types.SimpleNamespace(**kwargs)
112
113		kwargs['self'] = mock
114		posix_pattern = gf.vformat(path_format, (), kwargs.copy())
115		del kwargs['self']
116		py_pattern = parse.compile(path_format)
117
118		for filename in glob.iglob(posix_pattern):
119		pathfields = py_pattern.parse(filename).named
120		fields = _unpack_parsed_fields(pathfields)
121		fields.update(kwargs)
122		yield _factory(**fields)
123
124
125		def load(instance):
126		"""Force the loading of a mapped object's file.
127
128		NOTE: Calling this function is unnecessary. It exists for the
129		aesthetic purpose of having symmetry between save and load.
130
131		"""
132		mapper = common.get_mapper(instance, expected=True)
133
134		mapper.load()
135
136		return instance
137
138
139		def save(instance):
140		"""Save a mapped object to file.
141
142		NOTE: Calling this function is unnecessary with 'auto_save' enabled.
143
144		"""
145		mapper = common.get_mapper(instance, expected=True)
146
147		if mapper.deleted:
148		msg = "{!r} was deleted".format(mapper.path)
149		raise exceptions.DeletedFileError(msg)
150
151		if not mapper.exists:
152		mapper.create()
153
154		mapper.save()
155
156		return instance
157
158
159		def delete(instance):
160		"""Delete a mapped object's file."""
161		mapper = common.get_mapper(instance, expected=True)
162
163		mapper.delete()
164
165		return None
166
167
168		def _instantiate(class_or_instance, args, *kwargs):
169		if inspect.isclass(class_or_instance):
170		instance = class_or_instance(args, *kwargs)
171		else:
172		assert not args
173		instance = class_or_instance
174
175		return instance
176

jacebrowning / yorm

Pull Request — develop (#143)

GlobFormatter.get_field() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like