Description

Login

Syncing account progress…

Back

Advanced · 18 min

Project: event pipeline

Combine validation, generators, data classes, and aggregation.

Validate at the boundary

Convert raw dictionaries into trusted records before calculating. Rejecting bad input early keeps later code simpler.

def valid(event):
    return "type" in event and isinstance(event.get("value"), int)

events = [{"type": "view", "value": 2}, {"value": 4}]
print([event for event in events if valid(event)])

Output

[{'type': 'view', 'value': 2}]

Stream then aggregate

A generator can normalize accepted records while a dictionary accumulates totals by category.

events = [{"type": "view", "value": 2}, {"type": "view", "value": 3}]
totals = {}
for event in events:
    totals[event["type"]] = totals.get(event["type"], 0) + event["value"]
print(totals)

Output

{'view': 5}

Project: event pipeline

  1. Read both project: event pipeline examples before answering.
  2. Run the task, compare its exact output, and revise the code if needed.

Try it yourself

Code runs on this device. When you are signed in, drafts sync to your account.

Build the pipeline

Complete summarize so it ignores events without a string type or integer value and totals valid values by type. Display the supplied summary.

Practice