Source code for lago.workdir

#
# Copyright 2016 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
#
# Refer to the README and COPYING files for full details of the license
#
"""
A workdir is the base directory where lago will store all the files it needs
and that are unique (not shared between workdirs).

It's basic structure is a directory with one soft link and multiple
directories, one per prefix. Where the link points to the default prefix to
use.

"""
import os
import logging
import shutil
from functools import partial, wraps

from . import (prefix, utils)
from .plugins import cli

LOGGER = logging.getLogger(__name__)


[docs]class WorkdirError(RuntimeError): """ Base exception for workdir errors, catch this one to catch any workdir error """ pass
[docs]class PrefixNotFound(WorkdirError): pass
[docs]class MalformedWorkdir(WorkdirError): pass
[docs]class PrefixAlreadyExists(WorkdirError): pass
[docs]def workdir_loaded(func): """ Decorator to make sure that the workdir is loaded when calling the decorated function """ @wraps(func) def decorator(workdir, *args, **kwargs): if not workdir.loaded: workdir.load() return func(workdir, *args, **kwargs) return decorator
[docs]class Workdir(object): """ This class reperesents a base workdir, where you can store multiple prefixes Properties: path(str): Path to the workdir perfixes(dict of str->self.prefix_class): dict with the prefixes in the workdir, by name current(str): Name of the current prefix prefix_class(type): Class to use when creating prefixes """ def __init__(self, path, prefix_class=prefix.Prefix): self.path = path self.prefixes = {} self.current = None self.loaded = False self.prefix_class = prefix_class
[docs] def join(self, *args): """ Gets a joined path prefixed with the workdir path Args: *args(str): path sections to join Returns: str: Joined path prefixed with the workdir path """ return os.path.join(self.path, *args)
[docs] def initialize(self, prefix_name='default', *args, **kwargs): """ Initializes a workdir by adding a new prefix to the workdir. Args: prefix_name(str): Name of the new prefix to add *args: args to pass along to the prefix constructor *kwargs: kwargs to pass along to the prefix constructor Returns: The newly created prefix Raises: PrefixAlreadyExists: if the prefix name already exists in the workdir """ if self.loaded: raise WorkdirError('Workdir %s already initialized' % self.path) if not os.path.exists(self.path): LOGGER.debug('Creating workdir %s', self.path) os.makedirs(self.path) self.prefixes[prefix_name] = self.prefix_class( self.join(prefix_name), *args, **kwargs ) self.prefixes[prefix_name].initialize() if self.current is None: self._set_current(prefix_name) self.load() return self.prefixes[prefix_name]
[docs] def load(self): """ Loads the prefixes that are available is the workdir Returns: None Raises: MalformedWorkdir: if the wordir is malformed """ if self.loaded: LOGGER.debug('Already loaded') return try: basepath, dirs, _ = os.walk(self.path).next() except StopIteration: raise MalformedWorkdir('Empty dir %s' % self.path) full_path = partial(os.path.join, basepath) found_current = False for dirname in dirs: if dirname == 'current' and os.path.islink(full_path('current')): self.current = os.path.basename( os.readlink(full_path('current')) ) found_current = True continue elif dirname == 'current': raise MalformedWorkdir( '"%s/current" should be a soft link' % self.path ) self.prefixes[dirname] = self.prefix_class( prefix=self.join(dirname) ) if not found_current: raise MalformedWorkdir( '"%s/current" should exist and be a soft link' % self.path ) self._update_current()
[docs] def _update_current(self): """ Makes sure that a current is set """ if not self.current or self.current not in self.prefixes: if 'default' in self.prefixes: selected_current = 'default' elif self.prefixes: selected_current = sorted(self.prefixes.keys()).pop() else: # should never get here raise MalformedWorkdir( 'No current link and no prefixes in workdir %s' % self.path ) logging.info( 'Missing current link, setting it to %s', selected_current, ) self._set_current(selected_current)
[docs] def _set_current(self, new_current): """ Change the current default prefix, for internal usage Args: new_current(str): Name of the new current prefix, it must already exist Returns: None Raises: PrefixNotFound: if the given prefix name does not exist in the workdir """ new_cur_full_path = self.join(new_current) if not os.path.exists(new_cur_full_path): raise PrefixNotFound( 'Prefix "%s" does not exist in workdir %s' % (new_current, self.path) ) if os.path.lexists(self.join('current')): os.unlink(self.join('current')) os.symlink(new_current, self.join('current')) self.current = new_current
@workdir_loaded
[docs] def set_current(self, new_current): """ Change the current default prefix Args: new_current(str): Name of the new current prefix, it must already exist Returns: None Raises: PrefixNotFound: if the given prefix name does not exist in the workdir """ self._set_current(new_current)
@workdir_loaded
[docs] def add_prefix(self, name, *args, **kwargs): """ Adds a new prefix to the workdir. Args: name(str): Name of the new prefix to add *args: args to pass along to the prefix constructor *kwargs: kwargs to pass along to the prefix constructor Returns: The newly created prefix Raises: PrefixAlreadyExists: if the prefix name already exists in the workdir """ if os.path.exists(self.join(name)): raise PrefixAlreadyExists( 'Prefix with name %s already exists in workdir %s' % (name, self.path) ) self.prefixes[name] = self.prefix_class( self.join(name), *args, **kwargs ) self.prefixes[name].initialize() if self.current is None: self.set_current(name) return self.prefixes[name]
@workdir_loaded
[docs] def get_prefix(self, name): """ Retrieve a prefix, resolving the current one if needed Args: name(str): name of the prefix to retrieve, or current to get the current one Returns: self.prefix_class: instance of the prefix with the given name """ if name == 'current': name = self.current try: return self.prefixes[name] except KeyError: raise KeyError( 'Unable to find prefix "%s" in workdir %s' % (name, self.path) )
@workdir_loaded
[docs] def destroy(self, prefix_names=None): """ Destroy all the given prefixes and remove any left files if no more prefixes are left Args: prefix_names(list of str): list of prefix names to destroy, if None passed (default) will destroy all of them """ if prefix_names is None: self.destroy(prefix_names=self.prefixes.keys()) return for prefix_name in prefix_names: if prefix_name == 'current' and self.current in prefix_names: continue elif prefix_name == 'current': prefix_name = self.current self.get_prefix(prefix_name).destroy() self.prefixes.pop(prefix_name) if self.prefixes: self._update_current() if not self.prefixes: shutil.rmtree(self.path)
@classmethod
[docs] def resolve_workdir_path(cls, start_path=os.curdir): """ Look for an existing workdir in the given path, in a path/.lago dir, or in a .lago dir under any of it's parent directories Args: start_path (str): path to start the search from, if None passed, it will use the current dir Returns: str: path to the found prefix Raises: RuntimeError: if no prefix was found """ if start_path == 'auto': start_path = os.curdir cur_path = start_path LOGGER.debug( 'Checking if %s is a workdir', os.path.abspath(cur_path), ) if cls.is_workdir(cur_path): return os.path.abspath(cur_path) # now search for a .lago directory that's a workdir on any parent dir cur_path = os.path.join(start_path, '.lago') while not cls.is_workdir(cur_path): LOGGER.debug('%s is not a workdir', cur_path) cur_path = os.path.normpath( os.path.join(cur_path, '..', '..', '.lago') ) if os.path.realpath(os.path.join(cur_path, '..')) == '/': raise RuntimeError( 'Unable to find workdir for %s' % os.path.abspath(start_path) ) LOGGER.debug('Checking %s for a workdir', cur_path) return os.path.abspath(cur_path)
@classmethod
[docs] def is_workdir(cls, path): """ Check if the given path is a workdir Args: path(str): Path to check Return: bool: True if the given path is a workdir """ try: cls(path=path).load() except Exception: return False return True
@cli.cli_plugin( help=( 'Change the current prefix link, so the default prefix that is used ' 'is a new one' ), ) @cli.cli_plugin_add_argument( 'prefix_name', action='store', help='Name of the prefix to set as current', ) @utils.in_prefix( prefix_class=prefix.Prefix, workdir_class=Workdir, ) def set_current(prefix_name, parent_workdir, **kwargs): """ Changes the current to point to the given prefix Args: prefix_name(str): name of the prefix to set the current to workdir(str): path to the workdir to change the current of """ parent_workdir.set_current(new_current=prefix_name)