wrapped() - Code Metrics - Inspection of "Add ORM-like utilities" - jacebrowning/yorm - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — develop (#116)

by Jace

created 2016-04-13 03:06 UTC

wrapped() A

↳ Parent: file_required()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	2.0625

Metric	Value
dl	0
loc	10
rs	9.4285
ccs	6
cts	8
cp	0.75
cc	2
crap	2.0625

"""Core object-file mapping functionality."""

import functools
from pprint import pformat

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

log = common.logger(__name__)


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

    @functools.wraps(method)
    def wrapped(self, *args, **kwargs):
        if self.deleted:
            msg = "File deleted: {}".format(self.path)
            raise exceptions.DeletedFileError(msg)
        elif self.missing and not settings.fake:
            msg = "File missing: {}".format(self.path)
            raise exceptions.MissingFileError(msg)
        else:
            return method(self, *args, **kwargs)

    return wrapped


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

    @functools.wraps(method)
    def wrapped(self, *args, **kwargs):
        # pylint: disable=protected-access
        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 Mapper:
    """Utility class to map an object's attributes to a file.

    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, obj, path, attrs, *,
                 auto_create=True, auto_store=True, auto_attr=False):
        self._obj = obj
        self.path = path
        self.attrs = attrs
        self.auto_create = auto_create
        self.auto_store = auto_store
        self.auto_attr = auto_attr

        self.exists = diskutils.exists(self.path)
        self.deleted = False
        self.auto_store_after_fetch = False

        self._activity = False
        self._timestamp = 0
        self._fake = ""

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

    @property
    def missing(self):
        return not self.exists

    @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:
            # TODO: this raises an exception is the file is missing
            was = self._timestamp
            now = diskutils.stamp(self.path)
            return was != now

    @modified.setter
    @file_required
    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 = diskutils.stamp(self.path)
            log.debug("Marked %s as unmodified", prefix(self))

    @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.rstrip())
        self.modified = True

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

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

        # Parse data from file
        text = self._read()
        data = diskutils.load(text=text, path=self.path)
        log.trace("Loaded data: \n%s", pformat(data))

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

            # Find a matching converter
            try:
                converter = self.attrs[name]
            except KeyError:
                if self.auto_attr:
                    converter = types.match(name, data)
                    self.attrs[name] = converter
                else:
                    msg = "Ignored unknown file attribute: %s = %r"
                    log.warning(msg, name, data)
                    continue

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

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

        # Set meta attributes
        self.modified = False

    def _remap(self, obj, root):
        """Restore mapping on nested attributes."""
        if isinstance(obj, Container):
            common.set_mapper(obj, root)

            if isinstance(obj, dict):
                for obj2 in obj.values():
                    self._remap(obj2, root)
            else:
                assert isinstance(obj, list)
                for obj2 in obj:
                    self._remap(obj2, root)

    @file_required
    @prevent_recursion
    def store(self):
        """Format and save the object's mapped attributes to its file."""
        log.info("Storing %r to %s...", self._obj, prefix(self))

        # Format the data items
        data = self.attrs.__class__()
        for name, converter in self.attrs.items():
            try:
                value = getattr(self._obj, name)
            except AttributeError:
                data2 = converter.to_data(None)
                msg = "Default data for missing object attribute: %s = %r"
                log.warning(msg, name, data2)
            else:
                data2 = converter.to_data(value)

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

        # Dump data to file
        text = diskutils.dump(data=data, path=self.path)
        self._write(text)

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

    def delete(self):
        """Delete the object's file from the file system."""
        if self.exists:
            log.info("Deleting %s...", prefix(self))
            diskutils.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."""
        if settings.fake:
            return self._fake
        elif not self.exists:
            return ""
        else:
            return diskutils.read(self.path)

    @file_required
    def _write(self, text):
        """Write text to the object's file."""
        if settings.fake:
            self._fake = text
        else:
            diskutils.write(text, self.path)


1		"""Core object-file mapping functionality."""
2
3	1	import functools
4	1	from pprint import pformat
5
6	1	from . import common, diskutils, exceptions, types, settings
7	1	from .bases import Container
8
9	1	log = common.logger(__name__)
10
11
12	1	def file_required(method):
13		"""Decorator for methods that require the file to exist."""
14
15	1	@functools.wraps(method)
16		def wrapped(self, args, *kwargs):
17	1	if self.deleted:
18	1	msg = "File deleted: {}".format(self.path)
19	1	raise exceptions.DeletedFileError(msg)
20	1	elif self.missing and not settings.fake:
21		msg = "File missing: {}".format(self.path)
22		raise exceptions.MissingFileError(msg)
23		else:
24	1	return method(self, args, *kwargs)
25
26	1	return wrapped
27
28
29	1	def prevent_recursion(method):
30		"""Decorator to prevent indirect recursive calls."""
31
32	1	@functools.wraps(method)
33		def wrapped(self, args, *kwargs):
34		# pylint: disable=protected-access
35	1	if self._activity:
36	1	return
37	1	self._activity = True
38	1	result = method(self, args, *kwargs)
39	1	self._activity = False
40	1	return result
41
42	1	return wrapped
43
44
45	1	def prefix(obj):
46		"""Prefix a string with a fake designator if enabled."""
47	1	fake = "(fake) " if settings.fake else ""
48	1	name = obj if isinstance(obj, str) else "'{}'".format(obj)
49	1	return fake + name
50
51
52	1	class Mapper:
53		"""Utility class to map an object's attributes to a file.
54
55		To start mapping attributes to a file:
56
57		create -> [empty] -> FILE
58
59		When getting an attribute:
60
61		FILE -> read -> [text] -> load -> [dict] -> fetch -> ATTRIBUTES
62
63		When setting an attribute:
64
65		ATTRIBUTES -> store -> [dict] -> dump -> [text] -> write -> FILE
66
67		After the mapped file is no longer needed:
68
69		delete -> [null] -> FILE
70
71		"""
72
73	1	def __init__(self, obj, path, attrs, *,
74		auto_create=True, auto_store=True, auto_attr=False):
75	1	self._obj = obj
76	1	self.path = path
77	1	self.attrs = attrs
78	1	self.auto_create = auto_create
79	1	self.auto_store = auto_store
80	1	self.auto_attr = auto_attr
81
82	1	self.exists = diskutils.exists(self.path)
83	1	self.deleted = False
84	1	self.auto_store_after_fetch = False
85
86	1	self._activity = False
87	1	self._timestamp = 0
88	1	self._fake = ""
89
90	1	def __str__(self):
91	1	return str(self.path)
92
93	1	@property
94		def missing(self):
95	1	return not self.exists
96
97	1	@property
98		def modified(self):
99		"""Determine if the file has been modified."""
100	1	if settings.fake:
101	1	changes = self._timestamp is not None
102	1	return changes
103	1	elif not self.exists:
104	1	return True
105		else:
106		# TODO: this raises an exception is the file is missing
107	1	was = self._timestamp
108	1	now = diskutils.stamp(self.path)
109	1	return was != now
110
111	1	@modified.setter
112	1	@file_required
113		def modified(self, changes):
114		"""Mark the file as modified if there are changes."""
115	1	if changes:
116	1	log.debug("Marked %s as modified", prefix(self))
117	1	self._timestamp = 0
118		else:
119	1	if settings.fake or self.path is None:
120	1	self._timestamp = None
121		else:
122	1	self._timestamp = diskutils.stamp(self.path)
123	1	log.debug("Marked %s as unmodified", prefix(self))
124
125	1	@property
126		def text(self):
127		"""Get file contents."""
128	1	log.info("Getting contents of %s...", prefix(self))
129	1	if settings.fake:
130	1	text = self._fake
131		else:
132	1	text = self._read()
133	1	log.trace("Text read: \n%s", text[:-1])
134	1	return text
135
136	1	@text.setter
137		def text(self, text):
138		"""Set file contents."""
139	1	log.info("Setting contents of %s...", prefix(self))
140	1	if settings.fake:
141	1	self._fake = text
142		else:
143	1	self._write(text)
144	1	log.trace("Text wrote: \n%s", text.rstrip())
145	1	self.modified = True
146
147	1	def create(self):
148		"""Create a new file for the object."""
149	1	log.info("Creating %s for %r...", prefix(self), self._obj)
150	1	if self.exists:
151	1	log.warning("Already created: %s", self)
152	1	return
153	1	if not settings.fake:
154	1	diskutils.touch(self.path)
155	1	self.exists = True
156	1	self.deleted = False
157
158	1	@file_required
159	1	@prevent_recursion
160		def fetch(self):
161		"""Update the object's mapped attributes from its file."""
162	1	log.info("Fetching %r from %s...", self._obj, prefix(self))
163
164		# Parse data from file
165	1	text = self._read()
166	1	data = diskutils.load(text=text, path=self.path)
167	1	log.trace("Loaded data: \n%s", pformat(data))
168
169		# Update all attributes
170	1	attrs2 = self.attrs.copy()
171	1	for name, data in data.items():
172	1	attrs2.pop(name, None)
173
174		# Find a matching converter
175	1	try:
176	1	converter = self.attrs[name]
177	1	except KeyError:
178	1	if self.auto_attr:
179	1	converter = types.match(name, data)
180	1	self.attrs[name] = converter
181		else:
182	1	msg = "Ignored unknown file attribute: %s = %r"
183	1	log.warning(msg, name, data)
184	1	continue
185
186		# Convert the loaded attribute
187	1	attr = getattr(self._obj, name, None)
188	1	if all((isinstance(attr, converter),
189		issubclass(converter, Container))):
190	1	attr.update_value(data, auto_attr=self.auto_attr)
191		else:
192	1	attr = converter.to_value(data)
193	1	setattr(self._obj, name, attr)
194	1	self._remap(attr, self)
195	1	log.trace("Value fetched: %s = %r", name, attr)
196
197		# Add missing attributes
198	1	for name, converter in attrs2.items():
199	1	if not hasattr(self._obj, name):
200	1	value = converter.to_value(None)
201	1	msg = "Default value for missing object attribute: %s = %r"
202	1	log.warning(msg, name, value)
203	1	setattr(self._obj, name, value)
204
205		# Set meta attributes
206	1	self.modified = False
207
208	1	def _remap(self, obj, root):
209		"""Restore mapping on nested attributes."""
210	1	if isinstance(obj, Container):
211	1	common.set_mapper(obj, root)
212
213	1	if isinstance(obj, dict):
214	1	for obj2 in obj.values():
215	1	self._remap(obj2, root)
216		else:
217	1	assert isinstance(obj, list)
218	1	for obj2 in obj:
219	1	self._remap(obj2, root)
220
221	1	@file_required
222	1	@prevent_recursion
223		def store(self):
224		"""Format and save the object's mapped attributes to its file."""
225	1	log.info("Storing %r to %s...", self._obj, prefix(self))
226
227		# Format the data items
228	1	data = self.attrs.__class__()
229	1	for name, converter in self.attrs.items():
230	1	try:
231	1	value = getattr(self._obj, name)
232	1	except AttributeError:
233	1	data2 = converter.to_data(None)
234	1	msg = "Default data for missing object attribute: %s = %r"
235	1	log.warning(msg, name, data2)
236		else:
237	1	data2 = converter.to_data(value)
238
239	1	log.trace("Data to store: %s = %r", name, data2)
240	1	data[name] = data2
241
242		# Dump data to file
243	1	text = diskutils.dump(data=data, path=self.path)
244	1	self._write(text)
245
246		# Set meta attributes
247	1	self.modified = True
248	1	self.auto_store_after_fetch = self.auto_store
249
250	1	def delete(self):
251		"""Delete the object's file from the file system."""
252	1	if self.exists:
253	1	log.info("Deleting %s...", prefix(self))
254	1	diskutils.delete(self.path)
255		else:
256	1	log.warning("Already deleted: %s", self)
257	1	self.exists = False
258	1	self.deleted = True
259
260	1	@file_required
261		def _read(self):
262		"""Read text from the object's file."""
263	1	if settings.fake:
264	1	return self._fake
265	1	elif not self.exists:
266		return ""
267		else:
268	1	return diskutils.read(self.path)
269
270	1	@file_required
271		def _write(self, text):
272		"""Write text to the object's file."""
273	1	if settings.fake:
274	1	self._fake = text
275		else:
276		diskutils.write(text, self.path)
277

jacebrowning / yorm

Pull Request — develop (#116)

wrapped() A

Complexity

Size

Duplication

Code Coverage

Duplication Side-by-Side

Filter issues like