Be more conservative with dictifying stuff
@@ -115,9 +115,11 @@ def get_assignment(result, skip=False, **kwargs):
return one, iter([first_result]) if one else result
-def assign(item, assignment, key, one=False):
- value = next(assignment) if one else list(assignment)
- yield DotDict(merge([item, {key: value}]))
+def assign(item, assignment, **kwargs):
+ key = kwargs.get('assign')
+ value = next(assignment) if kwargs.get('one') else list(assignment)
+ merged = merge([item, {key: value}])
+ yield DotDict(merged) if kwargs.get('dictize') else merged
class processor(object):
@@ -323,9 +325,9 @@ class processor(object):
conf = {k: combined[k] for k in self.defaults}
conf.update(kwargs.get('conf', {}))
combined.update({'conf': conf})
- # replace conf with dictized version so we can access its
- # attributes even if we already extracted a value
- updates = {'conf': DotDict(conf), 'assign': combined.get('assign')}
+
+ uconf = DotDict(conf) if combined.get('dictize') else conf
+ updates = {'conf': uconf, 'assign': combined.get('assign')}
kwargs.update(updates)
item = item or {}
@@ -352,8 +354,7 @@ class processor(object):
if skip or combined.get('emit'):
stream = assignment
elif not skip:
- key = combined.get('assign')
- stream = assign(_input, assignment, key, one=one)
+ stream = assign(_input, assignment, one=one, **combined)
if self.async:
return_value(stream)
@@ -588,9 +589,8 @@ class operator(object):
conf.update(kwargs.get('conf', {}))
combined.update({'conf': conf})
- # replace conf with dictized version so we can access its
- # attributes even if we already extracted a value
- updates = {'conf': DotDict(conf), 'assign': combined.get('assign')}
+ uconf = DotDict(conf) if combined.get('dictize') else conf
+ updates = {'conf': uconf, 'assign': combined.get('assign')}
kwargs.update(updates)
items = items or iter([])
@@ -630,8 +630,9 @@ class operator(object):
stream = assignment
else:
singles = (iter([v]) for v in assignment)
- key = combined.get('assign')
- assigned = (assign({}, s, key, one=True) for s in singles)
+ assigned = (
+ assign({}, s, one=True, **combined) for s in singles)
+
stream = multiplex(assigned)
if self.async:
