Mercurial > piecrust2
view piecrust/processing/less.py @ 215:a47580a0955b
bake: Better error handling for the processing pipeline.
Pipeline jobs now keep track of whether they've seen any errors. This is
aggregated into an overall "success" flag for the processing record. Also, jobs
keep going as long as there's no critical (i.e. internal) failure happening.
Errors raised by processors are also better tracked: the actual processor that
failed, along with the input file, are tracks in the processing record.
The `bake` command returns a failure exit code if processing saw any error.
author | Ludovic Chabant <ludovic@chabant.com> |
---|---|
date | Sat, 31 Jan 2015 17:08:02 -0800 |
parents | b4724e577a8c |
children | f82262f59600 |
line wrap: on
line source
import os import os.path import json import hashlib import logging import platform import subprocess from piecrust.processing.base import SimpleFileProcessor from piecrust.processing.tree import FORCE_BUILD logger = logging.getLogger(__name__) class LessProcessor(SimpleFileProcessor): PROCESSOR_NAME = 'less' def __init__(self): super(LessProcessor, self).__init__({'less': 'css'}) self._conf = None self._map_dir = None def onPipelineStart(self, pipeline): self._map_dir = os.path.join(pipeline.tmp_dir, 'less') if not os.path.isdir(self._map_dir): os.makedirs(self._map_dir) def getDependencies(self, path): map_path = self._getMapPath(path) try: with open(map_path, 'r') as f: dep_map = json.load(f) # Check the version, since the `sources` list has changed # meanings over time. if dep_map.get('version') != 3: logger.warning("Unknown LESS map version. Force rebuilding.") return FORCE_BUILD # Get the sources, but make all paths absolute. sources = dep_map.get('sources') path_dir = os.path.dirname(path) def _makeAbs(p): return os.path.join(path_dir, p) deps = list(map(_makeAbs, sources)) return [map_path] + deps except IOError: # Map file not found... rebuild. logger.debug("No map file found for LESS file '%s' at '%s'. " "Rebuilding" % (path, map_path)) return FORCE_BUILD def _doProcess(self, in_path, out_path): self._ensureInitialized() map_path = self._getMapPath(in_path) map_path = os.path.relpath(map_path) args = [self._conf['bin'], '--source-map=%s' % map_path] args += self._conf['options'] args.append(in_path) args.append(out_path) logger.debug("Processing LESS file: %s" % args) # On Windows, we need to run the process in a shell environment # otherwise it looks like `PATH` isn't taken into account. shell = (platform.system() == 'Windows') try: retcode = subprocess.call(args, shell=shell) except FileNotFoundError as ex: logger.error("Tried running LESS processor with command: %s" % args) raise Exception("Error running LESS processor. " "Did you install it?") from ex if retcode != 0: raise Exception("Error occured in LESS compiler. Please check " "log messages above for more information.") return True def _ensureInitialized(self): if self._conf is not None: return self._conf = self.app.config.get('less') or {} self._conf.setdefault('bin', 'lessc') self._conf.setdefault('options', ['--compress']) if not isinstance(self._conf['options'], list): raise Exception("The `less/options` configuration setting " "must be an array of arguments.") def _getMapPath(self, path): map_name = "%s_%s.map" % ( os.path.basename(path), hashlib.md5(path.encode('utf8')).hexdigest()) map_path = os.path.join(self._map_dir, map_name) return map_path