yorm.wrapped() - Code Metrics - Inspection of "Refactor common tests" - jacebrowning/yorm - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( 4ae7a2...f1348d )

by Jace

created 2016-02-19 02:23 UTC

yorm.wrapped() B

↳ Parent: yorm.file_required()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	6

Metric	Value
cc	6
dl	0
loc	12
ccs	9
cts	9
cp	1
crap	6
rs	8

"""Core YAML mapping functionality."""

import os
import abc
import functools
from pprint import pformat

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:
        return object.__getattribute__(obj, MAPPER)
    except AttributeError:
        return None


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):
            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."""

    # pylint: disable=protected-access

    @functools.wraps(method)
    def wrapped(self, *args, **kwargs):
        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: \n%s", text[:-1])
        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: \n%s", text[:-1])
        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))

    @property
    def ext(self):
        if '.' in self.path:
            return self.path.split('.')[-1]
        else:
            return 'yml'

    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=text, path=self.path, ext=self.ext)
        log.trace("Loaded data: \n%s", pformat(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 .types 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=data, ext=self.ext)
        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, ext):  # 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, ext):  # 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, ext):
        """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_file(text, path, ext)

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

        :param data: dictionary of YAML data

        :return: text to write to a file

        """
        return common.dump_file(data, ext)


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

    # TODO: rewrite this class to use composition
    # pylint: disable=arguments-differ

    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):
        super().create(self.obj)

    def fetch(self):
        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):
        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)


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

jacebrowning / yorm

Push — develop ( 4ae7a2...f1348d )

yorm.wrapped() B

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like