Feature/cdm line item

CHANGELOG.md

@@ -1,3 +1,8 @@
+# dbt_recharge v0.3.0
+[PR #17](https://github.com/fivetran/dbt_recharge/pull/17) includes the following updates:
+## Feature Updates
+- Introduced the new `recharge__line_item_enhanced` model. This model includes a line item enriched with invoice, subscription, payment, and refund information. This model has been built with the intention of retaining a common line item schema across all other Fivetran billing data models.
+
 # dbt_recharge v0.2.0
 [PR #16](https://github.com/fivetran/dbt_recharge/pull/16) includes the following updates:
 ## Features

README.md

@@ -44,7 +44,7 @@ Include the following recharge package version in your `packages.yml` file.
 ```yaml
 packages:
   - package: fivetran/recharge
-    version: [">=0.2.0", "<0.3.0"] # we recommend using ranges to capture non-breaking changes automatically
+    version: [">=0.3.0", "<0.4.0"] # we recommend using ranges to capture non-breaking changes automatically
 ```
 Do **NOT** include the `recharge_source` package in this file. The transformation package itself has a dependency on it and will install the source package as well.
 
@@ -167,7 +167,7 @@ This dbt package is dependent on the following dbt packages. Please be aware tha
 ```yml
 packages:
     - package: fivetran/recharge_source
-      version: [">=0.2.0", "<0.3.0"]
+      version: [">=0.3.0", "<0.4.0"]
 
     - package: fivetran/fivetran_utils
       version: [">=0.4.0", "<0.5.0"]

dbt_project.yml

@@ -1,6 +1,6 @@
 
 name: 'recharge'
-version: '0.2.0'
+version: '0.3.0'
 config-version: 2
 require-dbt-version: [">=1.3.0", "<2.0.0"]
 
@@ -15,6 +15,7 @@ vars:
     charge_order_attribute: "{{ ref('stg_recharge__charge_order_attribute') }}"
     charge_tax_line: "{{ ref('stg_recharge__charge_tax_line') }}"
     charge_shipping_line: "{{ ref('stg_recharge__charge_shipping_line') }}"
+    checkout: "{{ ref('stg_recharge__checkout') }}"
     customer: "{{ ref('stg_recharge__customer') }}"
     discount: "{{ ref('stg_recharge__discount') }}"
     one_time_product: "{{ ref('stg_recharge__one_time_product') }}"
@@ -26,7 +27,8 @@ vars:
 
     recharge__address_passthrough_columns: [] 
     recharge__charge_passthrough_columns: [] 
-    recharge__charge_line_item_passthrough_columns: [] 
+    recharge__charge_line_item_passthrough_columns: []
+    recharge__checkout_passthrough_columns: [] 
     recharge__order_passthrough_columns: [] 
     recharge__order_line_passthrough_columns: [] 
     recharge__subscription_passthrough_columns: [] 

integration_tests/dbt_project.yml

@@ -1,7 +1,7 @@
 config-version: 2
 
 name: 'recharge_integration_tests'
-version: '0.2.0'
+version: '0.3.0'
 
 profile: 'integration_tests' 
 
@@ -17,6 +17,7 @@ vars:
     recharge_charge_order_attribute_identifier: "charge_order_attribute_data"
     recharge_charge_shipping_line_identifier: "charge_shipping_line_data"
     recharge_charge_tax_line_identifier: "charge_tax_line_data"
+    recharge_checkout_identifier: "checkout_data"
     recharge_customer_identifier: "customer_data"
     recharge_discount_identifier: "discount_data"
     recharge_one_time_product_identifier: "one_time_product_data"
@@ -52,6 +53,7 @@ seeds:
       EXTERNAL_PRODUCT_ID_ECOMMERCE: "{{ 'INT64' if target.type == 'bigquery' else 'bigint' }}"
       EXTERNAL_VARIANT_ID_ECOMMERCE: "{{ 'INT64' if target.type == 'bigquery' else 'bigint' }}"
       SUBSCRIPTION_ID: "{{ 'INT64' if target.type == 'bigquery' else 'bigint' }}"
+      COMPANY: "{{ 'string' if target.name in ['bigquery', 'spark', 'databricks'] else 'varchar' }}"
     charge_data:
       +column_types:
         LAST_CHARGE_ATTEMPT_DATE: "timestamp"

integration_tests/seeds/checkout_data.csv

@@ -0,0 +1,2 @@
+token, charge_id
+abc, 123

models/common data models/recharge__line_item_enhanced.sql

@@ -0,0 +1,202 @@
+with charge_line_items as (
+
+    select * 
+    from {{ var('charge_line_item')}}
+
+), charges as (
+
+    select * 
+    from {{ var('charge') }}
+
+), charge_shipping_lines as (
+
+    select
+        charge_id,
+        sum(price) as total_shipping
+    from {{ var('charge_shipping_line') }}
+    group by 1
+
+{% if var('recharge__using_checkout', false) %}
+), checkouts as (
+
+    select *
+    from {{ var('checkout') }}
+
+{% endif %}
+
+), addresses as (
+
+    select * 
+    from {{ var('address') }}
+
+), customers as (
+
+    select * 
+    from {{ var('customer') }}
+
+), subscriptions as (
+
+    select *
+    from {{ var('subscription_history') }} 
+    where is_most_recent_record
+
+), enhanced as (
+    select
+        charge_line_items.charge_id as header_id,
+        charge_line_items.index as line_item_id,
+        row_number() over (partition by charge_line_items.charge_id
+            order by charge_line_items.index) as line_item_index,
+
+        -- header level items
+        charges.charge_created_at as created_at,
+        charges.charge_status as header_status,
+        charges.total_discounts as discount_amount,
+        charges.total_refunds as refund_amount,
+        charge_shipping_lines.total_shipping as fee_amount,
+        addresses.payment_method_id,
+        charges.external_transaction_id_payment_processor as payment_id,
+        charges.payment_processor as payment_method,
+        charges.charge_processed_at as payment_at,
+        charges.charge_type as billing_type,  -- possible values: checkout, recurring
+
+        {% if var('recharge__using_checkout', false) %}
+        checkouts.currency,
+        {% else %}
+        cast(null as {{ dbt.type_string() }}) as currency, 
+        -- currency is in the charges api but not the fivetran schema, so relying on checkouts for now. this only has 20% utilization though so we will want to switch if they add it.
+        {% endif %}
+
+        charge_line_items.purchase_item_type as transaction_type, -- possible values: subscription, onetime
+        charge_line_items.external_product_id_ecommerce as product_id,
+        charge_line_items.title as product_name,
+        -- product_type unknown for now
+        cast(null as {{ dbt.type_string() }}) as product_type,
+        charge_line_items.quantity,
+        charge_line_items.unit_price as unit_amount,
+        charge_line_items.tax_due as tax_amount,
+        charge_line_items.total_price as total_amount,
+        case when charge_line_items.purchase_item_type = 'subscription'
+            then charge_line_items.purchase_item_id
+            else null
+            end as subscription_id,
+        subscriptions.subscription_created_at as subscription_period_started_at,
+        subscriptions.subscription_cancelled_at as subscription_period_ended_at,
+        subscriptions.subscription_status,
+        'customer' as customer_level,
+        charges.customer_id as customer_id,
+        -- coalesces are since information may be incomplete in various tables
+        coalesce(charges.email, customers.email) as customer_email,
+        coalesce(
+            {{ dbt.concat(["customers.billing_first_name", "' '", "customers.billing_last_name"]) }},
+            {{ dbt.concat(["addresses.first_name", "' '", "addresses.last_name"]) }}
+            ) as customer_name,
+        coalesce(customers.billing_company, addresses.company) as customer_company,
+        coalesce(customers.billing_city, addresses.city) as customer_city,
+        coalesce(customers.billing_country, addresses.country) as customer_country
+
+    from charge_line_items
+
+    left join charges
+        on charges.charge_id = charge_line_items.charge_id
+
+    left join addresses
+        on addresses.address_id = charges.address_id
+
+    left join customers
+        on customers.customer_id = charges.customer_id
+
+    {% if var('recharge__using_checkout', false) %}
+    left join checkouts
+        on checkouts.charge_id = charges.charge_id
+    {% endif %}
+
+    left join charge_shipping_lines
+        on charge_shipping_lines.charge_id = charges.charge_id
+
+    left join subscriptions
+        on subscriptions.subscription_id = charge_line_items.purchase_item_id
+
+), final as (
+
+    -- line item level
+    select 
+        header_id,
+        line_item_id,
+        line_item_index,
+        'line_item' as record_type,
+        created_at,
+        header_status,
+        billing_type,
+        currency,
+        product_id,
+        product_name,
+        product_type,
+        transaction_type,
+        quantity,
+        unit_amount,
+        cast(null as {{ dbt.type_float() }}) as discount_amount,
+        cast(null as {{ dbt.type_float() }}) as refund_amount,
+        cast(null as {{ dbt.type_float() }}) as fee_amount,
+        tax_amount,
+        total_amount,
+        payment_id,
+        payment_method_id,
+        payment_method,
+        payment_at,
+        subscription_id,
+        subscription_period_started_at,
+        subscription_period_ended_at,
+        subscription_status,
+        customer_id,
+        customer_level,
+        customer_name,
+        customer_company,
+        customer_email,
+        customer_city,
+        customer_country
+    from enhanced
+
+    union all
+
+    -- header level
+    select
+        header_id,
+        cast(null as {{ dbt.type_int() }}) as line_item_id,
+        cast(0 as {{ dbt.type_int() }}) as line_item_index,
+        'header' as record_type,
+        created_at,
+        header_status,
+        billing_type,
+        currency,
+        cast(null as {{ dbt.type_int() }}) as product_id,
+        cast(null as {{ dbt.type_string() }}) as product_name,
+        cast(null as {{ dbt.type_string() }}) as product_type,
+        cast(null as {{ dbt.type_string() }}) as transaction_type,
+        cast(null as {{ dbt.type_int() }}) as quantity,
+        cast(null as {{ dbt.type_float() }}) as unit_amount,
+        discount_amount,
+        refund_amount,
+        fee_amount,
+        cast(null as {{ dbt.type_float() }}) as tax_amount,
+        cast(null as {{ dbt.type_float() }}) as total_amount,
+        payment_id,
+        payment_method_id,
+        payment_method,
+        payment_at,
+        cast(null as {{ dbt.type_int() }}) as subscription_id,
+        cast(null as {{ dbt.type_timestamp() }}) as subscription_period_started_at,
+        cast(null as {{ dbt.type_timestamp() }}) as subscription_period_ended_at,
+        cast(null as {{ dbt.type_string() }}) as subscription_status,
+        customer_id,
+        customer_level,
+        customer_name,
+        customer_company,
+        customer_email,
+        customer_city,
+        customer_country
+    from enhanced
+    where line_item_index = 1
+)
+
+select *
+from final

packages.yml

@@ -1,3 +1,6 @@
 packages:
- - package: fivetran/recharge_source
-   version: [">=0.2.0", "<0.3.0"]
+#  - package: fivetran/recharge_source
+#    version: [">=0.3.0", "<0.4.0"]
+- git: https://github.com/fivetran/dbt_recharge_source.git
+  revision: feature/cdm-line-item
+  warn-unpinned: false