gitman.models.config.Config.install_dependencies() - Code Metrics - Inspection of "Feature/sources type svn support" - jacebrowning/gitman - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#168)

unknown

created 2018-11-06 10:23 UTC

gitman.models.config.Config.install_dependencies() C

↳ Parent: gitman.models.config

Complexity

Conditions

Size

Total Lines	51
Code Lines	40

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	33
CRAP Score	10

Importance

Changes

Metric	Value
cc	10
eloc	40
nop	8
dl	0
loc	51
ccs	33
cts	33
cp	1
crap	10
rs	5.9999
c	0
b	0
f	0

How to fix Long Method Complexity Many Parameters

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like gitman.models.config.Config.install_dependencies() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

If you can get the data from another object that the method knows about already: Replace the parameter with method call
If several parameters are part of another object: Preserve Whole Object
If parameters are not yet connected: Introduce Parameter Object

import logging
import os

import yorm
from yorm.types import SortedList, String

from . import Source
from .. import common, shell


log = logging.getLogger(__name__)


@yorm.attr(location=String)
@yorm.attr(sources=SortedList.of_type(Source))
@yorm.attr(sources_locked=SortedList.of_type(Source))
@yorm.sync("{self.root}/{self.filename}", auto_save=False)
class Config(yorm.ModelMixin):
    """Specifies all dependencies for a project."""

    LOG = "gitman.log"

    def __init__(self, root=None,
                 filename="gitman.yml", location="gitman_sources"):
        super().__init__()
        self.root = root or os.getcwd()
        self.filename = filename
        self.location = location
        self.sources = []
        self.sources_locked = []


    def _on_post_load(self):
        for source in self.sources:
            source._on_post_load()
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
        for source in self.sources_locked:
            source._on_post_load()
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent

    @property
    def config_path(self):
        """Get the full path to the config file."""
        return os.path.normpath(os.path.join(self.root, self.filename))
    path = config_path

    @property
    def log_path(self):
        """Get the full path to the log file."""
        return os.path.normpath(os.path.join(self.location_path, self.LOG))

    @property
    def location_path(self):
        """Get the full path to the dependency storage location."""
        return os.path.normpath(os.path.join(self.root, self.location))

    def get_path(self, name=None):
        """Get the full path to a dependency or internal file."""
        base = self.location_path
        if name == '__config__':
            return self.path
        if name == '__log__':
            return self.log_path
        if name:
            return os.path.normpath(os.path.join(base, name))
        return base

    def install_dependencies(self, *names, depth=None,
                             update=True, recurse=False,
                             force=False, fetch=False, clean=True):
        """Download or update the specified dependencies."""
        if depth == 0:
            log.info("Skipped directory: %s", self.location_path)
            return 0

        sources = self._get_sources(use_locked=False if update else None)
        sources_filter = list(names) if names else [s.name for s in sources]

        if not os.path.isdir(self.location_path):
            shell.mkdir(self.location_path)
        shell.cd(self.location_path)
        common.newline()
        common.indent()

        count = 0
        for source in sources:
            if source.name in sources_filter:
                sources_filter.remove(source.name)
            else:
                log.info("Skipped dependency: %s", source.name)
                continue

            source.update_files(force=force, fetch=fetch, clean=clean)
            source.create_link(self.root, force=force)
            common.newline()
            count += 1

            config = load_config(search=False)
            if config:
                common.indent()
                count += config.install_dependencies(
                    depth=None if depth is None else max(0, depth - 1),
                    update=update and recurse,
                    recurse=recurse,
                    force=force,
                    fetch=fetch,
                    clean=clean,
                )
                common.dedent()

            shell.cd(self.location_path, _show=False)

        common.dedent()
        if sources_filter:
            log.error("No such dependency: %s", ' '.join(sources_filter))
            return 0

        return count

    def run_scripts(self, *names, depth=None, force=False):
        """Run scripts for the specified dependencies."""
        if depth == 0:
            log.info("Skipped directory: %s", self.location_path)
            return 0

        sources = self._get_sources()
        sources_filter = list(names) if names else [s.name for s in sources]

        shell.cd(self.location_path)
        common.newline()
        common.indent()

        count = 0
        for source in sources:
            if source.name in sources_filter:
                source.run_scripts(force=force)
                count += 1

                config = load_config(search=False)
                if config:
                    common.indent()
                    count += config.run_scripts(
                        depth=None if depth is None else max(0, depth - 1),
                        force=force,
                    )
                    common.dedent()

                shell.cd(self.location_path, _show=False)

        common.dedent()

        return count

    def lock_dependencies(self, *names, obey_existing=True):
        """Lock down the immediate dependency versions."""
        sources = self._get_sources(use_locked=obey_existing).copy()
        sources_filter = list(names) if names else [s.name for s in sources]

        shell.cd(self.location_path)
        common.newline()
        common.indent()

        count = 0
        for source in sources:
            if source.name not in sources_filter:
                log.info("Skipped dependency: %s", source.name)
                continue

            try:
                index = self.sources_locked.index(source)
            except ValueError:
                self.sources_locked.append(source.lock())
            else:
                self.sources_locked[index] = source.lock()
            count += 1

            shell.cd(self.location_path, _show=False)

        if count:
            self.save()

        return count

    def uninstall_dependencies(self):
        """Delete the dependency storage location."""
        shell.cd(self.root)
        shell.rm(self.location_path)
        common.newline()

    def clean_dependencies(self):
        """Delete the dependency storage location."""
        for path in self.get_top_level_dependencies():

            if path == self.location_path:
                log.info("Skipped dependency: %s", path)
            else:
                shell.rm(path)

            common.newline()

        shell.rm(self.log_path)

    def get_top_level_dependencies(self):
        """Yield the path, repository, and hash of top-level dependencies."""
        if not os.path.exists(self.location_path):
            return

        shell.cd(self.location_path)
        common.newline()
        common.indent()

        for source in self.sources:

            yield os.path.join(self.location_path, source.name)

            shell.cd(self.location_path, _show=False)

        common.dedent()

    def get_dependencies(self, depth=None, allow_dirty=True):
        """Yield the path, repository, and hash of each dependency."""
        if not os.path.exists(self.location_path):
            return

        shell.cd(self.location_path)
        common.newline()
        common.indent()

        for source in self.sources:

            if depth == 0:
                log.info("Skipped dependency: %s", source.name)
                continue

            yield source.identify(allow_dirty=allow_dirty)

            config = load_config(search=False)
            if config:
                common.indent()
                yield from config.get_dependencies(
                    depth=None if depth is None else max(0, depth - 1),
                    allow_dirty=allow_dirty,
                )
                common.dedent()

            shell.cd(self.location_path, _show=False)

        common.dedent()

    def log(self, message="", *args):
        """Append a message to the log file."""
        with open(self.log_path, 'a') as outfile:
            outfile.write(message.format(*args) + '\n')

    def _get_sources(self, *, use_locked=None):
        """Merge source lists using the requested section as the base."""
        if use_locked is True:
            if self.sources_locked:
                return self.sources_locked
            log.info("No locked sources, defaulting to none...")
            return []

        sources = []
        if use_locked is False:
            sources = self.sources
        else:
            if self.sources_locked:
                log.info("Defaulting to locked sources...")
                sources = self.sources_locked
            else:
                log.info("No locked sources, using latest...")
                sources = self.sources

        extras = []
        for source in self.sources + self.sources_locked:
            if source not in sources:
                log.info("Source %r missing from selected section",
                         source.name)
                extras.append(source)

        return sources + extras


def load_config(start=None, *, search=True):
    """Load the config for the current project."""
    if start:
        start = os.path.abspath(start)
    else:
        start = os.getcwd()

    if search:
        log.debug("Searching for config...")

    path = start
    while path != os.path.dirname(path):
        log.debug("Looking for config in: %s", path)

        for filename in os.listdir(path):
            if _valid_filename(filename):
                config = Config(path, filename)
                config._on_post_load()
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                log.debug("Found config: %s", config.path)
                return config

        if search:
            path = os.path.dirname(path)
        else:
            break

    if search:
        log.debug("No config found starting from: %s", start)
    else:
        log.debug("No config found in: %s", start)

    return None


def _valid_filename(filename):
    name, ext = os.path.splitext(filename.lower())
    if name.startswith('.'):
        name = name[1:]
    return name in ['gitman', 'gdm'] and ext in ['.yml', '.yaml']


1	1	import logging
2	1	import os
3
4	1	import yorm
5	1	from yorm.types import SortedList, String
6
7	1	from . import Source
8	1	from .. import common, shell
9	1
10
11	1	log = logging.getLogger(__name__)
12
13
14	1	@yorm.attr(location=String)
15	1	@yorm.attr(sources=SortedList.of_type(Source))
16	1	@yorm.attr(sources_locked=SortedList.of_type(Source))
17	1	@yorm.sync("{self.root}/{self.filename}", auto_save=False)
18	1	class Config(yorm.ModelMixin):
19		"""Specifies all dependencies for a project."""
20
21	1	LOG = "gitman.log"
22
23	1	def __init__(self, root=None,
24		filename="gitman.yml", location="gitman_sources"):
25	1	super().__init__()
26	1	self.root = root or os.getcwd()
27	1	self.filename = filename
28	1	self.location = location
29	1	self.sources = []
30	1	self.sources_locked = []
31
32	1
33		def _on_post_load(self):
34		for source in self.sources:
35	1	source._on_post_load()
		0 ignored issues – show Coding Style Best Practice introduced 2018-06-26 08:04 UTC by Report Bug Copy Issue Report It seems like `_on_post_load` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
36	1	for source in self.sources_locked:
37		source._on_post_load()
		0 ignored issues – show Coding Style Best Practice introduced 2018-06-26 08:04 UTC by Report Bug Copy Issue Report It seems like `_on_post_load` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
38	1
39		@property
40		def config_path(self):
41	1	"""Get the full path to the config file."""
42		return os.path.normpath(os.path.join(self.root, self.filename))
43	1	path = config_path
44
45		@property
46	1	def log_path(self):
47		"""Get the full path to the log file."""
48	1	return os.path.normpath(os.path.join(self.location_path, self.LOG))
49
50	1	@property
51	1	def location_path(self):
52	1	"""Get the full path to the dependency storage location."""
53	1	return os.path.normpath(os.path.join(self.root, self.location))
54	1
55	1	def get_path(self, name=None):
56	1	"""Get the full path to a dependency or internal file."""
57		base = self.location_path
58	1	if name == '__config__':
59		return self.path
60	1	if name == '__log__':
61		return self.log_path
62		if name:
63		return os.path.normpath(os.path.join(base, name))
64	1	return base
65	1
66	1	def install_dependencies(self, *names, depth=None,
67		update=True, recurse=False,
68	1	force=False, fetch=False, clean=True):
69	1	"""Download or update the specified dependencies."""
70		if depth == 0:
71	1	log.info("Skipped directory: %s", self.location_path)
72	1	return 0
73	1
74	1	sources = self._get_sources(use_locked=False if update else None)
75	1	sources_filter = list(names) if names else [s.name for s in sources]
76
77	1	if not os.path.isdir(self.location_path):
78	1	shell.mkdir(self.location_path)
79	1	shell.cd(self.location_path)
80	1	common.newline()
81		common.indent()
82	1
83	1	count = 0
84		for source in sources:
85	1	if source.name in sources_filter:
86	1	sources_filter.remove(source.name)
87	1	else:
88	1	log.info("Skipped dependency: %s", source.name)
89		continue
90	1
91	1	source.update_files(force=force, fetch=fetch, clean=clean)
92	1	source.create_link(self.root, force=force)
93	1	common.newline()
94		count += 1
95
96		config = load_config(search=False)
97		if config:
98		common.indent()
99		count += config.install_dependencies(
100		depth=None if depth is None else max(0, depth - 1),
101	1	update=update and recurse,
102		recurse=recurse,
103	1	force=force,
104		fetch=fetch,
105	1	clean=clean,
106	1	)
107	1	common.dedent()
108	1
109		shell.cd(self.location_path, _show=False)
110	1
111		common.dedent()
112	1	if sources_filter:
113		log.error("No such dependency: %s", ' '.join(sources_filter))
114	1	return 0
115	1
116	1	return count
117
118	1	def run_scripts(self, *names, depth=None, force=False):
119	1	"""Run scripts for the specified dependencies."""
120		if depth == 0:
121	1	log.info("Skipped directory: %s", self.location_path)
122	1	return 0
123	1
124		sources = self._get_sources()
125	1	sources_filter = list(names) if names else [s.name for s in sources]
126	1
127	1	shell.cd(self.location_path)
128	1	common.newline()
129	1	common.indent()
130
131	1	count = 0
132	1	for source in sources:
133	1	if source.name in sources_filter:
134	1	source.run_scripts(force=force)
135		count += 1
136
137		config = load_config(search=False)
138	1	if config:
139		common.indent()
140	1	count += config.run_scripts(
141		depth=None if depth is None else max(0, depth - 1),
142	1	force=force,
143		)
144	1	common.dedent()
145
146	1	shell.cd(self.location_path, _show=False)
147
148	1	common.dedent()
149	1
150		return count
151	1
152	1	def lock_dependencies(self, *names, obey_existing=True):
153	1	"""Lock down the immediate dependency versions."""
154		sources = self._get_sources(use_locked=obey_existing).copy()
155	1	sources_filter = list(names) if names else [s.name for s in sources]
156	1
157	1	shell.cd(self.location_path)
158	1	common.newline()
159	1	common.indent()
160
161	1	count = 0
162	1	for source in sources:
163	1	if source.name not in sources_filter:
164	1	log.info("Skipped dependency: %s", source.name)
165		continue
166	1
167	1	try:
168		index = self.sources_locked.index(source)
169	1	except ValueError:
170		self.sources_locked.append(source.lock())
171	1	else:
172	1	self.sources_locked[index] = source.lock()
173		count += 1
174	1
175		shell.cd(self.location_path, _show=False)
176	1
177		if count:
178	1	self.save()
179	1
180	1	return count
181
182	1	def uninstall_dependencies(self):
183		"""Delete the dependency storage location."""
184	1	shell.cd(self.root)
185	1	shell.rm(self.location_path)
186		common.newline()
187	1
188	1	def clean_dependencies(self):
189	1	"""Delete the dependency storage location."""
190		for path in self.get_top_level_dependencies():
191	1
192		if path == self.location_path:
193	1	log.info("Skipped dependency: %s", path)
194	1	else:
195	1	shell.rm(path)
196
197	1	common.newline()
198
199	1	shell.rm(self.log_path)
200	1
201	1	def get_top_level_dependencies(self):
202	1	"""Yield the path, repository, and hash of top-level dependencies."""
203		if not os.path.exists(self.location_path):
204		return
205
206	1	shell.cd(self.location_path)
207		common.newline()
208	1	common.indent()
209
210	1	for source in self.sources:
211
212	1	yield os.path.join(self.location_path, source.name)
213
214	1	shell.cd(self.location_path, _show=False)
215	1
216		common.dedent()
217	1
218		def get_dependencies(self, depth=None, allow_dirty=True):
219	1	"""Yield the path, repository, and hash of each dependency."""
220	1	if not os.path.exists(self.location_path):
221	1	return
222
223	1	shell.cd(self.location_path)
224	1	common.newline()
225		common.indent()
226	1
227	1	for source in self.sources:
228	1
229		if depth == 0:
230	1	log.info("Skipped dependency: %s", source.name)
231	1	continue
232	1
233		yield source.identify(allow_dirty=allow_dirty)
234	1
235	1	config = load_config(search=False)
236		if config:
237	1	common.indent()
238	1	yield from config.get_dependencies(
239	1	depth=None if depth is None else max(0, depth - 1),
240	1	allow_dirty=allow_dirty,
241	1	)
242		common.dedent()
243	1
244		shell.cd(self.location_path, _show=False)
245
246	1	common.dedent()
247
248	1	def log(self, message="", *args):
249	1	"""Append a message to the log file."""
250		with open(self.log_path, 'a') as outfile:
251	1	outfile.write(message.format(*args) + '\n')
252
253	1	def _get_sources(self, *, use_locked=None):
254	1	"""Merge source lists using the requested section as the base."""
255		if use_locked is True:
256	1	if self.sources_locked:
257	1	return self.sources_locked
258	1	log.info("No locked sources, defaulting to none...")
259		return []
260	1
261	1	sources = []
262	1	if use_locked is False:
263	1	sources = self.sources
264	1	else:
265		if self.sources_locked:
266	1	log.info("Defaulting to locked sources...")
267	1	sources = self.sources_locked
268		else:
269	1	log.info("No locked sources, using latest...")
270		sources = self.sources
271	1
272	1	extras = []
273		for source in self.sources + self.sources_locked:
274	1	if source not in sources:
275		log.info("Source %r missing from selected section",
276	1	source.name)
277		extras.append(source)
278
279	1	return sources + extras
280	1
281	1
282	1	def load_config(start=None, *, search=True):
283	1	"""Load the config for the current project."""
284		if start:
285		start = os.path.abspath(start)
286		else:
287		start = os.getcwd()
288
289		if search:
290		log.debug("Searching for config...")
291
292		path = start
293		while path != os.path.dirname(path):
294		log.debug("Looking for config in: %s", path)
295
296		for filename in os.listdir(path):
297		if _valid_filename(filename):
298		config = Config(path, filename)
299		config._on_post_load()
		0 ignored issues – show Coding Style Best Practice introduced 2018-06-26 08:04 UTC by Report Bug Copy Issue Report It seems like `_on_post_load` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
300		log.debug("Found config: %s", config.path)
301		return config
302
303		if search:
304		path = os.path.dirname(path)
305		else:
306		break
307
308		if search:
309		log.debug("No config found starting from: %s", start)
310		else:
311		log.debug("No config found in: %s", start)
312
313		return None
314
315
316		def _valid_filename(filename):
317		name, ext = os.path.splitext(filename.lower())
318		if name.startswith('.'):
319		name = name[1:]
320		return name in ['gitman', 'gdm'] and ext in ['.yml', '.yaml']
321

jacebrowning / gitman

Pull Request — develop (#168)

gitman.models.config.Config.install_dependencies() C

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity Many Parameters

Long Method

Complexity

Many Parameters

Duplication Side-by-Side

Filter issues like