CSV Import

Category: Data & Analytics

Get this pack →

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

Overview

Accept a posted CSV body, parse it into rows, and bulk-insert them into a Postgres contacts table. Demonstrates parse_csv over the raw request body plus a lookup fan-out insert.

Endpoints

• POST contacts/import — parse a posted CSV and bulk-insert rows

If any row is refused, the request answers 422 ("some rows were not imported") and lists the refused rows, with their errors, under CheckRows.data.failed. The other rows are stored, and sending the CSV again is safe: an email that is already stored is skipped.

Required variables

• ap_var::DATABASE_URL — Postgres connection string

Configuration

csv_import.yml

name: CsvImport
description: >
  Accept a posted CSV body, parse it into rows, and bulk-insert them into a
  Postgres contacts table. Demonstrates parse_csv over the raw request body plus
  a lookup fan-out insert.

docs: true

# Required managed variables:
#   DATABASE_URL — Postgres connection string

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"
interfaces:
  contacts/import:
    output: http
    method: POST
    summary: Parse a posted CSV and bulk-insert the rows
    actions:
      - name: ParseCsv
        input: a|raw_body|
        post_transforms:
          - parse_csv: "$"
      # lookup_partition sorts the rows into succeeded and failed. Without it a lookup fails
      # only when every row fails, so a CSV where some rows were refused answered "imported".
      - name: InsertRows
        run_when_succeeded: [ParseCsv]
        lookup: a|ParseCsv|
        lookup_partition: true
        actions:
          # Values are bound parameters, never pasted into the SQL text: a name like
          # "Mary O'Brien" is stored as written, and nothing in the CSV can change the statement.
          - name: InsertOne
            database: main
            query: |-
              INSERT INTO contacts (name, email)
              VALUES ($1::text, $2::text)
              ON CONFLICT (email) DO NOTHING
            params:
              - a|body::name|
              - a|body::email|
      # Any refused row fails the request (422), and the rows that failed, with their errors,
      # are in this step's data. The rows that went in stay in: re-sending the CSV is safe,
      # since an email already stored is skipped.
      - name: CheckRows
        run_when_succeeded: [InsertRows]
        input: a|InsertRows|
        assert:
          error_message: "some rows were not imported"
          http_code_on_error: 422
          tests:
            - value: failed
              is_empty: true
      - name: Respond
        run_when_succeeded: [CheckRows]
        json_output: |-
          { "status": "imported" }

seed.yml

name: CsvImportSeed
description: Creates the contacts table used by the import endpoint. Safe to re-run.
docs: true
global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"
interfaces:
  api/seed:
    output: http
    method: POST
    summary: Create and clear the contacts table
    actions:
      - name: CreateContactsTable
        database: main
        hide_data_on_success: true
        query: |
          CREATE TABLE IF NOT EXISTS contacts (
            id    BIGSERIAL   PRIMARY KEY,
            name  TEXT,
            email TEXT UNIQUE NOT NULL
          );
      - name: TruncateContacts
        run_when_succeeded: [CreateContactsTable]
        database: main
        hide_data_on_success: true
        query: TRUNCATE contacts RESTART IDENTITY;
      - name: Respond
        run_when_succeeded: [TruncateContacts]
        json_output: |-
          { "status": "seeded" }