yorm.mapper.Mapper.__init__() - Code Metrics - jacebrowning/yorm - Measure and Improve Code Quality continuously with Scrutinizer

yorm.mapper.Mapper.init() A
last analyzed 2018-12-09 21:40 UTC

↳ Parent: yorm.mapper

Complexity

Conditions

Size

Total Lines	18
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	16
nop	9
dl	0
loc	18
ccs	14
cts	14
cp	1
crap	1
rs	9.6
c	0
b	0
f	0

How to fix Many Parameters

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

import functools
from pprint import pformat
import logging

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

log = logging.getLogger(__name__)


def file_required(method):
    """Decorate 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):
    """Decorate methods to prevent indirect recursive calls."""

    @functools.wraps(method)
    def wrapped(self, *args, **kwargs):
        # pylint: disable=protected-access
        if self._activity:
            return None
        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] -> parse -> [dict] -> load -> ATTRIBUTES

    When setting an attribute:

        ATTRIBUTES -> save -> [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_save=True,
                 auto_track=False, auto_resolve=False):
        self._obj = obj
        self.path = path
        self.attrs = attrs
        self.auto_create = auto_create
        self.auto_save = auto_save
        self.auto_track = auto_track
        self.auto_resolve = auto_resolve

        self.exists = diskutils.exists(self.path)
        self.deleted = False
        self.auto_save_after_load = 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 as a string."""
        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 from a string."""
        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

    @property
    def data(self):
        """Get the file values as a dictionary."""
        text = self._read()
        try:
            data = diskutils.parse(text, self.path)
        except ValueError as e:
            if not self.auto_resolve:
                raise e from None

            log.debug(e)
            log.warning("Clearing invalid contents: %s", self.path)
            self._write("")
            return {}

        log.trace("Parsed data: \n%s", pformat(data))
        return data

    @data.setter
    def data(self, data):
        """Set the file values from a dictionary."""
        text = diskutils.dump(data, self.path)
        self._write(text)

    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 load(self):
        """Update the object's mapped attributes from its file."""
        log.info("Loading %r from %s...", self._obj, prefix(self))

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

            # Find a matching converter
            try:
                converter = self.attrs[name]
            except KeyError:
                if self.auto_track:
                    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 parsed value to the attribute's final type
            attr = getattr(self._obj, name, None)
            if isinstance(attr, converter) and \
                    issubclass(converter, Container):
                attr.update_value(data, auto_track=self.auto_track)
            else:
                log.trace("Converting attribute %r using %r", name, converter)
                attr = converter.to_value(data)
                setattr(self._obj, name, attr)
            self._remap(attr, self)
            log.trace("Value loaded: %s = %r", name, attr)

        # Add missing attributes
        for name, converter in attrs2.items():
            try:
                existing_attr = getattr(self._obj, name)
            except AttributeError:
                value = converter.create_default()
                msg = "Default value for missing object attribute: %s = %r"
                log.warning(msg, name, value)
                setattr(self._obj, name, value)
                self._remap(value, self)
            else:
                if issubclass(converter, Container):
                    if isinstance(existing_attr, converter):
                        pass  # TODO: Update 'existing_attr' values to replace None values
                    else:
                        msg = "Converting container attribute %r using %r"
                        log.trace(msg, name, converter)
                        value = converter.create_default()
                        setattr(self._obj, name, value)
                        self._remap(value, self)
                else:
                    pass  # TODO: Figure out when this case occurs

        # Set meta attributes
        self.modified = False

    def _remap(self, obj, root):
        """Attach mapper 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 save(self):
        """Format and save the object's mapped attributes to its file."""
        log.info("Saving %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 save: %s = %r", name, data2)
            data[name] = data2

        # Save the formatted to disk
        self.data = data

        # Set meta attributes
        self.modified = True
        self.auto_save_after_load = self.auto_save

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

jacebrowning / yorm

yorm.mapper.Mapper.__init__() A last analyzed 2018-12-09 21:40 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Many Parameters

Many Parameters

Duplication Side-by-Side

Filter issues like

yorm.mapper.Mapper.init() A
last analyzed 2018-12-09 21:40 UTC