payments: warehouse staging and source tables for enghouse data

warehouse/models/staging/payments/enghouse/_enghouse.yml

@@ -0,0 +1,155 @@
+version: 2
+
+sources:
+  - name: external_enghouse
+    description: Hive-partitioned external tables reading Enghouse payments data.
+    database: "{{ env_var('GOOGLE_CLOUD_PROJECT', var('GOOGLE_CLOUD_PROJECT')) }}"
+    schema: external_enghouse
+    tables:
+      - name: taps
+      - name: tickets
+      - name: transactions
+      - name: pay_windows
+
+models:
+  - name: stg_enghouse__taps
+    description: Add description
+    columns:
+      - name: tap_id
+        description: To do
+      - name: mapping_terminal_id
+        description: To do
+      - name: mapping_merchant_id
+        description: To do
+      - name: terminal
+        description: To do
+      - name: token
+        description: To do
+      - name: masked_pan
+        description: To do
+      - name: expiry
+        description: To do
+      - name: server_date
+        description: To do
+      - name: terminal_date
+        description: To do
+      - name: tx_number
+        description: To do
+      - name: tx_status
+        description: To do
+      - name: payment_reference
+        description: To do
+      - name: terminal_spdh_code
+        description: To do
+      - name: denylist_version
+        description: To do
+      - name: transit_data
+        description: To do
+      - name: currency
+        description: To do
+      - name: par
+        description: To do
+
+  - name: stg_enghouse__tickets
+    description: Add description
+    columns:
+      - name: id
+        description: To do
+      - name: ticket_id
+        description: To do
+      - name: station_name
+        description: To do
+      - name: amount
+        description: To do
+      - name: clearing_id
+        description: To do
+      - name: operator_id
+        description: To do
+      - name: reason
+        description: To do
+      - name: tap_id
+        description: To do
+      - name: ticket_type
+        description: To do
+      - name: created_dttm
+        description: To do
+      - name: kafka_send_status
+        description: To do
+      - name: line
+        description: To do
+      - name: start_station
+        description: To do
+      - name: end_station
+        description: To do
+      - name: start_dttm
+        description: To do
+      - name: end_dttm
+        description: To do
+      - name: ticket_code
+        description: To do
+      - name: additional_infos
+        description: To do
+
+  - name: stg_enghouse__transactions
+    description: Add description
+    columns:
+      - name: id
+        description: To do
+      - name: operation
+        description: To do
+      - name: terminal_id
+        description: To do
+      - name: mapping_terminal_id
+        description: To do
+      - name: mapping_merchant_id
+        description: To do
+      - name: timestamp
+        description: To do
+      - name: amount
+        description: To do
+      - name: payment_reference
+        description: To do
+      - name: spdh_response
+        description: To do
+      - name: response_type
+        description: To do
+      - name: response_message
+        description: To do
+      - name: token
+        description: To do
+      - name: issuer_response
+        description: To do
+      - name: core_response
+        description: To do
+      - name: rrn
+        description: To do
+      - name: authorization_code
+        description: To do
+      - name: par
+        description: To do
+      - name: brand
+        description: To do
+
+  - name: stg_enghouse__pay_windows
+    description: Add description
+    columns:
+      - name: id
+        description: To do
+      - name: token
+        description: To do
+      - name: amount_settled
+        description: To do
+      - name: amount_to_settle
+        description: To do
+      - name: debt_settled
+        description: To do
+      - name: stage
+        description: To do
+      - name: vs
+        description: To do
+      - name: terminal_id
+        description: To do
+      - name: open_date
+        description: To do
+      - name: close_date
+        description: To do

warehouse/models/staging/payments/enghouse/stg_enghouse__pay_windows.sql

@@ -0,0 +1,55 @@
+WITH source AS (
+    SELECT * FROM {{ source('external_enghouse', 'pay_windows') }}
+),
+
+clean_columns AS (
+    SELECT
+        {{ trim_make_empty_string_null('id') }} AS id,
+        {{ trim_make_empty_string_null('token') }} AS token,
+        {{ safe_cast('amount_settled', type_numeric()) }} AS amount_settled,
+        {{ safe_cast('amount_to_settle', type_numeric()) }} AS amount_to_settle,
+        {{ safe_cast('debt_settled', type_numeric()) }} AS debt_settled,
+        {{ trim_make_empty_string_null('stage') }} AS stage,
+        {{ trim_make_empty_string_null('vs') }} AS vs,
+        {{ trim_make_empty_string_null('terminal_id') }} AS terminal_id,
+        {{ safe_cast('open_date', type_timestamp()) }} AS open_date,
+        {{ safe_cast('close_date', type_timestamp()) }} AS close_date,
+        CAST(_line_number AS INTEGER) AS _line_number,
+        `instance`,
+        extract_filename,
+        -- revisit the use of this littlepay-specific macro
+        {{ extract_littlepay_filename_ts() }} AS enghouse_export_ts,
+        {{ extract_littlepay_filename_date() }} AS enghouse_export_date,
+        ts,
+        -- hash all content not generated by us to enable deduping full dup rows
+        -- hashing at this step will preserve distinction between nulls and empty strings in case that is meaningful upstream
+        {{ dbt_utils.generate_surrogate_key(['id', 'token', 'amount_settled', 'amount_to_settle',
+            'debt_settled', 'stage', 'vs', 'terminal_id', 'open_date', 'close_date']) }} AS _content_hash,
+    FROM source
+),
+
+stg_enghouse__pay_windows AS (
+    SELECT
+        id,
+        token,
+        amount_settled,
+        amount_to_settle,
+        debt_settled,
+        stage,
+        vs,
+        terminal_id,
+        open_date,
+        close_date,
+         _line_number,
+        `instance`,
+        extract_filename,
+        enghouse_export_ts,
+        enghouse_export_date,
+        ts,
+        -- _key,
+        -- _payments_key,
+        _content_hash,
+    FROM clean_columns
+)
+
+SELECT * FROM stg_enghouse__pay_windows

warehouse/models/staging/payments/enghouse/stg_enghouse__taps.sql

@@ -0,0 +1,71 @@
+WITH source AS (
+    SELECT * FROM {{ source('external_enghouse', 'taps') }}
+),
+
+clean_columns AS (
+    SELECT
+        {{ trim_make_empty_string_null('tap_id') }} AS tap_id,
+        {{ trim_make_empty_string_null('mapping_terminal_id') }} AS mapping_terminal_id,
+        {{ trim_make_empty_string_null('mapping_merchant_id') }} AS mapping_merchant_id,
+        {{ trim_make_empty_string_null('terminal') }} AS terminal,
+        {{ trim_make_empty_string_null('token') }} AS token,
+        {{ trim_make_empty_string_null('masked_pan') }} AS masked_pan,
+        {{ safe_cast('expiry', type_int()) }} AS expiry,
+        {{ safe_cast('server_date', type_timestamp()) }} AS server_date,
+        {{ safe_cast('terminal_date', type_timestamp()) }} AS terminal_date,
+        {{ safe_cast('tx_number', type_int()) }} AS tx_number,
+        {{ trim_make_empty_string_null('tx_status') }} AS tx_status,
+        {{ trim_make_empty_string_null('payment_reference') }} AS payment_reference,
+        {{ trim_make_empty_string_null('terminal_spdh_code') }} AS terminal_spdh_code,
+        {{ trim_make_empty_string_null('denylist_version') }} AS denylist_version,
+        {{ trim_make_empty_string_null('transit_data') }} AS transit_data,
+        {{ trim_make_empty_string_null('currency') }} AS currency,
+        {{ trim_make_empty_string_null('par') }} AS par,
+        CAST(_line_number AS INTEGER) AS _line_number,
+        `instance`,
+        extract_filename,
+        -- revisit the use of this littlepay-specific macro
+        {{ extract_littlepay_filename_ts() }} AS enghouse_export_ts,
+        {{ extract_littlepay_filename_date() }} AS enghouse_export_date,
+        ts,
+        -- hash all content not generated by us to enable deduping full dup rows
+        -- hashing at this step will preserve distinction between nulls and empty strings in case that is meaningful upstream
+        {{ dbt_utils.generate_surrogate_key(['tap_id', 'mapping_terminal_id', 'mapping_merchant_id', 'terminal',
+            'token', 'masked_pan', 'expiry', 'server_date', 'terminal_date', 'tx_number', 'tx_status',
+            'payment_reference', 'terminal_spdh_code', 'denylist_version', 'transit_data',
+            'currency', 'par']) }} AS _content_hash,
+    FROM source
+),
+
+stg_enghouse__taps AS (
+    SELECT
+        tap_id,
+        mapping_terminal_id,
+        mapping_merchant_id,
+        terminal,
+        token,
+        masked_pan,
+        expiry,
+        server_date,
+        terminal_date,
+        tx_number,
+        tx_status,
+        payment_reference,
+        terminal_spdh_code,
+        denylist_version,
+        transit_data,
+        currency,
+        par,
+         _line_number,
+        `instance`,
+        extract_filename,
+        enghouse_export_ts,
+        enghouse_export_date,
+        ts,
+        -- _key,
+        -- _payments_key,
+        _content_hash,
+    FROM clean_columns
+)
+
+SELECT * FROM stg_enghouse__taps

warehouse/models/staging/payments/enghouse/stg_enghouse__tickets.sql

@@ -0,0 +1,72 @@
+WITH source AS (
+    SELECT * FROM {{ source('external_enghouse', 'tickets') }}
+),
+
+clean_columns AS (
+    SELECT
+        {{ trim_make_empty_string_null('id') }} AS id,
+        {{ trim_make_empty_string_null('ticket_id') }} AS ticket_id,
+        {{ trim_make_empty_string_null('station_name') }} AS station_name,
+        {{ safe_cast('amount', type_numeric()) }} AS amount,
+        {{ trim_make_empty_string_null('clearing_id') }} AS clearing_id,
+        {{ trim_make_empty_string_null('operator_id') }} AS operator_id,
+        {{ trim_make_empty_string_null('reason') }} AS reason,
+        {{ trim_make_empty_string_null('tap_id') }} AS tap_id,
+        {{ trim_make_empty_string_null('ticket_type') }} AS ticket_type,
+        {{ safe_cast('created_dttm', type_timestamp()) }} AS created_dttm,
+        {{ trim_make_empty_string_null('kafka_send_status') }} AS kafka_send_status,
+        {{ trim_make_empty_string_null('line') }} AS line,
+        {{ trim_make_empty_string_null('start_station') }} AS start_station,
+        {{ trim_make_empty_string_null('end_station') }} AS end_station,
+        {{ safe_cast('start_dttm', type_timestamp()) }} AS start_dttm,
+        {{ safe_cast('end_dttm', type_timestamp()) }} AS end_dttm,
+        {{ trim_make_empty_string_null('ticket_code') }} AS ticket_code,
+        {{ trim_make_empty_string_null('additional_infos') }} AS additional_infos,
+        CAST(_line_number AS INTEGER) AS _line_number,
+        `instance`,
+        extract_filename,
+        -- revisit the use of this littlepay-specific macro
+        {{ extract_littlepay_filename_ts() }} AS enghouse_export_ts,
+        {{ extract_littlepay_filename_date() }} AS enghouse_export_date,
+        ts,
+        -- hash all content not generated by us to enable deduping full dup rows
+        -- hashing at this step will preserve distinction between nulls and empty strings in case that is meaningful upstream
+        {{ dbt_utils.generate_surrogate_key(['id', 'ticket_id', 'station_name', 'amount', 'clearing_id', 'operator_id',
+            'reason', 'tap_id', 'ticket_type', 'created_dttm', 'kafka_send_status', 'line', 'start_station',
+            'end_station', 'start_dttm', 'end_dttm', 'ticket_code', 'additional_infos']) }} AS _content_hash,
+    FROM source
+),
+
+stg_enghouse__tickets AS (
+    SELECT
+        id,
+        ticket_id,
+        station_name,
+        amount,
+        clearing_id,
+        operator_id,
+        reason,
+        tap_id,
+        ticket_type,
+        created_dttm,
+        kafka_send_status,
+        line,
+        start_station,
+        end_station,
+        start_dttm,
+        end_dttm,
+        ticket_code,
+        additional_infos,
+         _line_number,
+        `instance`,
+        extract_filename,
+        enghouse_export_ts,
+        enghouse_export_date,
+        ts,
+        -- _key,
+        -- _payments_key,
+        _content_hash,
+    FROM clean_columns
+)
+
+SELECT * FROM stg_enghouse__tickets