BaseFileObject.save() - Code Metrics - Inspection of "GUI Improvement Part 2" - jacebrowning/doorstop - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#301)

unknown

created 2018-06-01 02:53 UTC

BaseFileObject.save() A

↳ Parent: BaseFileObject

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

"""Base classes and decorators for the doorstop.core package."""

import os
import abc
import functools

import yaml

from doorstop import common
from doorstop.common import DoorstopError, DoorstopWarning, DoorstopInfo
from doorstop import settings

log = common.logger(__name__)


def add_item(func):

    """Add and cache the returned item."""
    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        item = func(*args, **kwargs)
        if settings.ADDREMOVE_FILES and item.tree:
            item.tree.vcs.add(item.path)
        # pylint: disable=W0212
        if item.document and item not in item.document._items:
            item.document._items.append(item)
        if settings.CACHE_ITEMS and item.tree:
            item.tree._item_cache[item.uid] = item
            log.trace("cached item: {}".format(item))
        return item
    return wrapped


def edit_item(func):
    """Mark the returned item as modified."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        item = func(self, *args, **kwargs) or self
        if settings.ADDREMOVE_FILES and item.tree:
            item.tree.vcs.edit(item.path)
        return item
    return wrapped


def delete_item(func):

    """Remove and expunge the returned item."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        item = func(self, *args, **kwargs) or self
        if settings.ADDREMOVE_FILES and item.tree:
            item.tree.vcs.delete(item.path)
        # pylint: disable=W0212
        if item.document and item in item.document._items:
            item.document._items.remove(item)
        if settings.CACHE_ITEMS and item.tree:
            item.tree._item_cache[item.uid] = None
            log.trace("expunged item: {}".format(item))
        BaseFileObject.delete(item, item.path)
        return item
    return wrapped


def add_document(func):
    """Add and cache the returned document."""
    @functools.wraps(func)
    def wrapped(*args, **kwargs):
        document = func(*args, **kwargs) or kwargs["tree"]
        if settings.ADDREMOVE_FILES and document.tree:
            document.tree.vcs.add(document.config)
        # pylint: disable=W0212
        if settings.CACHE_DOCUMENTS and document.tree:
            document.tree._document_cache[document.prefix] = document
            log.trace("cached document: {}".format(document))
        return document
    return wrapped


def edit_document(func):
    """Mark the returned document as modified."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        document = func(self, *args, **kwargs) or self
        if settings.ADDREMOVE_FILES and document.tree:
            document.tree.vcs.edit(document.config)
        return document
    return wrapped


def delete_document(func):
    """Remove and expunge the returned document."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        document = func(self, *args, **kwargs) or self
        if settings.ADDREMOVE_FILES and document.tree:
            document.tree.vcs.delete(document.config)
        # pylint: disable=W0212
        if settings.CACHE_DOCUMENTS and document.tree:
            document.tree._document_cache[document.prefix] = None
            log.trace("expunged document: {}".format(document))
        try:
            os.rmdir(document.path)
        except OSError:
            # Directory wasn't empty
            pass
        return document
    return wrapped


class BaseValidatable(object, metaclass=abc.ABCMeta):
    """Abstract Base Class for objects that can be validated."""

    def validate(self, skip=None, document_hook=None, item_hook=None):
        """Check the object for validity.

        :param skip: list of document prefixes to skip
        :param document_hook: function to call for custom document
            validation
        :param item_hook: function to call for custom item validation

        :return: indication that the object is valid

        """
        valid = True
        # Display all issues
        for issue in self.get_issues(skip=skip, document_hook=document_hook,
                                     item_hook=item_hook):
            if isinstance(issue, DoorstopInfo) and not settings.WARN_ALL:
                log.info(issue)
            elif isinstance(issue, DoorstopWarning) and not settings.ERROR_ALL:
                log.warning(issue)
            else:
                assert isinstance(issue, DoorstopError)
                log.error(issue)
                valid = False
        # Return the result
        return valid

    @abc.abstractmethod
    def get_issues(self, skip=None, document_hook=None, item_hook=None, only_active=True):
        """Yield all the objects's issues.

        :param skip: list of document prefixes to skip
        :param document_hook: function to call for custom document
            validation
        :param item_hook: function to call for custom item validation

        :return: generator of :class:`~doorstop.common.DoorstopError`,
                              :class:`~doorstop.common.DoorstopWarning`,
                              :class:`~doorstop.common.DoorstopInfo`

        """

    @property
    def issues(self):
        """Get a list of the item's issues."""
        return list(self.get_issues())


def auto_load(func):
    """Call self.load() before execution."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        self.load()
        return func(self, *args, **kwargs)
    return wrapped


def auto_save(func):
    """Call self.save() after execution."""
    @functools.wraps(func)
    def wrapped(self, *args, **kwargs):
        result = func(self, *args, **kwargs)
        if self.is_auto_save:
            self.save()
        return result
    return wrapped


class BaseFileObject(object, metaclass=abc.ABCMeta):
    """Abstract Base Class for objects whose attributes save to a file.

    For properties that are saved to a file, decorate their getters
    with :func:`auto_load` and their setters with :func:`auto_save`.

    """

    def __init__(self, is_auto_save):
        self.path = None
        self.root = None
        self._data = {}
        self._exists = True
        self._loaded = False
        self.is_auto_save = is_auto_save

    def __hash__(self):
        return hash(self.path)

    def __eq__(self, other):
        return isinstance(other, self.__class__) and self.path == other.path

    def __ne__(self, other):
        return not self == other

    @staticmethod
    def _create(path, name):  # pragma: no cover (integration test)
        """Create a new file for the object.

        :param path: path to new file
        :param name: humanized name for this file

        :raises: :class:`~doorstop.common.DoorstopError` if the file
            already exists

        """
        if os.path.exists(path):
            raise DoorstopError("{} already exists: {}".format(name, path))
        common.create_dirname(path)
        common.touch(path)

    @abc.abstractmethod
    def load(self, reload=False):  # pragma: no cover (abstract method)
        """Load the object's properties from its file."""
        # 1. Start implementations of this method with:
        if self._loaded and not reload:
            return
        # 2. Call self._read() and update properties here
        # 3. End implementations of this method with:
        self._loaded = True

    def _read(self, path):  # pragma: no cover (integration test)
        """Read text from the object's file.

        :param path: path to a text file

        :return: contexts of text file

        """
        if not self._exists:
            msg = "cannot read from deleted: {}".format(self.path)
            raise DoorstopError(msg)
        return common.read_text(path)

    @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)

    @abc.abstractmethod
    def save(self):  # pragma: no cover (abstract method)
        """Format and save the object's properties to its file."""
        # 1. Call self._write() with the current properties here
        # 2. End implementations of this method with:
        self._loaded = False

    def _write(self, text, path):  # pragma: no cover (integration test)
        """Write text to the object's file.

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

        """
        if not self._exists:
            raise DoorstopError("cannot save to deleted: {}".format(self))
        common.write_text(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)

    # properties #############################################################

    @property
    def relpath(self):
        """Get the item's relative path string."""
        relpath = os.path.relpath(self.path, self.root)
        return "@{}{}".format(os.sep, relpath)

    # extended attributes ####################################################

    @property
    @auto_load
    def extended(self):
        """Get a list of all extended attribute names."""
        names = []
        for name in self._data:
            if not hasattr(self, name):
                names.append(name)
        return sorted(names)

    @auto_load
    def get(self, name, default=None):
        """Get an extended attribute.

        :param name: name of extended attribute
        :param default: value to return for missing attributes

        :return: value of extended attribute

        """
        if hasattr(self, name):
            cname = self.__class__.__name__
            msg = "'{n}' can be accessed from {c}.{n}".format(n=name, c=cname)
            log.trace(msg)
            return getattr(self, name)
        else:
            return self._data.get(name, default)

    @auto_load
    @auto_save
    def set(self, name, value):
        """Set an extended attribute.

        :param name: name of extended attribute
        :param value: value to set

        """
        if hasattr(self, name):
            cname = self.__class__.__name__
            msg = "'{n}' can be set from {c}.{n}".format(n=name, c=cname)
            log.trace(msg)
            setattr(self, name, value)
        else:
            self._data[name] = value

    @auto_load
    @auto_save
    def remove(self, name):
        """Remove an extended attribute.

        :param name: name of extended attribute

        """
        if hasattr(self, name):
            cname = self.__class__.__name__
            msg = "'{n}' can't be remove from {c}".format(n=name, c=cname)
            log.error(msg)
            assert False, msg
        else:
            try:
                del self._data[name]
            except KeyError:
                pass  # Ignore the exception since  the attribute is already not there.

    # actions ################################################################

    def delete(self, path):
        """Delete the object's file from the file system."""
        if self._exists:
            log.info("deleting {}...".format(path))
            common.delete(path)
            self._loaded = False  # force the object to reload
            self._exists = False  # but, prevent future access
        else:
            log.warning("already deleted: {}".format(self))


1			"""Base classes and decorators for the doorstop.core package."""
2
3	1		import os
4	1		import abc
5	1		import functools
6
7	1		import yaml
8
9	1		from doorstop import common
10	1		from doorstop.common import DoorstopError, DoorstopWarning, DoorstopInfo
11	1		from doorstop import settings
12
13	1		log = common.logger(__name__)
14
15
16	1	View Code Duplication	def add_item(func):
			0 ignored issues – show Duplication introduced 2016-06-08 15:21 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
17			"""Add and cache the returned item."""
18	1		@functools.wraps(func)
19			def wrapped(args, *kwargs):
20			item = func(args, *kwargs)
21	1		if settings.ADDREMOVE_FILES and item.tree:
22	1		item.tree.vcs.add(item.path)
23	1		# pylint: disable=W0212
24			if item.document and item not in item.document._items:
25	1		item.document._items.append(item)
26	1		if settings.CACHE_ITEMS and item.tree:
27	1		item.tree._item_cache[item.uid] = item
28	1		log.trace("cached item: {}".format(item))
29	1		return item
30	1		return wrapped
31	1
32
33			def edit_item(func):
34	1		"""Mark the returned item as modified."""
35			@functools.wraps(func)
36	1		def wrapped(self, args, *kwargs):
37			item = func(self, args, *kwargs) or self
38			if settings.ADDREMOVE_FILES and item.tree:
39	1		item.tree.vcs.edit(item.path)
40	1		return item
41	1		return wrapped
42	1
43	1
44		View Code Duplication	def delete_item(func):
			0 ignored issues – show Duplication introduced 2018-02-17 19:53 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
45			"""Remove and expunge the returned item."""
46	1		@functools.wraps(func)
47			def wrapped(self, args, *kwargs):
48	1		item = func(self, args, *kwargs) or self
49			if settings.ADDREMOVE_FILES and item.tree:
50			item.tree.vcs.delete(item.path)
51	1		# pylint: disable=W0212
52	1		if item.document and item in item.document._items:
53	1		item.document._items.remove(item)
54			if settings.CACHE_ITEMS and item.tree:
55	1		item.tree._item_cache[item.uid] = None
56	1		log.trace("expunged item: {}".format(item))
57	1		BaseFileObject.delete(item, item.path)
58	1		return item
59	1		return wrapped
60	1
61	1
62	1		def add_document(func):
63			"""Add and cache the returned document."""
64			@functools.wraps(func)
65	1		def wrapped(args, *kwargs):
66			document = func(args, *kwargs) or kwargs["tree"]
67	1		if settings.ADDREMOVE_FILES and document.tree:
68			document.tree.vcs.add(document.config)
69			# pylint: disable=W0212
70	1		if settings.CACHE_DOCUMENTS and document.tree:
71	1		document.tree._document_cache[document.prefix] = document
72	1		log.trace("cached document: {}".format(document))
73			return document
74	1		return wrapped
75	1
76	1
77	1		def edit_document(func):
78	1		"""Mark the returned document as modified."""
79			@functools.wraps(func)
80			def wrapped(self, args, *kwargs):
81	1		document = func(self, args, *kwargs) or self
82			if settings.ADDREMOVE_FILES and document.tree:
83	1		document.tree.vcs.edit(document.config)
84			return document
85			return wrapped
86	1
87	1
88	1		def delete_document(func):
89	1		"""Remove and expunge the returned document."""
90	1		@functools.wraps(func)
91			def wrapped(self, args, *kwargs):
92			document = func(self, args, *kwargs) or self
93	1		if settings.ADDREMOVE_FILES and document.tree:
94			document.tree.vcs.delete(document.config)
95	1		# pylint: disable=W0212
96			if settings.CACHE_DOCUMENTS and document.tree:
97			document.tree._document_cache[document.prefix] = None
98	1		log.trace("expunged document: {}".format(document))
99	1		try:
100	1		os.rmdir(document.path)
101			except OSError:
102	1		# Directory wasn't empty
103	1		pass
104	1		return document
105	1		return wrapped
106	1
107	1
108			class BaseValidatable(object, metaclass=abc.ABCMeta):
109	1		"""Abstract Base Class for objects that can be validated."""
110	1
111	1		def validate(self, skip=None, document_hook=None, item_hook=None):
112			"""Check the object for validity.
113
114	1		:param skip: list of document prefixes to skip
115			:param document_hook: function to call for custom document
116			validation
117	1		:param item_hook: function to call for custom item validation
118
119			:return: indication that the object is valid
120
121			"""
122			valid = True
123			# Display all issues
124			for issue in self.get_issues(skip=skip, document_hook=document_hook,
125			item_hook=item_hook):
126			if isinstance(issue, DoorstopInfo) and not settings.WARN_ALL:
127			log.info(issue)
128	1		elif isinstance(issue, DoorstopWarning) and not settings.ERROR_ALL:
129			log.warning(issue)
130	1		else:
131			assert isinstance(issue, DoorstopError)
132	1		log.error(issue)
133	1		valid = False
134	1		# Return the result
135	1		return valid
136
137	1		@abc.abstractmethod
138	1		def get_issues(self, skip=None, document_hook=None, item_hook=None, only_active=True):
139	1		"""Yield all the objects's issues.
140
141	1		:param skip: list of document prefixes to skip
142			:param document_hook: function to call for custom document
143	1		validation
144	1		:param item_hook: function to call for custom item validation
145
146			:return: generator of :class:`~doorstop.common.DoorstopError`,
147			:class:`~doorstop.common.DoorstopWarning`,
148			:class:`~doorstop.common.DoorstopInfo`
149
150			"""
151
152			@property
153			def issues(self):
154			"""Get a list of the item's issues."""
155			return list(self.get_issues())
156
157
158	1		def auto_load(func):
159			"""Call self.load() before execution."""
160			@functools.wraps(func)
161	1		def wrapped(self, args, *kwargs):
162			self.load()
163			return func(self, args, *kwargs)
164	1		return wrapped
165
166	1
167			def auto_save(func):
168			"""Call self.save() after execution."""
169	1		@functools.wraps(func)
170	1		def wrapped(self, args, *kwargs):
171	1		result = func(self, args, *kwargs)
172			if self.is_auto_save:
173			self.save()
174	1		return result
175			return wrapped
176	1
177
178			class BaseFileObject(object, metaclass=abc.ABCMeta):
179	1		"""Abstract Base Class for objects whose attributes save to a file.
180	1
181	1		For properties that are saved to a file, decorate their getters
182	1		with :func:`auto_load` and their setters with :func:`auto_save`.
183	1
184			"""
185
186	1		def __init__(self, is_auto_save):
187			self.path = None
188			self.root = None
189			self._data = {}
190			self._exists = True
191			self._loaded = False
192			self.is_auto_save = is_auto_save
193
194	1		def __hash__(self):
195			return hash(self.path)
196	1
197	1		def __eq__(self, other):
198	1		return isinstance(other, self.__class__) and self.path == other.path
199	1
200	1		def __ne__(self, other):
201	1		return not self == other
202
203	1		@staticmethod
204	1		def _create(path, name): # pragma: no cover (integration test)
205			"""Create a new file for the object.
206	1
207	1		:param path: path to new file
208			:param name: humanized name for this file
209	1
210	1		:raises: :class:`~doorstop.common.DoorstopError` if the file
211			already exists
212	1
213			"""
214			if os.path.exists(path):
215			raise DoorstopError("{} already exists: {}".format(name, path))
216			common.create_dirname(path)
217			common.touch(path)
218
219			@abc.abstractmethod
220			def load(self, reload=False): # pragma: no cover (abstract method)
221			"""Load the object's properties from its file."""
222			# 1. Start implementations of this method with:
223			if self._loaded and not reload:
224			return
225			# 2. Call self._read() and update properties here
226			# 3. End implementations of this method with:
227			self._loaded = True
228	1
229			def _read(self, path): # pragma: no cover (integration test)
230			"""Read text from the object's file.
231
232			:param path: path to a text file
233
234			:return: contexts of text file
235
236			"""
237			if not self._exists:
238			msg = "cannot read from deleted: {}".format(self.path)
239			raise DoorstopError(msg)
240			return common.read_text(path)
241
242			@staticmethod
243			def _load(text, path):
244			"""Load YAML data from text.
245
246			:param text: text read from a file
247			:param path: path to the file (for displaying errors)
248
249			:return: dictionary of YAML data
250
251	1		"""
252			return common.load_yaml(text, path)
253
254			@abc.abstractmethod
255			def save(self): # pragma: no cover (abstract method)
256			"""Format and save the object's properties to its file."""
257			# 1. Call self._write() with the current properties here
258			# 2. End implementations of this method with:
259			self._loaded = False
260
261	1		def _write(self, text, path): # pragma: no cover (integration test)
262			"""Write text to the object's file.
263	1
264			:param text: text to write to a file
265			:param path: path to the file
266
267			"""
268			if not self._exists:
269			raise DoorstopError("cannot save to deleted: {}".format(self))
270			common.write_text(text, path)
271
272			@staticmethod
273			def _dump(data):
274			"""Dump YAML data to text.
275
276			:param data: dictionary of YAML data
277
278			:return: text to write to a file
279
280			"""
281			return yaml.dump(data, default_flow_style=False, allow_unicode=True)
282	1
283			# properties #############################################################
284
285			@property
286			def relpath(self):
287			"""Get the item's relative path string."""
288			relpath = os.path.relpath(self.path, self.root)
289			return "@{}{}".format(os.sep, relpath)
290
291	1		# extended attributes ####################################################
292
293			@property
294			@auto_load
295	1		def extended(self):
296			"""Get a list of all extended attribute names."""
297			names = []
298	1		for name in self._data:
299	1		if not hasattr(self, name):
300			names.append(name)
301			return sorted(names)
302
303	1		@auto_load
304	1		def get(self, name, default=None):
305			"""Get an extended attribute.
306
307	1		:param name: name of extended attribute
308	1		:param default: value to return for missing attributes
309	1
310	1		:return: value of extended attribute
311	1
312			"""
313	1		if hasattr(self, name):
314	1		cname = self.__class__.__name__
315			msg = "'{n}' can be accessed from {c}.{n}".format(n=name, c=cname)
316			log.trace(msg)
317			return getattr(self, name)
318			else:
319			return self._data.get(name, default)
320
321			@auto_load
322			@auto_save
323	1		def set(self, name, value):
324	1		"""Set an extended attribute.
325	1
326	1		:param name: name of extended attribute
327	1		:param value: value to set
328
329	1		"""
330			if hasattr(self, name):
331	1		cname = self.__class__.__name__
332	1		msg = "'{n}' can be set from {c}.{n}".format(n=name, c=cname)
333			log.trace(msg)
334			setattr(self, name, value)
335			else:
336			self._data[name] = value
337
338			@auto_load
339			@auto_save
340	1		def remove(self, name):
341	1		"""Remove an extended attribute.
342	1
343	1		:param name: name of extended attribute
344	1
345			"""
346	1		if hasattr(self, name):
347			cname = self.__class__.__name__
348			msg = "'{n}' can't be remove from {c}".format(n=name, c=cname)
349			log.error(msg)
350	1		assert False, msg
351			else:
352	1		try:
353	1		del self._data[name]
354	1		except KeyError:
355	1		pass # Ignore the exception since the attribute is already not there.
356	1
357			# actions ################################################################
358	1
359			def delete(self, path):
360			"""Delete the object's file from the file system."""
361			if self._exists:
362			log.info("deleting {}...".format(path))
363			common.delete(path)
364			self._loaded = False # force the object to reload
365			self._exists = False # but, prevent future access
366			else:
367			log.warning("already deleted: {}".format(self))
368

jacebrowning / doorstop

Pull Request — develop (#301)

BaseFileObject.save() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like