piecrust2: piecrust/serving/procloop.py comparison

comparison piecrust/serving/procloop.py @ 854:08e02c2a2a1a

core: Keep refactoring, this time to prepare for generator sources. - Make a few APIs simpler. - Content pipelines create their own jobs, so that generator sources can keep aborting in `getContents`, but rely on their pipeline to generate pages for baking.

author	Ludovic Chabant <ludovic@chabant.com>
date	Sun, 04 Jun 2017 23:34:28 -0700
parents	c2ea75e37540
children	448710d84121

comparison

equal deleted inserted replaced

-:f070a4fc033c
+:08e02c2a2a1a
 import queue
 import logging
 import itertools
 import threading
 from piecrust import CONFIG_PATH, THEME_CONFIG_PATH
-from piecrust.app import PieCrust
+from piecrust.pipelines.base import (
-from piecrust.processing.pipeline import ProcessorPipeline
+PipelineJobCreateContext, PipelineJobRunContext, PipelineJobResult,
+PipelineManager)
+from piecrust.pipelines.records import (
+MultiRecord, MultiRecordHistory)
 logger = logging.getLogger(__name__)
 # This flag is for cancelling all long running requests like SSEs.
 logger.debug("Closing pipeline status SSE.")
 self._proc_loop.removeObserver(self)
 self._running = 2
+class _AssetProcessingInfo:
+def __init__(self, source):
+self.source = source
+self.paths = set()
+self.last_bake_time = time.time()
 class ProcessingLoop(threading.Thread):
 def __init__(self, appfactory, out_dir):
-super(ProcessingLoop, self).__init__(
+super().__init__(name='pipeline-reloader', daemon=True)
-name='pipeline-reloader', daemon=True)
 self.appfactory = appfactory
 self.out_dir = out_dir
 self.last_status_id = 0
 self.interval = 1
-self.app = None
+self._app = None
-self._roots = []
+self._proc_infos = None
-self._monitor_assets_root = False
+self._last_records = None
-self._paths = set()
-self._record = None
-self._last_bake = 0
 self._last_config_mtime = 0
 self._obs = []
 self._obs_lock = threading.Lock()
 config_name = (
 THEME_CONFIG_PATH if appfactory.theme_site else CONFIG_PATH)
 self._config_path = os.path.join(appfactory.root_dir, config_name)
 def addObserver(self, obs):
 with self._obs_lock:
 self._obs.append(obs)
 def removeObserver(self, obs):
 with self._obs_lock:
 self._obs.remove(obs)
 def run(self):
-self._initPipeline()
+self._init()
-self._last_bake = time.time()
 self._last_config_mtime = os.path.getmtime(self._config_path)
-self._record = self.pipeline.run()
 while True:
 cur_config_time = os.path.getmtime(self._config_path)
 if self._last_config_mtime < cur_config_time:
 logger.info("Site configuration changed, reloading pipeline.")
 self._last_config_mtime = cur_config_time
-self._initPipeline()
+self._init()
-for root in self._roots:
+self._runPipelines()
-self._runPipeline(root)
+continue
-continue
+for procinfo in self._proc_infos:
-if self._monitor_assets_root:
+# For each assets folder we try to find the first new or
-assets_dir = os.path.join(self.app.root_dir, 'assets')
-if os.path.isdir(assets_dir):
-logger.info("Assets directory was created, reloading "
-"pipeline.")
-self._initPipeline()
-self._runPipeline(assets_dir)
-continue
-for root in self._roots:
-# For each mount root we try to find the first new or
 # modified file. If any, we just run the pipeline on
-# that mount.
+# that source.
 found_new_or_modified = False
-for dirpath, dirnames, filenames in os.walk(root):
+for item in procinfo.source.getAllContents():
-for filename in filenames:
+path = item.spec
-path = os.path.join(dirpath, filename)
+if path not in procinfo.paths:
-if path not in self._paths:
+logger.debug("Found new asset: %s" % path)
-logger.debug("Found new asset: %s" % path)
+procinfo.paths.add(path)
-self._paths.add(path)
+found_new_or_modified = True
-found_new_or_modified = True
-break
-if os.path.getmtime(path) > self._last_bake:
-logger.debug("Found modified asset: %s" % path)
-found_new_or_modified = True
-break
-if found_new_or_modified:
 break
+if os.path.getmtime(path) > procinfo.last_bake_time:
+logger.debug("Found modified asset: %s" % path)
+found_new_or_modified = True
+break
 if found_new_or_modified:
-self._runPipeline(root)
+self._runPipeline(procinfo)
 time.sleep(self.interval)
-def _initPipeline(self):
+def _init(self):
-# Create the app and pipeline.
+self._app = self.appfactory.create()
-self.app = self.appfactory.create()
+self._last_records = MultiRecord()
-self.pipeline = ProcessorPipeline(self.app, self.out_dir)
+self._proc_infos = []
-# Get the list of assets directories.
+for src in self._app.sources:
-self._roots = list(self.pipeline.mounts.keys())
+if src.config['pipeline'] != 'asset':
+continue
-# The 'assets' folder may not be in the mounts list if it doesn't
-# exist yet, but we want to monitor for when the user creates it.
+procinfo = _AssetProcessingInfo(src)
-default_root = os.path.join(self.app.root_dir, 'assets')
+self._proc_infos.append(procinfo)
-self._monitor_assets_root = (default_root not in self._roots)
+# Build the list of initial asset files.
-# Build the list of initial asset files.
+for item in src.getAllContents():
-self._paths = set()
+procinfo.paths.add(item.spec)
-for root in self._roots:
-for dirpath, dirnames, filenames in os.walk(root):
+def _runPipelines(self):
-self._paths |= set([os.path.join(dirpath, f)
+record_histories = MultiRecordHistory(MultiRecord(), self._records)
-for f in filenames])
+self._ppmngr = PipelineManager(
+self._app, self.out_dir, record_histories)
-def _runPipeline(self, root):
-self._last_bake = time.time()
+# Create the pipelines, but also remember some stuff for what
-try:
+# we want to do.
-self._record = self.pipeline.run(
+for src in self._app.sources:
-root,
+if src.config['pipeline'] != 'asset':
-previous_record=self._record,
+continue
-save_record=False)
+ppinfo = self._ppmngr.createPipeline(src)
-status_id = self.last_status_id + 1
+api = _AssetProcessingInfo()
-self.last_status_id += 1
+ppinfo.userdata = api
-if self._record.success:
+current_records = MultiRecord()
-changed = filter(
+record_histories = MultiRecordHistory(
-lambda i: not i.was_collapsed_from_last_run,
+self._records, current_records)
-self._record.entries)
-changed = itertools.chain.from_iterable(
+for ppinfo, procinfo in self._pipelines:
-map(lambda i: i.rel_outputs, changed))
+self._runPipeline(ppinfo, procinfo, record_histories)
-changed = list(changed)
-item = {
+status_id = self.last_status_id + 1
-'id': status_id,
+self.last_status_id += 1
-'type': 'pipeline_success',
-'assets': changed}
+if self._records.success:
+changed = filter(
-self._notifyObservers(item)
+lambda i: not i.was_collapsed_from_last_run,
-else:
+self._record.entries)
-item = {
+changed = itertools.chain.from_iterable(
-'id': status_id,
+map(lambda i: i.rel_outputs, changed))
-'type': 'pipeline_error',
+changed = list(changed)
-'assets': []}
+item = {
-for entry in self._record.entries:
+'id': status_id,
-if entry.errors:
+'type': 'pipeline_success',
-asset_item = {
+'assets': changed}
-'path': entry.path,
-'errors': list(entry.errors)}
+self._notifyObservers(item)
-item['assets'].append(asset_item)
+else:
+item = {
-self._notifyObservers(item)
+'id': status_id,
-except Exception as ex:
+'type': 'pipeline_error',
-logger.exception(ex)
+'assets': []}
+for entry in self._record.entries:
+if entry.errors:
+asset_item = {
+'path': entry.path,
+'errors': list(entry.errors)}
+item['assets'].append(asset_item)
+self._notifyObservers(item)
+def _runPipeline(self, procinfo):
+procinfo.last_bake_time = time.time()
+src = procinfo.source
+current_records = MultiRecord()
+record_histories = MultiRecordHistory(
+self._last_records, current_records)
+ppmngr = PipelineManager(
+self._app, self.out_dir, record_histories)
+ppinfo = ppmngr.createPipeline(src)
+logger.debug("Running pipeline '%s' on: %s" %
+(ppinfo.pipeline_name, src.name))
+# Process all items in the source.
+pp = ppinfo.pipeline
+cr = ppinfo.record_history.current
+jobctx = PipelineJobCreateContext(src)
+for item in src.getAllContents():
+job = pp.createJob(item, jobctx)
+ppres = PipelineJobResult()
+ppres.record_entry = pp.createRecordEntry(job)
+runctx = PipelineJobRunContext(
+ppinfo.pipeline_ctx, job, record_histories)
+try:
+pp.run(item, runctx, ppres)
+except Exception as e:
+ppres.record_entry.errors.append(str(e))
+if ppres.next_pass_job is not None:
+logger.error("The processing loop for the server "
+"doesn't support multi-pass pipelines.")
+cr.addEntry(ppres.record_entry)
+if not ppres.record_entry.success:
+cr.success = False
+current_records.success = False
+logger.error("Errors found in %s:" % item.spec)
+for e in ppres.record_entry.errors:
+logger.error("  " + e)
+# Do all the final stuff.
+ppmngr.buildHistoryDiffs()
+ppmngr.deleteStaleOutputs()
+ppmngr.collapseRecords()
+ppmngr.shutdownPipelines()
+# Swap the old record with the next record.
+pr = ppinfo.record_history.previous
+self._last_records.records.remove(pr)
+self._last_records.records.append(cr)
 def _notifyObservers(self, item):
 with self._obs_lock:
 observers = list(self._obs)
 for obs in observers:

Mercurial > piecrust2

comparison piecrust/serving/procloop.py @ 854:08e02c2a2a1a