python-webassets 3:0.10.1-2 / usr / lib / python2.7 / dist-packages / webassets / script.py

from __future__ import print_function
import shutil
import os, sys
import time
import logging

from webassets.loaders import PythonLoader, YAMLLoader
from webassets.bundle import get_all_bundle_files
from webassets.exceptions import BuildError
from webassets.updater import TimestampUpdater
from webassets.merge import MemoryHunk
from webassets.version import get_manifest
from webassets.cache import FilesystemCache
from webassets.utils import set, StringIO


__all__ = ('CommandError', 'CommandLineEnvironment', 'main')


# logging has WARNING as default level, for the CLI we want INFO. Set this
# as early as possible, so that user customizations will not be overwritten.
logging.getLogger('webassets.script').setLevel(logging.INFO)


class CommandError(Exception):
    pass


class Command(object):
    """Base-class for a command used by :class:`CommandLineEnvironment`.

    Each command being a class opens up certain possibilities with respect to
    subclassing and customizing the default CLI.
    """

    def __init__(self, cmd_env):
        self.cmd = cmd_env

    def __getattr__(self, name):
        # Make stuff from cmd environment easier to access
        return getattr(self.cmd, name)

    def __call__(self, *args, **kwargs):
        raise NotImplementedError()


class BuildCommand(Command):

    def __call__(self, bundles=None, output=None, directory=None, no_cache=None,
              manifest=None, production=None):
        """Build assets.

        ``bundles``
            A list of bundle names. If given, only this list of bundles
            should be built.

        ``output``
            List of (bundle, filename) 2-tuples. If given, only these
            bundles will be built, using the custom output filenames.
            Cannot be used with ``bundles``.

        ``directory``
            Custom output directory to use for the bundles. The original
            basenames defined in the bundle ``output`` attribute will be
            used. If the ``output`` of the bundles are pointing to different
            directories, they will be offset by their common prefix.
            Cannot be used with ``output``.

        ``no_cache``
            If set, a cache (if one is configured) will not be used.

        ``manifest``
            If set, the given manifest instance will be used, instead of
            any that might have been configured in the Environment. The value
            passed will be resolved through ``get_manifest()``. If this fails,
            a file-based manifest will be used using the given value as the
            filename.

        ``production``
            If set to ``True``, then :attr:`Environment.debug`` will forcibly
            be disabled (set to ``False``) during the build.
        """

        # Validate arguments
        if bundles and output:
            raise CommandError(
                'When specifying explicit output filenames you must '
                'do so for all bundles you want to build.')
        if directory and output:
            raise CommandError('A custom output directory cannot be '
                               'combined with explicit output filenames '
                               'for individual bundles.')

        if production:
            # TODO: Reset again (refactor commands to be classes)
            self.environment.debug = False

        # TODO: Oh how nice it would be to use the future options stack.
        if manifest is not None:
            try:
                manifest = get_manifest(manifest, env=self.environment)
            except ValueError:
                manifest = get_manifest(
                    # abspath() is important, or this will be considered
                    # relative to Environment.directory.
                    "file:%s" % os.path.abspath(manifest),
                    env=self.environment)
            self.environment.manifest = manifest

        # Use output as a dict.
        if output:
            output = dict(output)

        # Validate bundle names
        bundle_names = bundles if bundles else (output.keys() if output else [])
        for name in bundle_names:
            if not name in self.environment:
                raise CommandError(
                    'I do not know a bundle name named "%s".' % name)

        # Make a list of bundles to build, and the filename to write to.
        if bundle_names:
            # TODO: It's not ok to use an internal property here.
            bundles = [(n,b) for n, b in self.environment._named_bundles.items()
                             if n in bundle_names]
        else:
            # Includes unnamed bundles as well.
            bundles = [(None, b) for b in self.environment]

        # Determine common prefix for use with ``directory`` option.
        if directory:
            prefix = os.path.commonprefix(
                [os.path.normpath(b.resolve_output())
                 for _, b in bundles if b.output])
            # dirname() gives the right value for a single file.
            prefix = os.path.dirname(prefix)

        to_build = []
        for name, bundle in bundles:
            # TODO: We really should support this. This error here
            # is just in place of a less understandable error that would
            # otherwise occur.
            if bundle.is_container and directory:
                raise CommandError(
                    'A custom output directory cannot currently be '
                    'used with container bundles.')

            # Determine which filename to use, if not the default.
            overwrite_filename = None
            if output:
                overwrite_filename = output[name]
            elif directory:
                offset = os.path.normpath(
                    bundle.resolve_output())[len(prefix)+1:]
                overwrite_filename = os.path.join(directory, offset)
            to_build.append((bundle, overwrite_filename, name,))

        # Build.
        built = []
        for bundle, overwrite_filename, name in to_build:
            if name:
                # A name is not necessary available of the bundle was
                # registered without one.
                self.log.info("Building bundle: %s (to %s)" % (
                    name, overwrite_filename or bundle.output))
            else:
                self.log.info("Building bundle: %s" % bundle.output)

            try:
                if not overwrite_filename:
                    with bundle.bind(self.environment):
                        bundle.build(force=True, disable_cache=no_cache)
                else:
                    # TODO: Rethink how we deal with container bundles here.
                    # As it currently stands, we write all child bundles
                    # to the target output, merged (which is also why we
                    # create and force writing to a StringIO instead of just
                    # using the ``Hunk`` objects that build() would return
                    # anyway.
                    output = StringIO()
                    with bundle.bind(self.environment):
                        bundle.build(force=True, output=output,
                            disable_cache=no_cache)
                    if directory:
                        # Only auto-create directories in this mode.
                        output_dir = os.path.dirname(overwrite_filename)
                        if not os.path.exists(output_dir):
                            os.makedirs(output_dir)
                    MemoryHunk(output.getvalue()).save(overwrite_filename)
                built.append(bundle)
            except BuildError as e:
                self.log.error("Failed, error was: %s" % e)
        if len(built):
            self.event_handlers['post_build']()
        if len(built) != len(to_build):
            return 2


class WatchCommand(Command):

    def __call__(self, loop=None):
        """Watch assets for changes.

        ``loop``
            A callback, taking no arguments, to be called once every loop
            iteration. Can be useful to integrate the command with other code.
            If not specified, the loop wil call ``time.sleep()``.
        """
        # TODO: This should probably also restart when the code changes.
        mtimes = {}

        try:
            # Before starting to watch for changes, also recognize changes
            # made while we did not run, and apply those immediately.
            for bundle in self.environment:
                print('Bringing up to date: %s' % bundle.output)
                bundle.build(force=False)

            self.log.info("Watching %d bundles for changes..." %
                          len(self.environment))

            while True:
                changed_bundles = self.check_for_changes(mtimes)

                built = []
                for bundle in changed_bundles:
                    print("Building bundle: %s ..." % bundle.output, end=' ')
                    sys.stdout.flush()
                    try:
                        bundle.build(force=True)
                        built.append(bundle)
                    except BuildError as e:
                        print("")
                        print("Failed: %s" % e)
                    else:
                        print("done")

                if len(built):
                    self.event_handlers['post_build']()

                do_end = loop() if loop else time.sleep(0.1)
                if do_end:
                    break
        except KeyboardInterrupt:
            pass

    def check_for_changes(self, mtimes):
        # Do not update original mtimes dict right away, so that we detect
        # all bundle changes if a file is in multiple bundles.
        _new_mtimes = mtimes.copy()

        changed_bundles = set()
        # TODO: An optimization was lost here, skipping a bundle once
        # a single file has been found to have changed. Bring back.
        for filename, bundles_to_update in self.yield_files_to_watch():
            stat = os.stat(filename)
            mtime = stat.st_mtime
            if sys.platform == "win32":
                mtime -= stat.st_ctime

            if mtimes.get(filename, mtime) != mtime:
                if callable(bundles_to_update):
                    # Hook for when file has changed
                    try:
                        bundles_to_update = bundles_to_update()
                    except EnvironmentError:
                        # EnvironmentError is what the hooks is allowed to
                        # raise for a temporary problem, like an invalid config
                        import traceback
                        traceback.print_exc()
                        # Don't update anything, wait for another change
                        bundles_to_update = set()

                if bundles_to_update is True:
                    # Indicates all bundles should be rebuilt for the change
                    bundles_to_update = set(self.environment)
                changed_bundles |= bundles_to_update
                _new_mtimes[filename] = mtime
            _new_mtimes[filename] = mtime

        mtimes.update(_new_mtimes)
        return changed_bundles

    def yield_files_to_watch(self):
        for bundle in self.environment:
            for filename in get_all_bundle_files(bundle):
                yield filename, set([bundle])


class CleanCommand(Command):

    def __call__(self):
        """Delete generated assets.
        """
        self.log.info('Cleaning generated assets...')
        for bundle in self.environment:
            if not bundle.output:
                continue
            file_path = bundle.resolve_output(self.environment)
            if os.path.exists(file_path):
                os.unlink(file_path)
                self.log.info("Deleted asset: %s" % bundle.output)
        if isinstance(self.environment.cache, FilesystemCache):
            shutil.rmtree(self.environment.cache.directory)


class CheckCommand(Command):

    def __call__(self):
        """Check to see if assets need to be rebuilt.

        A non-zero exit status will be returned if any of the input files are
        newer (based on mtime) than their output file. This is intended to be
        used in pre-commit hooks.
        """
        needsupdate = False
        updater = self.environment.updater
        if not updater:
            self.log.debug('no updater configured, using TimestampUpdater')
            updater = TimestampUpdater()
        for bundle in self.environment:
            self.log.info('Checking asset: %s', bundle.output)
            if updater.needs_rebuild(bundle, self.environment):
                self.log.info('  needs update')
                needsupdate = True
        if needsupdate:
            sys.exit(-1)


class CommandLineEnvironment(object):
    """Implements the core functionality for a command line frontend to
    ``webassets``, abstracted in a way to allow frameworks to integrate the
    functionality into their own tools, for example, as a Django management
    command, or a command for ``Flask-Script``.
    """

    def __init__(self, env, log, post_build=None, commands=None):
        self.environment = env
        self.log = log
        self.event_handlers = dict(post_build=lambda: True)
        if callable(post_build):
            self.event_handlers['post_build'] = post_build

        # Instantiate each command
        command_def = self.DefaultCommands.copy()
        command_def.update(commands or {})
        self.commands = {}
        for name, construct in command_def.items():
            if not construct:
                continue
            if not isinstance(construct, (list, tuple)):
                construct = [construct, (), {}]
            self.commands[name] = construct[0](
                self, *construct[1], **construct[2])

    def __getattr__(self, item):
        # Allow method-like access to commands.
        if item in self.commands:
            return self.commands[item]
        raise AttributeError(item)

    def invoke(self, command, args):
        """Invoke ``command``, or throw a CommandError.

        This is essentially a simple validation mechanism. Feel free
        to call the individual command methods manually.
        """
        try:
            function = self.commands[command]
        except KeyError as e:
            raise CommandError('unknown command: %s' % e)
        else:
            return function(**args)

    # List of commands installed
    DefaultCommands = {
        'build': BuildCommand,
        'watch': WatchCommand,
        'clean': CleanCommand,
        'check': CheckCommand
    }


class GenericArgparseImplementation(object):
    """Generic command line utility to interact with an webassets environment.

    This is effectively a reference implementation of a command line utility
    based on the ``CommandLineEnvironment`` class. Implementers may find it
    feasible to simple base their own command line utility on this, rather than
    implementing something custom on top of ``CommandLineEnvironment``. In
    fact, if that is possible, you are encouraged to do so for greater
    consistency across implementations.
    """

    class WatchCommand(WatchCommand):
        """Extended watch command that also looks at the config file itself."""

        def __init__(self, cmd_env, argparse_ns):
            WatchCommand.__init__(self, cmd_env)
            self.ns = argparse_ns

        def yield_files_to_watch(self):
            for result in WatchCommand.yield_files_to_watch(self):
                yield result
            # If the config changes, rebuild all bundles
            if getattr(self.ns, 'config', None):
                yield self.ns.config, self.reload_config

        def reload_config(self):
            try:
                self.cmd.environment = YAMLLoader(self.ns.config).load_environment()
            except Exception as e:
                raise EnvironmentError(e)
            return True


    def __init__(self, env=None, log=None, prog=None, no_global_options=False):
        try:
            import argparse
        except ImportError:
            raise RuntimeError(
                'The webassets command line now requires the '
                '"argparse" library on Python versions <= 2.6.')
        else:
            self.argparse = argparse
        self.env = env
        self.log = log
        self._construct_parser(prog, no_global_options)

    def _construct_parser(self, prog=None, no_global_options=False):
        self.parser = parser = self.argparse.ArgumentParser(
            description="Manage assets.",
            prog=prog)

        if not no_global_options:
            # Start with the base arguments that are valid for any command.
            # XXX: Add those to the subparser?
            parser.add_argument("-v", dest="verbose", action="store_true",
                help="be verbose")
            parser.add_argument("-q", action="store_true", dest="quiet",
                help="be quiet")
            if self.env is None:
                loadenv = parser.add_mutually_exclusive_group()
                loadenv.add_argument("-c", "--config", dest="config",
                    help="read environment from a YAML file")
                loadenv.add_argument("-m", "--module", dest="module",
                    help="read environment from a Python module")

        # Add subparsers.
        subparsers = parser.add_subparsers(dest='command')
        for command in CommandLineEnvironment.DefaultCommands.keys():
            command_parser = subparsers.add_parser(command)
            maker = getattr(self, 'make_%s_parser' % command, False)
            if maker:
                maker(command_parser)

    @staticmethod
    def make_build_parser(parser):
        parser.add_argument(
            'bundles', nargs='*', metavar='BUNDLE',
            help='Optional bundle names to process. If none are '
                 'specified, then all known bundles will be built.')
        parser.add_argument(
            '--output', '-o', nargs=2, action='append',
            metavar=('BUNDLE', 'FILE'),
            help='Build the given bundle, and use a custom output '
                 'file. Can be given multiple times.')
        parser.add_argument(
            '--directory', '-d',
            help='Write built files to this directory, using the '
                 'basename defined by the bundle. Will offset '
                 'the original bundle output paths on their common '
                 'prefix. Cannot be used with --output.')
        parser.add_argument(
            '--no-cache', action='store_true',
            help='Do not use a cache that might be configured.')
        parser.add_argument(
            '--manifest',
            help='Write a manifest to the given file. Also supports '
                 'the id:arg format, if you want to use a different '
                 'manifest implementation.')
        parser.add_argument(
            '--production', action='store_true',
            help='Forcably turn off debug mode for the build. This '
                 'only has an effect if debug is set to "merge".')

    def _setup_logging(self, ns):
        if self.log:
            log = self.log
        else:
            log = logging.getLogger('webassets.script')
            if not log.handlers:
                # In theory, this could run multiple times (e.g. tests)
                handler = logging.StreamHandler()
                log.addHandler(handler)
                # Note that setting the level filter at the handler level is
                # better than the logger level, since this is "our" handler,
                # we create it, for the purposes of having a default output.
                # The logger itself the user may be modifying.
                handler.setLevel(logging.DEBUG if ns.verbose else (
                    logging.WARNING if ns.quiet else logging.INFO))
        return log

    def _setup_assets_env(self, ns, log):
        env = self.env
        if env is None:
            assert not (ns.module and ns.config)
            if ns.module:
                env = PythonLoader(ns.module).load_environment()
            if ns.config:
                env = YAMLLoader(ns.config).load_environment()
        return env

    def _setup_cmd_env(self, assets_env, log, ns):
        return CommandLineEnvironment(assets_env, log, commands={
            'watch': (GenericArgparseImplementation.WatchCommand, (ns,), {})
        })

    def _prepare_command_args(self, ns):
        # Prepare a dict of arguments cleaned of values that are not
        # command-specific, and which the command method would not accept.
        args = vars(ns).copy()
        for action in self.parser._actions:
            dest = action.dest
            if dest in args:
                del args[dest]
        return args

    def run_with_ns(self, ns):
        log = self._setup_logging(ns)
        env = self._setup_assets_env(ns, log)
        if env is None:
            raise CommandError(
                "Error: No environment given or found. Maybe use -m?")
        cmd = self._setup_cmd_env(env, log, ns)

        # Run the selected command
        args = self._prepare_command_args(ns)
        return cmd.invoke(ns.command, args)

    def run_with_argv(self, argv):
        try:
            ns = self.parser.parse_args(argv)
        except SystemExit as e:
            # We do not want the main() function to exit the program.
            # See run() instead.
            return e.args[0]

        return self.run_with_ns(ns)

    def main(self, argv):
        """Parse the given command line.

        The commandline is expected to NOT including what would be sys.argv[0].
        """
        try:
            return self.run_with_argv(argv)
        except CommandError as e:
            print(e)
            return 1


def main(argv, env=None):
    """Execute the generic version of the command line interface.

    You only need to work directly with ``GenericArgparseImplementation`` if
    you desire to customize things.

    If no environment is given, additional arguments will be supported to allow
    the user to specify/construct the environment on the command line.
    """
    return GenericArgparseImplementation(env).main(argv)


def run():
    """Runs the command line interface via ``main``, then exits the process
    with a proper return code."""
    sys.exit(main(sys.argv[1:]) or 0)


if __name__ == '__main__':
    run()