yorm.decorator() - Code Metrics - Inspection of "Apply latest pep257 and pylint standards" - jacebrowning/yorm - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( 52a881...b9d7ca )

by Jace

created 2015-12-18 04:27 UTC

yorm.decorator() B

↳ Parent: yorm.file_required()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	11
CRAP Score	7

Metric	Value
cc	7
dl	0
loc	15
ccs	11
cts	11
cp	1
crap	7
rs	7.3333

"""Core YAML mapping functionality."""

import os
import abc
import functools

import yaml
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from . import common, exceptions, settings
from .bases import Container

MAPPER = '__mapper__'

log = common.logger(__name__)


def get_mapper(obj):
    """Get `Mapper` instance attached to an object."""
    try:
        mapper = getattr(obj, MAPPER)
    except AttributeError:
        if isinstance(obj, (dict, list)):
            return None
        else:
            msg = "mapped {!r} missing {!r} attribute".format(obj, MAPPER)
            raise AttributeError(msg) from None
    else:
        return mapper


def set_mapper(obj, path, attrs, auto=True, root=None):
    """Create and attach a `Mapper` instance to an object."""
    mapper = Mapper(obj, path, attrs, auto=auto, root=root)
    setattr(obj, MAPPER, mapper)
    return mapper


def file_required(create=False):
    """Decorator for methods that require the file to exist.

    :param create: boolean or the method to decorate

    """
    def decorator(method):

        @functools.wraps(method)
        def wrapped(self, *args, **kwargs):  # pylint: disable=W0621
            if not self.path:
                return None
            if not self.exists and self.auto:
                if create is True and not self.deleted:
                    self.create()
                else:
                    msg = "cannot access deleted: {}".format(self.path)
                    raise exceptions.FileDeletedError(msg)
            return method(self, *args, **kwargs)

        return wrapped

    if callable(create):
        return decorator(create)
    else:
        return decorator


def prevent_recursion(method):
    """Decorator to prevent indirect recursive calls."""

    @functools.wraps(method)
    def wrapped(self, *args, **kwargs):
        # pylint: disable=W0212
        if self._activity:
            return
        self._activity = True
        result = method(self, *args, **kwargs)
        self._activity = False
        return result

    return wrapped


def prefix(obj):
    """Prefix a string with a fake designator if enabled."""
    fake = "(fake) " if settings.fake else ""
    name = obj if isinstance(obj, str) else "'{}'".format(obj)
    return fake + name


class BaseHelper(metaclass=abc.ABCMeta):
    """Utility class to map attributes to text files.

    To start mapping attributes to a file:

        create -> [empty] -> FILE

    When getting an attribute:

        FILE -> read -> [text] -> load -> [dict] -> fetch -> ATTRIBUTES

    When setting an attribute:

        ATTRIBUTES -> store -> [dict] -> dump -> [text] -> write -> FILE

    After the mapped file is no longer needed:

        delete -> [null] -> FILE

    """

    def __init__(self, path, auto=True):
        self.path = path
        self.auto = auto
        self.auto_store = False
        self.exists = self.path and os.path.isfile(self.path)
        self.deleted = False
        self._activity = False
        self._timestamp = 0
        self._fake = ""

    def __str__(self):
        return str(self.path)

    @property
    def text(self):
        """Get file contents."""
        log.info("getting contents of %s...", prefix(self))
        if settings.fake:
            text = self._fake
        else:
            text = self._read()
        log.trace("text read: %r", text)
        return text

    @text.setter
    def text(self, text):
        """Set file contents."""
        log.info("setting contents of %s...", prefix(self))
        if settings.fake:
            self._fake = text
        else:
            self._write(text)
        log.trace("text wrote: %r", text)
        self.modified = True

    @property
    def modified(self):
        """Determine if the file has been modified."""
        if settings.fake:
            changes = self._timestamp is not None
            return changes
        elif not self.exists:
            return True
        else:
            was = self._timestamp
            now = common.stamp(self.path)
            return was != now

    @modified.setter
    def modified(self, changes):
        """Mark the file as modified if there are changes."""
        if changes:
            log.debug("marked %s as modified", prefix(self))
            self._timestamp = 0
        else:
            if settings.fake or self.path is None:
                self._timestamp = None
            else:
                self._timestamp = common.stamp(self.path)
            log.debug("marked %s as unmodified", prefix(self))

    def create(self, obj):
        """Create a new file for the object."""
        log.info("creating %s for %r...", prefix(self), obj)
        if self.exists:
            log.warning("already created: %s", self)
            return
        if not settings.fake:
            common.create_dirname(self.path)
            common.touch(self.path)
        self.exists = True
        self.deleted = False

    @file_required
    @prevent_recursion
    def fetch(self, obj, attrs):
        """Update the object's mapped attributes from its file."""
        log.info("fetching %r from %s...", obj, prefix(self))

        # Parse data from file
        text = self._read()
        data = self._load(text, self.path)
        log.trace("loaded: {}".format(data))

        # Update all attributes
        attrs2 = attrs.copy()
        for name, data in data.items():
            attrs2.pop(name, None)

            # Find a matching converter
            try:
                converter = attrs[name]
            except KeyError:
                # TODO: determine if runtime import is the best way to avoid
                # cyclic import
                from .converters import match
                converter = match(name, data)
                attrs[name] = converter

            # Convert the loaded attribute
            attr = getattr(obj, name, None)
            if all((isinstance(attr, converter),
                    issubclass(converter, Container))):
                attr.update_value(data)
            else:
                attr = converter.to_value(data)
                setattr(obj, name, attr)
            self._remap(attr)
            log.trace("value fetched: %s = %r", name, attr)

        # Add missing attributes
        for name, converter in attrs2.items():
            if not hasattr(obj, name):
                value = converter.to_value(None)
                msg = "fetched default value for missing attribute: %s = %r"
                log.warning(msg, name, value)
                setattr(obj, name, value)

        # Set meta attributes
        self.modified = False

    @file_required(create=True)
    @prevent_recursion
    def store(self, obj, attrs):
        """Format and save the object's mapped attributes to its file."""
        log.info("storing %r to %s...", obj, prefix(self))

        # Format the data items
        data = {}
        for name, converter in attrs.items():
            try:
                value = getattr(obj, name)
            except AttributeError:
                value = None
                msg = "storing default data for missing attribute '%s'..."
                log.warning(msg, name)

            data2 = converter.to_data(value)

            log.trace("data to store: %s = %r", name, data2)
            data[name] = data2

        # Dump data to file
        text = self._dump(data)
        self._write(text)

        # Set meta attributes
        self.modified = True
        self.auto_store = self.auto

    def delete(self):
        """Delete the object's file from the file system."""
        if self.exists:
            log.info("deleting %s...", prefix(self))
            if not settings.fake:
                common.delete(self.path)
        else:
            log.warning("already deleted: %s", self)
        self.exists = False
        self.deleted = True

    @file_required
    def _read(self):
        """Read text from the object's file.

        :param path: path to a text file

        :return: contexts of text file

        """
        if settings.fake:
            return self._fake
        elif not self.exists:
            return ""
        else:
            return common.read_text(self.path)

    @file_required
    def _write(self, text):
        """Write text to the object's file.

        :param text: text to write to a file
        :param path: path to the file

        """
        if settings.fake:
            self._fake = text
        else:
            common.write_text(text, self.path)

    @abc.abstractstaticmethod
    def _load(text, path):  # pragma: no cover (abstract method)
        """Parsed data from text.

        :param text: text read from a file
        :param path: path to the file (for displaying errors)

        :return: dictionary of parsed data

        """
        raise NotImplementedError

    @abc.abstractstaticmethod
    def _dump(data):  # pragma: no cover (abstract method)
        """Dump data to text.

        :param data: dictionary of data

        :return: text to write to a file

        """
        raise NotImplementedError

    def _remap(self, obj):
        """Restore mapping on nested attributes."""
        if isinstance(obj, Container):
            set_mapper(obj, None, common.attrs[obj.__class__], root=self)
            if isinstance(obj, dict):
                for obj2 in obj.values():
                    self._remap(obj2)
            else:
                assert isinstance(obj, list)
                for obj2 in obj:
                    self._remap(obj2)


class Helper(BaseHelper):
    """Utility class to map attributes to YAML files."""

    @staticmethod
    def _load(text, path):
        """Load YAML data from text.

        :param text: text read from a file
        :param path: path to the file (for displaying errors)

        :return: dictionary of YAML data

        """
        return common.load_yaml(text, path)

    @staticmethod
    def _dump(data):
        """Dump YAML data to text.

        :param data: dictionary of YAML data

        :return: text to write to a file

        """
        return yaml.dump(data, default_flow_style=False, allow_unicode=True)


class Mapper(Helper):
    """Maps an object's attribute to YAML files."""

    def __init__(self, obj, path, attrs, auto=True, root=None):
        super().__init__(path, auto=auto)
        self.obj = obj
        self.attrs = attrs
        self.root = root

    def create(self):  # pylint: disable=W0221
        super().create(self.obj)

    def fetch(self):  # pylint: disable=W0221
        if self.root and self.root.auto and not self._activity:
            self._activity = True
            self.root.fetch()
            self._activity = False
        super().fetch(self.obj, self.attrs)

    def store(self):  # pylint: disable=W0221
        if self.root and self.root.auto and not self._activity:
            self._activity = True
            self.root.store()
            self._activity = False
        super().store(self.obj, self.attrs)


Push — develop ( 52a881...b9d7ca )

yorm.decorator() B

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing init.py files

1		"""Core YAML mapping functionality."""
2
3	1	import os
4	1	import abc
5	1	import functools
6
7	1	import yaml
		0 ignored issues – show Configuration introduced 2015-12-18 03:00 UTC by Report Bug Copy Issue Report The import `yaml` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
8
9	1	from . import common, exceptions, settings
10	1	from .bases import Container
11
12	1	MAPPER = '__mapper__'
13
14	1	log = common.logger(__name__)
15
16
17	1	def get_mapper(obj):
18		"""Get `Mapper` instance attached to an object."""
19	1	try:
20	1	mapper = getattr(obj, MAPPER)
21	1	except AttributeError:
22	1	if isinstance(obj, (dict, list)):
23	1	return None
24		else:
25	1	msg = "mapped {!r} missing {!r} attribute".format(obj, MAPPER)
26	1	raise AttributeError(msg) from None
27		else:
28	1	return mapper
29
30
31	1	def set_mapper(obj, path, attrs, auto=True, root=None):
32		"""Create and attach a `Mapper` instance to an object."""
33	1	mapper = Mapper(obj, path, attrs, auto=auto, root=root)
34	1	setattr(obj, MAPPER, mapper)
35	1	return mapper
36
37
38	1	def file_required(create=False):
39		"""Decorator for methods that require the file to exist.
40
41		:param create: boolean or the method to decorate
42
43		"""
44	1	def decorator(method):
45
46	1	@functools.wraps(method)
47		def wrapped(self, args, *kwargs): # pylint: disable=W0621
48	1	if not self.path:
49	1	return None
50	1	if not self.exists and self.auto:
51	1	if create is True and not self.deleted:
52	1	self.create()
53		else:
54	1	msg = "cannot access deleted: {}".format(self.path)
55	1	raise exceptions.FileDeletedError(msg)
56	1	return method(self, args, *kwargs)
57
58	1	return wrapped
59
60	1	if callable(create):
61	1	return decorator(create)
62		else:
63	1	return decorator
64
65
66	1	def prevent_recursion(method):
67		"""Decorator to prevent indirect recursive calls."""
68
69	1	@functools.wraps(method)
70		def wrapped(self, args, *kwargs):
71		# pylint: disable=W0212
72	1	if self._activity:
73	1	return
74	1	self._activity = True
75	1	result = method(self, args, *kwargs)
76	1	self._activity = False
77	1	return result
78
79	1	return wrapped
80
81
82	1	def prefix(obj):
83		"""Prefix a string with a fake designator if enabled."""
84	1	fake = "(fake) " if settings.fake else ""
85	1	name = obj if isinstance(obj, str) else "'{}'".format(obj)
86	1	return fake + name
87
88
89	1	class BaseHelper(metaclass=abc.ABCMeta):
90		"""Utility class to map attributes to text files.
91
92		To start mapping attributes to a file:
93
94		create -> [empty] -> FILE
95
96		When getting an attribute:
97
98		FILE -> read -> [text] -> load -> [dict] -> fetch -> ATTRIBUTES
99
100		When setting an attribute:
101
102		ATTRIBUTES -> store -> [dict] -> dump -> [text] -> write -> FILE
103
104		After the mapped file is no longer needed:
105
106		delete -> [null] -> FILE
107
108		"""
109
110	1	def __init__(self, path, auto=True):
111	1	self.path = path
112	1	self.auto = auto
113	1	self.auto_store = False
114	1	self.exists = self.path and os.path.isfile(self.path)
115	1	self.deleted = False
116	1	self._activity = False
117	1	self._timestamp = 0
118	1	self._fake = ""
119
120	1	def __str__(self):
121	1	return str(self.path)
122
123	1	@property
124		def text(self):
125		"""Get file contents."""
126	1	log.info("getting contents of %s...", prefix(self))
127	1	if settings.fake:
128	1	text = self._fake
129		else:
130	1	text = self._read()
131	1	log.trace("text read: %r", text)
132	1	return text
133
134	1	@text.setter
135		def text(self, text):
136		"""Set file contents."""
137	1	log.info("setting contents of %s...", prefix(self))
138	1	if settings.fake:
139	1	self._fake = text
140		else:
141	1	self._write(text)
142	1	log.trace("text wrote: %r", text)
143	1	self.modified = True
144
145	1	@property
146		def modified(self):
147		"""Determine if the file has been modified."""
148	1	if settings.fake:
149	1	changes = self._timestamp is not None
150	1	return changes
151	1	elif not self.exists:
152	1	return True
153		else:
154	1	was = self._timestamp
155	1	now = common.stamp(self.path)
156	1	return was != now
157
158	1	@modified.setter
159		def modified(self, changes):
160		"""Mark the file as modified if there are changes."""
161	1	if changes:
162	1	log.debug("marked %s as modified", prefix(self))
163	1	self._timestamp = 0
164		else:
165	1	if settings.fake or self.path is None:
166	1	self._timestamp = None
167		else:
168	1	self._timestamp = common.stamp(self.path)
169	1	log.debug("marked %s as unmodified", prefix(self))
170
171	1	def create(self, obj):
172		"""Create a new file for the object."""
173	1	log.info("creating %s for %r...", prefix(self), obj)
174	1	if self.exists:
175	1	log.warning("already created: %s", self)
176	1	return
177	1	if not settings.fake:
178	1	common.create_dirname(self.path)
179	1	common.touch(self.path)
180	1	self.exists = True
181	1	self.deleted = False
182
183	1	@file_required
184	1	@prevent_recursion
185		def fetch(self, obj, attrs):
186		"""Update the object's mapped attributes from its file."""
187	1	log.info("fetching %r from %s...", obj, prefix(self))
188
189		# Parse data from file
190	1	text = self._read()
191	1	data = self._load(text, self.path)
192	1	log.trace("loaded: {}".format(data))
193
194		# Update all attributes
195	1	attrs2 = attrs.copy()
196	1	for name, data in data.items():
197	1	attrs2.pop(name, None)
198
199		# Find a matching converter
200	1	try:
201	1	converter = attrs[name]
202	1	except KeyError:
203		# TODO: determine if runtime import is the best way to avoid
204		# cyclic import
205	1	from .converters import match
206	1	converter = match(name, data)
207	1	attrs[name] = converter
208
209		# Convert the loaded attribute
210	1	attr = getattr(obj, name, None)
211	1	if all((isinstance(attr, converter),
212		issubclass(converter, Container))):
213	1	attr.update_value(data)
214		else:
215	1	attr = converter.to_value(data)
216	1	setattr(obj, name, attr)
217	1	self._remap(attr)
218	1	log.trace("value fetched: %s = %r", name, attr)
219
220		# Add missing attributes
221	1	for name, converter in attrs2.items():
222	1	if not hasattr(obj, name):
223	1	value = converter.to_value(None)
224	1	msg = "fetched default value for missing attribute: %s = %r"
225	1	log.warning(msg, name, value)
226	1	setattr(obj, name, value)
227
228		# Set meta attributes
229	1	self.modified = False
230
231	1	@file_required(create=True)
232	1	@prevent_recursion
233		def store(self, obj, attrs):
234		"""Format and save the object's mapped attributes to its file."""
235	1	log.info("storing %r to %s...", obj, prefix(self))
236
237		# Format the data items
238	1	data = {}
239	1	for name, converter in attrs.items():
240	1	try:
241	1	value = getattr(obj, name)
242	1	except AttributeError:
243	1	value = None
244	1	msg = "storing default data for missing attribute '%s'..."
245	1	log.warning(msg, name)
246
247	1	data2 = converter.to_data(value)
248
249	1	log.trace("data to store: %s = %r", name, data2)
250	1	data[name] = data2
251
252		# Dump data to file
253	1	text = self._dump(data)
254	1	self._write(text)
255
256		# Set meta attributes
257	1	self.modified = True
258	1	self.auto_store = self.auto
259
260	1	def delete(self):
261		"""Delete the object's file from the file system."""
262	1	if self.exists:
263	1	log.info("deleting %s...", prefix(self))
264	1	if not settings.fake:
265	1	common.delete(self.path)
266		else:
267	1	log.warning("already deleted: %s", self)
268	1	self.exists = False
269	1	self.deleted = True
270
271	1	@file_required
272		def _read(self):
273		"""Read text from the object's file.
274
275		:param path: path to a text file
276
277		:return: contexts of text file
278
279		"""
280	1	if settings.fake:
281	1	return self._fake
282	1	elif not self.exists:
283	1	return ""
284		else:
285	1	return common.read_text(self.path)
286
287	1	@file_required
288		def _write(self, text):
289		"""Write text to the object's file.
290
291		:param text: text to write to a file
292		:param path: path to the file
293
294		"""
295	1	if settings.fake:
296	1	self._fake = text
297		else:
298	1	common.write_text(text, self.path)
299
300	1	@abc.abstractstaticmethod
301		def _load(text, path): # pragma: no cover (abstract method)
302		"""Parsed data from text.
303
304		:param text: text read from a file
305		:param path: path to the file (for displaying errors)
306
307		:return: dictionary of parsed data
308
309		"""
310		raise NotImplementedError
311
312	1	@abc.abstractstaticmethod
313		def _dump(data): # pragma: no cover (abstract method)
314		"""Dump data to text.
315
316		:param data: dictionary of data
317
318		:return: text to write to a file
319
320		"""
321		raise NotImplementedError
322
323	1	def _remap(self, obj):
324		"""Restore mapping on nested attributes."""
325	1	if isinstance(obj, Container):
326	1	set_mapper(obj, None, common.attrs[obj.__class__], root=self)
327	1	if isinstance(obj, dict):
328	1	for obj2 in obj.values():
329	1	self._remap(obj2)
330		else:
331	1	assert isinstance(obj, list)
332	1	for obj2 in obj:
333	1	self._remap(obj2)
334
335
336	1	class Helper(BaseHelper):
337		"""Utility class to map attributes to YAML files."""
338
339	1	@staticmethod
340		def _load(text, path):
341		"""Load YAML data from text.
342
343		:param text: text read from a file
344		:param path: path to the file (for displaying errors)
345
346		:return: dictionary of YAML data
347
348		"""
349	1	return common.load_yaml(text, path)
350
351	1	@staticmethod
352		def _dump(data):
353		"""Dump YAML data to text.
354
355		:param data: dictionary of YAML data
356
357		:return: text to write to a file
358
359		"""
360	1	return yaml.dump(data, default_flow_style=False, allow_unicode=True)
361
362
363	1	class Mapper(Helper):
364		"""Maps an object's attribute to YAML files."""
365
366	1	def __init__(self, obj, path, attrs, auto=True, root=None):
367	1	super().__init__(path, auto=auto)
368	1	self.obj = obj
369	1	self.attrs = attrs
370	1	self.root = root
371
372	1	def create(self): # pylint: disable=W0221
373	1	super().create(self.obj)
374
375	1	def fetch(self): # pylint: disable=W0221
376	1	if self.root and self.root.auto and not self._activity:
377	1	self._activity = True
378	1	self.root.fetch()
379	1	self._activity = False
380	1	super().fetch(self.obj, self.attrs)
381
382	1	def store(self): # pylint: disable=W0221
383	1	if self.root and self.root.auto and not self._activity:
384	1	self._activity = True
385	1	self.root.store()
386	1	self._activity = False
387		super().store(self.obj, self.attrs)
388

jacebrowning / yorm

Push — develop ( 52a881...b9d7ca )

yorm.decorator() B

Complexity

Size

Duplication

Code Coverage

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files