Patrones de Transformación dbt

version: 2

sources:
  - name: shopify
    description: Raw Shopify data via Fivetran
    database: nutrishop-prod
    schema: raw_shopify
    loader: fivetran
    loaded_at_field: _fivetran_synced
    freshness:
      warn_after: { count: 6, period: hour }
      error_after: { count: 24, period: hour }
    tables:
      - name: orders
        columns:
          - name: id
            tests: [unique, not_null]
          - name: customer_id
            tests: [not_null]
          - name: financial_status
            tests:
              - accepted_values:
                  values: [paid, pending,
                           refunded, voided]

with source as (
    select * from
    {{ source('shopify', 'orders') }}
),

renamed as (
    select
        -- ids
        id as order_id,
        customer_id,
        location_id,

        -- status
        fulfillment_status,
        financial_status as payment_status,

        -- amounts (centavos → euros)
        {{ cents_to_euros('total_price_set') }}
            as total_price,
        {{ cents_to_euros('subtotal_price') }}
            as subtotal,
        {{ cents_to_euros('total_discounts') }}
            as total_discount,

        -- timestamps
        created_at,
        updated_at,
        processed_at,
        cancelled_at,

        -- metadata
        _fivetran_synced as _loaded_at

    from source
)

select * from renamed

{{
    config(
        materialized='table',
        unique_key='customer_key',
        cluster_by=['customer_segment']
    )
}}

with customers as (
    select * from
    {{ ref('int_customer_ltv') }}
),

final as (
    select
        -- surrogate key
        {{ dbt_utils.generate_surrogate_key(
            ['customer_id']) }}
            as customer_key,

        customer_id,
        email,
        customer_since,
        total_orders,
        lifetime_value,
        avg_order_value,
        first_order_at,
        last_order_at,
        customer_segment,

        -- tier económico
        case
            when lifetime_value >= 500
                then 'premium'
            when lifetime_value >= 100
                then 'standard'
            else 'entry'
        end as customer_tier,

        current_timestamp as _loaded_at
    from customers
)

select * from final

{{
    config(
        materialized='incremental',
        unique_key='order_id',
        incremental_strategy='merge',
        partition_by={
          "field": "created_date",
          "data_type": "date",
          "granularity": "day"
        },
        cluster_by=['customer_segment','payment_status']
    )
}}

with orders as (
    select * from
    {{ ref('stg_shopify__orders') }}

    {% if is_incremental() %}
    where updated_at > (
        select max(updated_at) from {{ this }}
    )
    {% endif %}
),

customers as (
    select customer_key, customer_id,
           customer_segment, customer_tier
    from {{ ref('dim_customers') }}
),

final as (
    select
        o.order_id,
        c.customer_key,
        c.customer_segment,
        c.customer_tier,
        o.payment_status,
        o.fulfillment_status,
        o.total_price,
        o.subtotal,
        o.total_discount,
        date(o.created_at) as created_date,
        o.created_at,
        o.updated_at,
        current_timestamp as _loaded_at
    from orders o
    left join customers c
        on o.customer_id = c.customer_id
)

select * from final

{%- macro cents_to_euros(column_name, precision=2) -%}
    round({{ column_name }} / 100.0, {{ precision }})
{%- endmacro -%}

-- Uso en cualquier modelo:
{{ cents_to_euros('amount') }} as amount_eur,
{{ cents_to_euros('total_price_set', precision=0) }}
    as total_price_eur

{%- macro segment_customer_rfm(ltv_col, last_order_col) -%}
    case
        when {{ ltv_col }} >= 500
             and {{ last_order_col }} >= dateadd(
               day, -90, current_date)
            then 'champion'
        when {{ ltv_col }} >= 200
             and {{ last_order_col }} >= dateadd(
               day, -180, current_date)
            then 'loyal'
        when {{ last_order_col }} >= dateadd(
               day, -30, current_date)
            then 'new_customer'
        when {{ last_order_col }} < dateadd(
               day, -365, current_date)
            then 'churned'
        else 'at_risk'
    end
{%- endmacro -%}

{%- macro limit_data_in_dev(column_name, days=7) -%}
    {%- if target.name == 'dev' -%}
        where {{ column_name }} >= dateadd(
            day, -{{ days }}, current_date)
    {%- endif -%}
{%- endmacro -%}

-- Uso: filtra solo 7 días en dev,
-- full scan en producción
select * from {{ ref('stg_shopify__orders') }}
{{ limit_data_in_dev('created_at') }}

{%- macro generate_schema_name(
        custom_schema_name, node) -%}
    {%- set default_schema = target.schema -%}
    {%- if custom_schema_name is none -%}
        {{ default_schema }}
    {%- else -%}
        {{ default_schema }}_{{ custom_schema_name }}
    {%- endif -%}
{%- endmacro -%}

-- Genera schemas como:
-- dev_mfernandez_staging
-- prod_staging, prod_analytics