# Path: datapackage_pipelines/wrapper/wrapper.py
# def ingest(debug=False):
#     params, datapackage, resource_iterator = _ingest(debug=debug)
#     return ProcessorContext(params, datapackage, resource_iterator)
# 
# Path: datapackage_pipelines/utilities/resources.py
# PROP_STREAMING = 'dpp:streaming'
# 
# Path: datapackage_pipelines/utilities/flow_utils.py
# def spew_flow(flow, ctx: ProcessorContext):
#     flow = Flow(
#         update_package(**ctx.datapackage),
#         load((ctx.datapackage, ctx.resource_iterator)),
#         flow,
#     )
#     datastream = flow.datastream()
#     ctx.datapackage = datastream.dp.descriptor
#     ctx.resource_iterator = datastream.res_iter
#     ctx.stats = MergeableStats(datastream.stats, ctx.stats)
# 
# def load_lazy_json(resources):
# 
#     def func(package):
#         matcher = ResourceMatcher(resources, package.pkg)
#         yield package.pkg
#         for rows in package:
#             if matcher.match(rows.res.name):
#                 yield (
#                     row.inner
#                     if isinstance(row, LazyJsonLine)
#                     else row
#                     for row in rows
#                 )
#             else:
#                 yield rows
# 
#     return func


from dataflows import Flow, join, update_resource
from datapackage_pipelines.wrapper import ingest
from datapackage_pipelines.utilities.resources import PROP_STREAMING
from datapackage_pipelines.utilities.flow_utils import spew_flow, load_lazy_json


def flow(parameters):
    source = parameters['source']
    target = parameters['target']
    return Flow(
        load_lazy_json(source['name']),
        join(
            source['name'],
            source['key'],
            target['name'],
            target['key'],
            parameters['fields'],
            parameters.get('full', None),
            parameters.get('mode', 'half-outer'),
            source.get('delete', False)
        ),
        update_resource(
            target['name'],
            **{
                PROP_STREAMING: True
            }
        )
    )


if __name__ == '__main__':

# Path: datapackage_pipelines/wrapper/wrapper.py
# def ingest(debug=False):
#     params, datapackage, resource_iterator = _ingest(debug=debug)
#     return ProcessorContext(params, datapackage, resource_iterator)
# 
# Path: datapackage_pipelines/utilities/resources.py
# PROP_STREAMING = 'dpp:streaming'
# 
# Path: datapackage_pipelines/utilities/flow_utils.py
# def spew_flow(flow, ctx: ProcessorContext):
#     flow = Flow(
#         update_package(**ctx.datapackage),
#         load((ctx.datapackage, ctx.resource_iterator)),
#         flow,
#     )
#     datastream = flow.datastream()
#     ctx.datapackage = datastream.dp.descriptor
#     ctx.resource_iterator = datastream.res_iter
#     ctx.stats = MergeableStats(datastream.stats, ctx.stats)
# 
# def load_lazy_json(resources):
# 
#     def func(package):
#         matcher = ResourceMatcher(resources, package.pkg)
#         yield package.pkg
#         for rows in package:
#             if matcher.match(rows.res.name):
#                 yield (
#                     row.inner
#                     if isinstance(row, LazyJsonLine)
#                     else row
#                     for row in rows
#                 )
#             else:
#                 yield rows
# 
#     return func


# Path: datapackage_pipelines/lib/join.py
from dataflows import Flow, join, update_resource
from datapackage_pipelines.wrapper import ingest
from datapackage_pipelines.utilities.resources import PROP_STREAMING
from datapackage_pipelines.utilities.flow_utils import spew_flow, load_lazy_json



def flow(parameters):
    source = parameters['source']
    target = parameters['target']
    return Flow(
        load_lazy_json(source['name']),
        join(
            source['name'],
            source['key'],
            target['name'],
            target['key'],
            parameters['fields'],
            parameters.get('full', None),
            parameters.get('mode', 'half-outer'),
            source.get('delete', False)
        ),
        update_resource(
            target['name'],
            **{
                PROP_STREAMING: True
            }
        )
    )


if __name__ == '__main__':

    with ingest() as ctx: