copy_dir_contents() - Code Metrics - Inspection of "Merge pull request #252 from jacebrowning/bugfix/a..." - jacebrowning/doorstop - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 551f29...aa48a4 )

by Jace

created 2017-03-16 15:06 UTC

copy_dir_contents() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	5

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	5
c	1
b	0
f	0
dl	0
loc	15
ccs	6
cts	6
cp	1
crap	5
rs	8.5454

"""Common exceptions, classes, and functions for Doorstop."""

import os
import shutil
import argparse
import logging
import glob

import yaml

verbosity = 0  # global verbosity setting for controlling string formatting
PRINT_VERBOSITY = 0  # minimum verbosity to using `print`
STR_VERBOSITY = 3  # minimum verbosity to use verbose `__str__`
MAX_VERBOSITY = 4  # maximum verbosity level implemented


def _trace(self, message, *args, **kws):  # pragma: no cover (manual test)
    """New logging level, TRACE."""
    if self.isEnabledFor(logging.DEBUG - 1):
        self._log(logging.DEBUG - 1, message, args, **kws)  # pylint: disable=W0212


logging.addLevelName(logging.DEBUG - 1, "TRACE")
logging.Logger.trace = _trace

logger = logging.getLogger
log = logger(__name__)


# exception classes ##########################################################


class DoorstopError(Exception):
    """Generic Doorstop error."""


class DoorstopFileError(DoorstopError, IOError):
    """Raised on IO errors."""


class DoorstopWarning(DoorstopError, Warning):
    """Generic Doorstop warning."""


class DoorstopInfo(DoorstopWarning, Warning):
    """Generic Doorstop info."""

# logging classes ############################################################


class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter):
    """Command-line help text formatter with wider help text."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, max_help_position=40, **kwargs)


class WarningFormatter(logging.Formatter, object):
    """Logging formatter that displays verbose formatting for WARNING+."""

    def __init__(self, default_format, verbose_format, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.default_format = default_format
        self.verbose_format = verbose_format

    def format(self, record):  # pragma: no cover (manual test)
        """Python 3 hack to change the formatting style dynamically."""
        if record.levelno > logging.INFO:
            self._style._fmt = self.verbose_format  # pylint: disable=W0212
        else:
            self._style._fmt = self.default_format  # pylint: disable=W0212
        return super().format(record)


# disk helper functions ######################################################


def create_dirname(path):
    """Ensure a parent directory exists for a path."""
    dirpath = os.path.dirname(path)
    if dirpath and not os.path.isdir(dirpath):
        log.info("creating directory {}...".format(dirpath))
        os.makedirs(dirpath)


def read_lines(path, encoding='utf-8'):
    """Read lines of text from a file.

    :param path: file to write lines
    :param encoding: output file encoding

    :return: path of new file

    """
    log.trace("reading lines from '{}'...".format(path))
    with open(path, 'r', encoding=encoding) as stream:
        for line in stream:
            yield line


def read_text(path, encoding='utf-8'):
    """Read text from a file.

    :param path: file path to read from
    :param encoding: input file encoding

    :return: string

    """
    log.trace("reading text from '{}'...".format(path))
    with open(path, 'r', encoding=encoding) as stream:
        text = stream.read()
    return text


def load_yaml(text, path):
    """Parse a dictionary from YAML text.

    :param text: string containing dumped YAML data
    :param path: file path for error messages

    :return: dictionary

    """
    # Load the YAML data
    try:
        data = yaml.load(text) or {}
    except yaml.error.YAMLError as exc:
        msg = "invalid contents: {}:\n{}".format(path, exc)
        raise DoorstopError(msg) from None
    # Ensure data is a dictionary
    if not isinstance(data, dict):
        msg = "invalid contents: {}".format(path)
        raise DoorstopError(msg)
    # Return the parsed data
    return data


def write_lines(lines, path, end='\n', encoding='utf-8'):
    """Write lines of text to a file.

    :param lines: iterator of strings
    :param path: file to write lines
    :param end: string to end lines
    :param encoding: output file encoding

    :return: path of new file

    """
    log.trace("writing lines to '{}'...".format(path))
    with open(path, 'wb') as stream:
        for line in lines:
            data = (line + end).encode(encoding)
            stream.write(data)
    return path


def write_text(text, path, encoding='utf-8'):
    """Write text to a file.

    :param text: string
    :param path: file to write text
    :param encoding: output file encoding

    :return: path of new file

    """
    if text:
        log.trace("writing text to '{}'...".format(path))
    with open(path, 'wb') as stream:
        data = text.encode(encoding)
        stream.write(data)
    return path


def touch(path):  # pragma: no cover (integration test)
    """Ensure a file exists."""
    if not os.path.exists(path):
        log.trace("creating empty '{}'...".format(path))
        write_text('', path)


def copy_dir_contents(src, dst):
    """Copy the contents of a directory."""
    for fpath in glob.glob('{}/*'.format(src)):
        dest_path = os.path.join(dst, os.path.split(fpath)[-1])
        if os.path.exists(dest_path):
            if os.path.basename(fpath) == "doorstop":
                msg = "Skipping '{}' as this directory name is required by doorstop".format(fpath)
            else:
                msg = "Skipping '{}' as a file or directory with this name already exists".format(fpath)
            log.warning(msg)
        else:
            if os.path.isdir(fpath):
                shutil.copytree(fpath, dest_path)
            else:
                shutil.copyfile(fpath, dest_path)


def delete(path):  # pragma: no cover (integration test)
    """Delete a file or directory with error handling."""
    if os.path.isdir(path):
        try:
            log.trace("deleting '{}'...".format(path))
            shutil.rmtree(path)
        except IOError:
            # bug: http://code.activestate.com/lists/python-list/159050
            msg = "unable to delete: {}".format(path)
            log.warning(msg)
    elif os.path.isfile(path):
        log.trace("deleting '{}'...".format(path))
        os.remove(path)


def delete_contents(dirname):
    """Delete the contents of a directory."""
    for file in glob.glob('{}/*'.format(dirname)):
        if os.path.isdir(file):
            shutil.rmtree(os.path.join(dirname, file))
        else:
            try:
                os.remove(os.path.join(dirname, file))
            except FileExistsError:
                log.warning("Two assets folders have files or directories " +
                            "with the same name")
                raise


1		"""Common exceptions, classes, and functions for Doorstop."""
2
3	1	import os
4	1	import shutil
5	1	import argparse
6	1	import logging
7	1	import glob
8
9	1	import yaml
10
11	1	verbosity = 0 # global verbosity setting for controlling string formatting
12	1	PRINT_VERBOSITY = 0 # minimum verbosity to using `print`
13	1	STR_VERBOSITY = 3 # minimum verbosity to use verbose `__str__`
14	1	MAX_VERBOSITY = 4 # maximum verbosity level implemented
15
16
17		def _trace(self, message, args, *kws): # pragma: no cover (manual test)
18		"""New logging level, TRACE."""
19		if self.isEnabledFor(logging.DEBUG - 1):
20		self._log(logging.DEBUG - 1, message, args, **kws) # pylint: disable=W0212
21
22
23	1	logging.addLevelName(logging.DEBUG - 1, "TRACE")
24	1	logging.Logger.trace = _trace
25
26	1	logger = logging.getLogger
27	1	log = logger(__name__)
28
29
30		# exception classes ##########################################################
31
32
33	1	class DoorstopError(Exception):
34		"""Generic Doorstop error."""
35
36
37	1	class DoorstopFileError(DoorstopError, IOError):
38		"""Raised on IO errors."""
39
40
41	1	class DoorstopWarning(DoorstopError, Warning):
42		"""Generic Doorstop warning."""
43
44
45	1	class DoorstopInfo(DoorstopWarning, Warning):
46		"""Generic Doorstop info."""
47
48		# logging classes ############################################################
49
50
51	1	class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter):
52		"""Command-line help text formatter with wider help text."""
53
54	1	def __init__(self, args, *kwargs):
55	1	super().__init__(args, max_help_position=40, *kwargs)
56
57
58	1	class WarningFormatter(logging.Formatter, object):
59		"""Logging formatter that displays verbose formatting for WARNING+."""
60
61	1	def __init__(self, default_format, verbose_format, args, *kwargs):
62	1	super().__init__(args, *kwargs)
63	1	self.default_format = default_format
64	1	self.verbose_format = verbose_format
65
66		def format(self, record): # pragma: no cover (manual test)
67		"""Python 3 hack to change the formatting style dynamically."""
68		if record.levelno > logging.INFO:
69		self._style._fmt = self.verbose_format # pylint: disable=W0212
70		else:
71		self._style._fmt = self.default_format # pylint: disable=W0212
72		return super().format(record)
73
74
75		# disk helper functions ######################################################
76
77
78	1	def create_dirname(path):
79		"""Ensure a parent directory exists for a path."""
80	1	dirpath = os.path.dirname(path)
81	1	if dirpath and not os.path.isdir(dirpath):
82	1	log.info("creating directory {}...".format(dirpath))
83	1	os.makedirs(dirpath)
84
85
86	1	def read_lines(path, encoding='utf-8'):
87		"""Read lines of text from a file.
88
89		:param path: file to write lines
90		:param encoding: output file encoding
91
92		:return: path of new file
93
94		"""
95	1	log.trace("reading lines from '{}'...".format(path))
96	1	with open(path, 'r', encoding=encoding) as stream:
97	1	for line in stream:
98	1	yield line
99
100
101	1	def read_text(path, encoding='utf-8'):
102		"""Read text from a file.
103
104		:param path: file path to read from
105		:param encoding: input file encoding
106
107		:return: string
108
109		"""
110	1	log.trace("reading text from '{}'...".format(path))
111	1	with open(path, 'r', encoding=encoding) as stream:
112	1	text = stream.read()
113	1	return text
114
115
116	1	def load_yaml(text, path):
117		"""Parse a dictionary from YAML text.
118
119		:param text: string containing dumped YAML data
120		:param path: file path for error messages
121
122		:return: dictionary
123
124		"""
125		# Load the YAML data
126	1	try:
127	1	data = yaml.load(text) or {}
128	1	except yaml.error.YAMLError as exc:
129	1	msg = "invalid contents: {}:\n{}".format(path, exc)
130	1	raise DoorstopError(msg) from None
131		# Ensure data is a dictionary
132	1	if not isinstance(data, dict):
133	1	msg = "invalid contents: {}".format(path)
134	1	raise DoorstopError(msg)
135		# Return the parsed data
136	1	return data
137
138
139	1	def write_lines(lines, path, end='\n', encoding='utf-8'):
140		"""Write lines of text to a file.
141
142		:param lines: iterator of strings
143		:param path: file to write lines
144		:param end: string to end lines
145		:param encoding: output file encoding
146
147		:return: path of new file
148
149		"""
150	1	log.trace("writing lines to '{}'...".format(path))
151	1	with open(path, 'wb') as stream:
152	1	for line in lines:
153	1	data = (line + end).encode(encoding)
154	1	stream.write(data)
155	1	return path
156
157
158	1	def write_text(text, path, encoding='utf-8'):
159		"""Write text to a file.
160
161		:param text: string
162		:param path: file to write text
163		:param encoding: output file encoding
164
165		:return: path of new file
166
167		"""
168	1	if text:
169	1	log.trace("writing text to '{}'...".format(path))
170	1	with open(path, 'wb') as stream:
171	1	data = text.encode(encoding)
172	1	stream.write(data)
173	1	return path
174
175
176		def touch(path): # pragma: no cover (integration test)
177		"""Ensure a file exists."""
178		if not os.path.exists(path):
179		log.trace("creating empty '{}'...".format(path))
180		write_text('', path)
181
182
183	1	def copy_dir_contents(src, dst):
184		"""Copy the contents of a directory."""
185	1	for fpath in glob.glob('{}/*'.format(src)):
186	1	dest_path = os.path.join(dst, os.path.split(fpath)[-1])
187	1	if os.path.exists(dest_path):
188	1	if os.path.basename(fpath) == "doorstop":
189		msg = "Skipping '{}' as this directory name is required by doorstop".format(fpath)
190	1	else:
191		msg = "Skipping '{}' as a file or directory with this name already exists".format(fpath)
192		log.warning(msg)
193		else:
194		if os.path.isdir(fpath):
195		shutil.copytree(fpath, dest_path)
196		else:
197		shutil.copyfile(fpath, dest_path)
198
199
200		def delete(path): # pragma: no cover (integration test)
201		"""Delete a file or directory with error handling."""
202		if os.path.isdir(path):
203		try:
204		log.trace("deleting '{}'...".format(path))
205		shutil.rmtree(path)
206		except IOError:
207		# bug: http://code.activestate.com/lists/python-list/159050
208	1	msg = "unable to delete: {}".format(path)
209		log.warning(msg)
210	1	elif os.path.isfile(path):
211	1	log.trace("deleting '{}'...".format(path))
212		os.remove(path)
213
214	1
215	1	def delete_contents(dirname):
216		"""Delete the contents of a directory."""
217		for file in glob.glob('{}/*'.format(dirname)):
218		if os.path.isdir(file):
219		shutil.rmtree(os.path.join(dirname, file))
220		else:
221		try:
222		os.remove(os.path.join(dirname, file))
223		except FileExistsError:
224		log.warning("Two assets folders have files or directories " +
225		"with the same name")
226		raise
227

jacebrowning / doorstop

Push — develop ( 551f29...aa48a4 )

copy_dir_contents() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like