gitman.models.config.Config.__init__() - Code Metrics - Inspection of "Switch to datafiles for YAML loading" - jacebrowning/gitman - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#203)

by Jace

created 2019-09-02 00:22 UTC

gitman.models.config.Config.init() A

↳ Parent: gitman.models.config

Complexity

Conditions

Size

Total Lines	8
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	8
nop	4
dl	0
loc	8
ccs	7
cts	7
cp	1
crap	1
rs	10
c	0
b	0
f	0

import os
from typing import List, Optional

import log

from datafiles import datafile, field

from .. import common, exceptions, shell
from .group import Group
from .source import Source


@datafile("{self.root}/{self.filename}", manual=True)
class Config:
    """Specifies all dependencies for a project."""

    root: Optional[str] = None

    filename: str = "gitman.yml"

    location: str = "gitman_sources"
    sources: List[Source] = field(default_factory=list)
    sources_locked: List[Source] = field(default_factory=list)
    groups: List[Group] = field(default_factory=list)

    LOG = "gitman.log"

    def __post_init__(self):
        if self.root is None:
            self.root = os.getcwd()

    def _on_post_load(self):
        for source in self.sources:
            source._on_post_load()  # pylint: disable=protected-access

        for source in self.sources_locked:
            source._on_post_load()  # pylint: disable=protected-access

        # check for conflicts between source names and group names
        for source in self.sources:
            for group in self.groups:
                if source.name == group.name:
                    msg = (
                        "Name conflict detected between source name and "
                        "group name \"{}\""
                    ).format(source.name)
                    raise exceptions.InvalidConfig(msg)

    @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,
        force_interactive=False,
        fetch=False,
        clean=True,
        skip_changes=False,
    ):  # pylint: disable=too-many-locals
        """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 = self._get_sources_filter(*names, sources=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,
                force_interactive=force_interactive,
                fetch=fetch,
                clean=clean,
                skip_changes=skip_changes,
            )
            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,
                    skip_changes=skip_changes,
                )
                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 = self._get_sources_filter(*names, sources=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, skip_changes=False):
        """Lock down the immediate dependency versions."""
        sources = self._get_sources(use_locked=obey_existing).copy()
        sources_filter = self._get_sources_filter(*names, sources=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

            source_locked = source.lock(skip_changes=skip_changes)

            if source_locked is not None:
                try:
                    index = self.sources_locked.index(source)
                except ValueError:
                    self.sources_locked.append(source_locked)
                else:
                    self.sources_locked[index] = source_locked
                count += 1

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

        if count:
            self.datafile.save()

        common.dedent()

        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: List[Source] = []
        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 _get_sources_filter(self, *names, sources):
        """Get filtered sublist of sources."""
        sources_filter = None

        groups_filter = [group for group in self.groups if group.name in list(names)]

        if groups_filter:
            sources_filter = [
                members for group in groups_filter for members in group.members
            ]
        else:
            sources_filter = list(names) if names else [s.name for s in sources]

        return sources_filter


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()  # pylint: disable=protected-access
                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 os
2	1	from typing import List, Optional
3
4	1	import log
5	1
6		from datafiles import datafile, field
7	1
8	1	from .. import common, exceptions, shell
9	1	from .group import Group
10		from .source import Source
11	1
12
13		@datafile("{self.root}/{self.filename}", manual=True)
14	1	class Config:
15	1	"""Specifies all dependencies for a project."""
16	1
17	1	root: Optional[str] = None
		0 ignored issues – show introduced 2019-09-01 19:00 UTC by Report Bug Copy Issue Report Variable annotation syntax is only supported in Python 3.6 and greater (<unknown>, line 17) Loading history...
18	1	filename: str = "gitman.yml"
19
20		location: str = "gitman_sources"
21	1	sources: List[Source] = field(default_factory=list)
22		sources_locked: List[Source] = field(default_factory=list)
23	1	groups: List[Group] = field(default_factory=list)
24
25	1	LOG = "gitman.log"
26	1
27	1	def __post_init__(self):
28	1	if self.root is None:
29	1	self.root = os.getcwd()
30	1
31		def _on_post_load(self):
32	1	for source in self.sources:
33		source._on_post_load() # pylint: disable=protected-access
34
35	1	for source in self.sources_locked:
36	1	source._on_post_load() # pylint: disable=protected-access
37
38	1	# check for conflicts between source names and group names
39		for source in self.sources:
40		for group in self.groups:
41	1	if source.name == group.name:
42		msg = (
43	1	"Name conflict detected between source name and "
44		"group name \"{}\""
45		).format(source.name)
46	1	raise exceptions.InvalidConfig(msg)
47
48	1	@property
49		def config_path(self):
50	1	"""Get the full path to the config file."""
51	1	return os.path.normpath(os.path.join(self.root, self.filename))
52	1
53	1	path = config_path
54	1
55	1	@property
56	1	def log_path(self):
57		"""Get the full path to the log file."""
58	1	return os.path.normpath(os.path.join(self.location_path, self.LOG))
59
60	1	@property
61		def location_path(self):
62		"""Get the full path to the dependency storage location."""
63		return os.path.normpath(os.path.join(self.root, self.location))
64	1
65	1	def get_path(self, name=None):
66	1	"""Get the full path to a dependency or internal file."""
67		base = self.location_path
68	1	if name == '__config__':
69	1	return self.path
70		if name == '__log__':
71	1	return self.log_path
72	1	if name:
73	1	return os.path.normpath(os.path.join(base, name))
74	1	return base
75	1
76		def install_dependencies(
77	1	self,
78	1	*names,
79	1	depth=None,
80	1	update=True,
81		recurse=False,
82	1	force=False,
83	1	force_interactive=False,
84		fetch=False,
85	1	clean=True,
86	1	skip_changes=False,
87	1	): # pylint: disable=too-many-locals
88	1	"""Download or update the specified dependencies."""
89		if depth == 0:
90	1	log.info("Skipped directory: %s", self.location_path)
91	1	return 0
92	1
93	1	sources = self._get_sources(use_locked=False if update else None)
94		sources_filter = self._get_sources_filter(*names, sources=sources)
95
96		if not os.path.isdir(self.location_path):
97		shell.mkdir(self.location_path)
98		shell.cd(self.location_path)
99		common.newline()
100		common.indent()
101	1
102		count = 0
103	1	for source in sources:
104		if source.name in sources_filter:
105	1	sources_filter.remove(source.name)
106	1	else:
107	1	log.info("Skipped dependency: %s", source.name)
108	1	continue
109
110	1	source.update_files(
111		force=force,
112	1	force_interactive=force_interactive,
113		fetch=fetch,
114	1	clean=clean,
115	1	skip_changes=skip_changes,
116	1	)
117		source.create_link(self.root, force=force)
118	1	common.newline()
119	1	count += 1
120
121	1	config = load_config(search=False)
122	1	if config:
123	1	common.indent()
124		count += config.install_dependencies(
125	1	depth=None if depth is None else max(0, depth - 1),
126	1	update=update and recurse,
127	1	recurse=recurse,
128	1	force=force,
129	1	fetch=fetch,
130		clean=clean,
131	1	skip_changes=skip_changes,
132	1	)
133	1	common.dedent()
134	1
135		shell.cd(self.location_path, _show=False)
136
137		common.dedent()
138	1	if sources_filter:
139		log.error("No such dependency: %s", ' '.join(sources_filter))
140	1	return 0
141
142	1	return count
143
144	1	def run_scripts(self, *names, depth=None, force=False):
145		"""Run scripts for the specified dependencies."""
146	1	if depth == 0:
147		log.info("Skipped directory: %s", self.location_path)
148	1	return 0
149	1
150		sources = self._get_sources()
151	1	sources_filter = self._get_sources_filter(*names, sources=sources)
152	1
153	1	shell.cd(self.location_path)
154		common.newline()
155	1	common.indent()
156	1
157	1	count = 0
158	1	for source in sources:
159	1	if source.name in sources_filter:
160		source.run_scripts(force=force)
161	1	count += 1
162	1
163	1	config = load_config(search=False)
164	1	if config:
165		common.indent()
166	1	count += config.run_scripts(
167	1	depth=None if depth is None else max(0, depth - 1), force=force
168		)
169	1	common.dedent()
170
171	1	shell.cd(self.location_path, _show=False)
172	1
173		common.dedent()
174	1
175		return count
176	1
177		def lock_dependencies(self, *names, obey_existing=True, skip_changes=False):
178	1	"""Lock down the immediate dependency versions."""
179	1	sources = self._get_sources(use_locked=obey_existing).copy()
180	1	sources_filter = self._get_sources_filter(*names, sources=sources)
181
182	1	shell.cd(self.location_path)
183		common.newline()
184	1	common.indent()
185	1
186		count = 0
187	1	for source in sources:
188	1	if source.name not in sources_filter:
189	1	log.info("Skipped dependency: %s", source.name)
190		continue
191	1
192		source_locked = source.lock(skip_changes=skip_changes)
193	1
194	1	if source_locked is not None:
195	1	try:
196		index = self.sources_locked.index(source)
197	1	except ValueError:
198		self.sources_locked.append(source_locked)
199	1	else:
200	1	self.sources_locked[index] = source_locked
201	1	count += 1
202	1
203		shell.cd(self.location_path, _show=False)
204
205		if count:
206	1	self.datafile.save()
207
208	1	common.dedent()
209
210	1	return count
211
212	1	def uninstall_dependencies(self):
213		"""Delete the dependency storage location."""
214	1	shell.cd(self.root)
215	1	shell.rm(self.location_path)
216		common.newline()
217	1
218		def clean_dependencies(self):
219	1	"""Delete the dependency storage location."""
220	1	for path in self.get_top_level_dependencies():
221	1
222		if path == self.location_path:
223	1	log.info("Skipped dependency: %s", path)
224	1	else:
225		shell.rm(path)
226	1
227	1	common.newline()
228	1
229		shell.rm(self.log_path)
230	1
231	1	def get_top_level_dependencies(self):
232	1	"""Yield the path, repository, and hash of top-level dependencies."""
233		if not os.path.exists(self.location_path):
234	1	return
235	1
236		shell.cd(self.location_path)
237	1	common.newline()
238	1	common.indent()
239	1
240	1	for source in self.sources:
241	1
242		yield os.path.join(self.location_path, source.name)
243	1
244		shell.cd(self.location_path, _show=False)
245
246	1	common.dedent()
247
248	1	def get_dependencies(self, depth=None, allow_dirty=True):
249	1	"""Yield the path, repository, and hash of each dependency."""
250		if not os.path.exists(self.location_path):
251	1	return
252
253	1	shell.cd(self.location_path)
254	1	common.newline()
255		common.indent()
256	1
257	1	for source in self.sources:
258	1
259		if depth == 0:
260	1	log.info("Skipped dependency: %s", source.name)
261	1	continue
262	1
263	1	yield source.identify(allow_dirty=allow_dirty)
264	1
265		config = load_config(search=False)
266	1	if config:
267	1	common.indent()
268		yield from config.get_dependencies(
269	1	depth=None if depth is None else max(0, depth - 1),
270		allow_dirty=allow_dirty,
271	1	)
272	1	common.dedent()
273
274	1	shell.cd(self.location_path, _show=False)
275
276	1	common.dedent()
277
278		def log(self, message="", *args):
279	1	"""Append a message to the log file."""
280	1	with open(self.log_path, 'a') as outfile:
281	1	outfile.write(message.format(*args) + '\n')
282	1
283	1	def _get_sources(self, *, use_locked=None):
284		"""Merge source lists using the requested section as the base."""
285		if use_locked is True:
286		if self.sources_locked:
287		return self.sources_locked
288		log.info("No locked sources, defaulting to none...")
289		return []
290
291		sources: List[Source] = []
292		if use_locked is False:
293		sources = self.sources
294		else:
295		if self.sources_locked:
296		log.info("Defaulting to locked sources...")
297		sources = self.sources_locked
298		else:
299		log.info("No locked sources, using latest...")
300		sources = self.sources
301
302		extras = []
303		for source in self.sources + self.sources_locked:
304		if source not in sources:
305		log.info("Source %r missing from selected section", source.name)
306		extras.append(source)
307
308		return sources + extras
309
310		def _get_sources_filter(self, *names, sources):
311		"""Get filtered sublist of sources."""
312		sources_filter = None
313
314		groups_filter = [group for group in self.groups if group.name in list(names)]
315
316		if groups_filter:
317		sources_filter = [
318		members for group in groups_filter for members in group.members
319		]
320		else:
321		sources_filter = list(names) if names else [s.name for s in sources]
322
323		return sources_filter
324
325
326		def load_config(start=None, *, search=True):
327		"""Load the config for the current project."""
328		if start:
329		start = os.path.abspath(start)
330		else:
331		start = os.getcwd()
332
333		if search:
334		log.debug("Searching for config...")
335
336		path = start
337		while path != os.path.dirname(path):
338		log.debug("Looking for config in: %s", path)
339
340		for filename in os.listdir(path):
341		if _valid_filename(filename):
342		config = Config(path, filename)
343		config._on_post_load() # pylint: disable=protected-access
344		log.debug("Found config: %s", config.path)
345		return config
346
347		if search:
348		path = os.path.dirname(path)
349		else:
350		break
351
352		if search:
353		log.debug("No config found starting from: %s", start)
354		else:
355		log.debug("No config found in: %s", start)
356
357		return None
358
359
360		def _valid_filename(filename):
361		name, ext = os.path.splitext(filename.lower())
362		if name.startswith('.'):
363		name = name[1:]
364		return name in ['gitman', 'gdm'] and ext in ['.yml', '.yaml']
365

jacebrowning / gitman

Pull Request — develop (#203)

gitman.models.config.Config.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

gitman.models.config.Config.init() A