Move sushi_dbt to test and replace with a self-contained sushi dbt project (#646)

Author: crericha

examples/sushi/config.py

@@ -27,9 +27,9 @@
 )
 
 
-airflow_config = Config(scheduler=AirflowSchedulerConfig())
+airflow_config = Config(**{"scheduler": AirflowSchedulerConfig()})
 
 
 airflow_config_docker = Config(
-    scheduler=AirflowSchedulerConfig(airflow_url="http://airflow-webserver:8080/"),
+    **{"scheduler": AirflowSchedulerConfig(airflow_url="http://airflow-webserver:8080/")},
 )

examples/sushi_dbt/dbt_project.yml

@@ -16,35 +16,13 @@ clean-targets:         # directories to be removed by `dbt clean`
   - "target"
   - "dbt_packages"
 
-# Configuring models
-# Full documentation: https://docs.getdbt.com/docs/configuring-models
-
 models:
   +start: Jan 1 2022
   sushi:
     +materialized: table
-    +pre-hook:
-      - '{{ log("pre-hook") }}'
-    +post-hook: 
-      - '{{ log("post-hook") }}'
-  customers:
-    +materialized: table
 
 seeds:
-  sushi:
-    +pre-hook:
-      - '{{ log("pre-hook") }}'
-    +post-hook: 
-      - '{{ log("post-hook") }}'
+  +schema: raw
 
 vars:
   top_waiters:limit: 10
-  'top_waiters:revenue': "revenue"
-
-  # The following are only used for testing purposes
-  customers:boo: ["a", "b"]
-
-  customers:
-    some_var: ["foo", "bar"]
-    'customers:bla': false
-    'customers:customer_id': "customer_id"

examples/sushi_dbt/models/customer_revenue_by_day.sql

@@ -0,0 +1,47 @@
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='delete+insert',
+        cluster_by=['ds'],
+        time_column='ds',
+    )
+}}
+
+WITH order_total AS (
+  SELECT
+    oi.order_id AS order_id,
+    SUM(oi.quantity * i.price) AS total,
+    oi.ds AS ds
+  FROM {{ ref('order_items') }} AS oi
+  LEFT JOIN {{ ref('items') }} AS i
+    ON oi.item_id = i.id AND oi.ds = i.ds
+{% if is_incremental() %}
+  WHERE
+    oi.ds > (select max(ds) from {{ this }})
+{% endif %}
+{% if sqlmesh is defined %}
+  WHERE
+      oi.ds BETWEEN '{{ start_ds }}' AND '{{ end_ds }}'
+{% endif %}
+  GROUP BY
+    oi.order_id,
+    oi.ds
+)
+SELECT
+  o.customer_id::INT AS customer_id, /* Customer id */
+  SUM(ot.total)::DOUBLE AS revenue, /* Revenue from orders made by this customer */
+  o.ds::TEXT AS ds /* Date */
+FROM {{ ref('orders') }} AS o
+LEFT JOIN order_total AS ot
+  ON o.id = ot.order_id AND o.ds = ot.ds
+{% if is_incremental() %}
+WHERE
+  o.ds > (select max(ds) from {{ this }})
+{% endif %}
+{% if sqlmesh is defined %}
+  WHERE
+      o.ds BETWEEN '{{ start_ds }}' AND '{{ end_ds }}'
+{% endif %}
+GROUP BY
+  o.customer_id,
+  o.ds

examples/sushi_dbt/models/customers.sql

@@ -0,0 +1,3 @@
+SELECT DISTINCT
+  customer_id::INT AS customer_id
+FROM {{ ref('orders') }} as o

examples/sushi_dbt/models/schema.yml

@@ -1,15 +1,9 @@
 version: 2
 
 models:
+  - name: customers
+  - name: customer_revenue_by_day
   - name: top_waiters
   - name: waiters
   - name: waiter_as_customer_by_day
   - name: waiter_revenue_by_day
-
-sources:
-  - name: raw
-    schema: raw
-    tables:
-      - name: items
-      - name: orders
-      - name: order_items

examples/sushi_dbt/models/top_waiters.sql

@@ -6,7 +6,7 @@
 
 SELECT
   waiter_id::INT AS waiter_id,
-  revenue::DOUBLE AS {{ var("top_waiters:revenue") }}
+  revenue::DOUBLE AS revenue
 FROM {{ ref('waiter_revenue_by_day') }}
 WHERE
   ds = (

examples/sushi_dbt/models/waiter_revenue_by_day.sql

@@ -7,17 +7,14 @@
     )
 }}
 
-{{ log_value(5) }}
-{{ log(adapter.dispatch('current_engine', 'customers')()) }}
-
 SELECT
   o.waiter_id::INT AS waiter_id, /* Waiter id */
   SUM(oi.quantity * i.price)::DOUBLE AS revenue, /* Revenue from orders taken by this waiter */
   o.ds::TEXT AS ds /* Date */
-FROM {{ source('raw', 'orders') }} AS o
-LEFT JOIN {{ source('raw', 'order_items') }} AS oi
+FROM {{ ref('orders') }} AS o
+LEFT JOIN {{ ref('order_items') }} AS oi
   ON o.id = oi.order_id AND o.ds = oi.ds
-LEFT JOIN {{ source('raw', 'items') }} AS i
+LEFT JOIN {{ ref('items') }} AS i
   ON oi.item_id = i.id AND oi.ds = i.ds
 {% if is_incremental() %}
   WHERE

examples/sushi_dbt/models/waiters.sql

@@ -7,5 +7,5 @@
 SELECT DISTINCT
   waiter_id::INT AS waiter_id,
   ds::TEXT AS ds
-FROM {{ source('raw', 'orders') }}
+FROM {{ ref('orders') }}
 {{ incremental_by_time('ds', 'ds') }}

examples/sushi_dbt/packages/customers/models/schema.yml

@@ -1,13 +0,0 @@
-version: 2
-
-models:
-  - name: customers
-  - name: customer_revenue_by_day
-
-sources:
-  - name: raw
-    schema: raw
-    tables:
-      - name: items
-      - name: orders
-      - name: order_items