# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config
    stats, error_log = await stop_error_collecting(failed_index)
    if success is False:
        stats = None

    ps.update_execution(execution_id, execution_log, hooks=True)
    ps.finish_execution(execution_id, success, stats, error_log)

    logging.info("%s DONE %s %s %r", execution_id[:8], 'V' if success else 'X', pipeline_id, stats)

    return success, stats, error_log


def execute_pipeline(spec,
                     execution_id,
                     trigger='manual',
                     use_cache=True):

    debug = trigger == 'manual' or os.environ.get('DPP_DEBUG')
    logging.info("%s RUNNING %s", execution_id[:8], spec.pipeline_id)

    loop = asyncio.get_event_loop()

    if debug:
        logging.info("%s Collecting dependencies", execution_id[:8])
    dependencies = {}
    for dep in spec.pipeline_details.get('dependencies', []):
        if 'pipeline' in dep:
            dep_pipeline_id = dep['pipeline']
            pipeline_execution = status_mgr().get(dep_pipeline_id).get_last_successful_execution()
            if pipeline_execution is not None:

# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


# Path: datapackage_pipelines/manager/tasks.py
import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config

    stats, error_log = await stop_error_collecting(failed_index)
    if success is False:
        stats = None

    ps.update_execution(execution_id, execution_log, hooks=True)
    ps.finish_execution(execution_id, success, stats, error_log)

    logging.info("%s DONE %s %s %r", execution_id[:8], 'V' if success else 'X', pipeline_id, stats)

    return success, stats, error_log


def execute_pipeline(spec,
                     execution_id,
                     trigger='manual',
                     use_cache=True):

    debug = trigger == 'manual' or os.environ.get('DPP_DEBUG')
    logging.info("%s RUNNING %s", execution_id[:8], spec.pipeline_id)

    loop = asyncio.get_event_loop()

    if debug:
        logging.info("%s Collecting dependencies", execution_id[:8])
    dependencies = {}
    for dep in spec.pipeline_details.get('dependencies', []):
        if 'pipeline' in dep:
            dep_pipeline_id = dep['pipeline']
            pipeline_execution = status_mgr().get(dep_pipeline_id).get_last_successful_execution()
            if pipeline_execution is not None:

                result_dp = pipeline_execution.stats.get(STATS_DPP_KEY, {}).get(STATS_OUT_DP_URL_KEY)