join.py

fromdataflowsimportFlow, join, update_resource
fromdatapackage_pipelines.wrapperimportingest
fromdatapackage_pipelines.utilities.resourcesimportPROP_STREAMING
fromdatapackage_pipelines.utilities.flow_utilsimportspew_flow, load_lazy_json


defflow(parameters):
source=parameters['source']
target=parameters['target']
returnFlow(
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__':
withingest() asctx:
spew_flow(flow(ctx.parameters), ctx)