Ecommerce Module EN

繁體中文版

Ecommerce Module — beam_pipeline_toolkit/ecommerce/

An optional, ready-to-adapt extension for e-commerce-shaped pipelines: sessions, orders, customers, and a semantic daily metric row. Built entirely on the public API of pipeline/, controlplane/, log/, safety/ — proof that the generic core needs no changes to support a real domain. No single company's field names or business rules are hardcoded here; adapt it, don't treat it as authoritative.

schemas.py

NamedTuples: SessionRecord, OrderRecord, CustomerRecord, DailyMetricRow (sessions_count, conversion_rate, order_count, average_order_value, revenue, registered_members), InvalidRecord (deliberately carries no raw payload — only enough to attribute a failure to a key + reason code). Money fields are decimal-literal strings, never binary floats.

metrics.py

compute_daily_metrics(sessions_count, order_count, revenue, registered_members) — pure function, Decimal-safe. average_order_value = revenue / order_count (0 if no orders), conversion_rate = order_count / sessions_count (0 if no sessions), both rounded with ROUND_HALF_UP.

order_policies.py

Pluggable order-validity policy pattern — different reporting paths can disagree about which orders count (cancelled? returned? a deleted_at flag?). Model each rule as its own function and choose explicitly, or run two to see exactly where they diverge.

• exclude_cancelled_and_returned(order) — the generic default: excludes cancelled/canceled status and returned/refunded return_status.
• accept_all_orders(order) — trivial baseline, always True.
• partition_by_policy(orders, policy) — ad hoc/offline helper splitting into (eligible, excluded).

validation.py

• build_validate_fn(required_fields, date_field="metric_date") — returns a validate_fn compatible with pipeline.validation.Validate: required-field presence, ISO date validity, non-negative sessions_count for session events, decimal-parseable total_price for order events.
• deduplicate_sessions(sessions, sequence_field="sequence") — keeps the first record (by sequence) among duplicates for the same key; returns (kept, duplicate_count). Needs a grouped view, so it's called from build_outcome_fn, not validate_fn.

pipeline.py

The complete worked example, composing everything above with the generic core directly (not via pipeline.main.build_pipeline, since it needs the extra classification/filter step order_policy provides):

raw_messages
  → ParseNormalize
  → Validate (build_validate_fn)
  → TagOrigin (session / customer / order) or TagInvalid
  → AggregateAndEmit, keyed by (run_id, shop_id, metric_date)
      → build_outcome_fn:
          - deduplicate_sessions
          - partition orders by order_policy
          - compute_daily_metrics
          - 4-stage telemetry timeline (SOURCE_READ, VALIDATE, FILTER, AGGREGATE)
          - invalid_summary (duplicate + structural counts)

• default_schema(metrics_namespace="ecommerce_pipeline") — the module's own PipelineSchema with stages {SOURCE_READ, VALIDATE, FILTER, AGGREGATE} and statuses {PASS, WARNING, COMPLETED}.
• build_outcome_fn_factory(schema, order_policy=exclude_cancelled_and_returned) — builds the build_outcome_fn described above.
• build_ecommerce_pipeline(raw_messages, schema=None, required_fields=DEFAULT_REQUIRED_FIELDS, order_policy=exclude_cancelled_and_returned) — the entry point. Copy this module into your own app and swap in real field names/rules, or call it directly if this shape already fits.