Events Filter Insert

Category: Data & Analytics

Get this pack →

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Overview

Ingest an event, flag it high-value (amount > 100) with the math transform, and insert only when it is. Demonstrates the math transform and a conditional insert.

Endpoints

• POST events/ingest — flags an event high-value and inserts only if so

The answer's inserted is true only when a row was stored: it is false for a low-value event and for an id that is already stored (the first row is kept). An insert the database refuses fails the request rather than reporting success.

Required variables

• ap_var::DATABASE_URL — Postgres connection string

Configuration

events_filter_insert.yml

name: EventsFilterInsert
description: >
  Ingest an event and flag whether it is high-value (amount > 100) with the
  math transform, inserting the row only when it is. Demonstrates the math
  transform and run_on_assertion as a conditional guard.

docs: true

# Required managed variables:
#   DATABASE_URL  — Postgres connection string

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"
interfaces:
  events/ingest:
    output: http
    method: POST
    summary: Flag an event high-value (amount > 100) with math and insert only if so
    actions:
      - name: ReadBody
        input: a|body|
        assert:
          error_message: "id and amount are required"
          http_code_on_error: 400
          tests:
            - value: id
              is_not_empty: true
            - value: amount
              is_not_null: true
      - name: Score
        run_when_succeeded: [ReadBody]
        input: a|body|
        post_transforms:
          - math:
              expressions:
                high_value: "target.amount > 100"
              eval_type: bool
      # Values are bound parameters, never pasted into the SQL text: an id or kind like
      # "o'reilly" is stored as written, and nothing in the request can change the statement.
      # The statement answers whether a row went in: false for a low-value event, and false
      # for an id already stored (ON CONFLICT DO NOTHING keeps the first row).
      - name: InsertHighValue
        run_when_succeeded: [Score]
        database: main
        query: |-
          WITH ins AS (
            INSERT INTO events (id, amount, kind)
            SELECT $1::text, $2::text::numeric, $3::text
            WHERE $2::text::numeric > 100
            ON CONFLICT (id) DO NOTHING
            RETURNING id
          )
          SELECT EXISTS (SELECT 1 FROM ins) AS inserted
        params:
          - a|ReadBody::id|
          - a|ReadBody::amount|
          - a|ReadBody::kind->default('')|
        post_transforms:
          - extract_value: "[0]"
      # Answers only after the insert, so a failed insert fails the request instead of
      # reporting success. ->json_escape keeps an id with a quote in it valid JSON.
      - name: Respond
        run_when_succeeded: [InsertHighValue]
        json_output: |-
          {
            "id": "a|ReadBody::id->json_escape|",
            "high_value": a|Score::high_value|,
            "inserted": a|InsertHighValue::inserted|
          }

seed.yml

name: EventsFilterInsertSeed
description: Creates the events table used by the ingest endpoint. Safe to re-run (starts empty; the pack inserts high-value events).

docs: true

# Required managed variables:
#   DATABASE_URL — Postgres connection string

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:
  api/seed:
    output: http
    method: POST
    summary: Create and clear the events table
    actions:
      - name: CreateEventsTable
        database: main
        hide_data_on_success: true
        query: |
          CREATE TABLE IF NOT EXISTS events (
            id     TEXT     PRIMARY KEY,
            amount NUMERIC  NOT NULL,
            kind   TEXT
          );
      - name: TruncateEvents
        run_when_succeeded: [CreateEventsTable]
        database: main
        hide_data_on_success: true
        query: TRUNCATE events;
      - name: Respond
        run_when_succeeded: [TruncateEvents]
        json_output: |-
          { "status": "seeded", "events": 0 }