Week 4 Challenge (#60)

* Week 4 Challenge

week_4/project/dbt_config.py

@@ -0,0 +1 @@
+DBT_PROJECT_PATH = "/opt/dagster/dagster_home/project/dbt_test_project"

week_4/project/dbt_test_project/dbt_project.yml

@@ -0,0 +1,22 @@
+name: 'test_dbt'
+version: '1.0.0'
+config-version: 2
+
+profile: 'test_dbt'
+
+source-paths: ["models"]
+analysis-paths: ["analysis"]
+test-paths: ["tests"]
+data-paths: ["data"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+target-path: "target"
+clean-targets:
+  - "target"
+  - "dbt_modules"
+
+models:
+  test_dbt:
+    example:
+      +materialized: view

week_4/project/dbt_test_project/models/my_first_dbt_model.sql

@@ -0,0 +1,5 @@
+{{ config(materialized='table') }}
+
+
+SELECT *
+FROM {{ source('postgresql', 'dbt_table') }}

week_4/project/dbt_test_project/models/my_second_dbt_model.sql

@@ -0,0 +1,5 @@
+{{ config(materialized='table') }}
+
+
+SELECT column_2 AS my_column
+FROM {{ ref('my_first_dbt_model') }}

week_4/project/dbt_test_project/models/schema.yml

@@ -0,0 +1,22 @@
+version: 2
+
+models:
+  - name: my_first_dbt_model
+    description: "A starter dbt model"
+    columns:
+      - name: column_1
+        tests:
+          - not_null
+      - name: column_2
+        tests:
+          - not_null
+      - name: column_3
+        tests:
+          - not_null
+
+  - name: my_second_dbt_model
+    description: "A starter dbt model"
+    columns:
+      - name: my_column
+        tests:
+          - not_null

week_4/project/dbt_test_project/models/sources.yml

@@ -0,0 +1,7 @@
+version: 2
+
+sources:
+  - name: postgresql
+    schema: analytics
+    tables:
+      - name: dbt_table

week_4/project/dbt_test_project/profiles.yml

@@ -0,0 +1,12 @@
+test_dbt:
+  target: test
+  outputs:
+    test:
+      type: postgres
+      host: postgresql
+      user: postgres_user
+      password: postgres_password
+      port: 5432
+      dbname: postgres_db
+      schema: analytics
+      threads: 4

week_4/project/repo.py

@@ -1,11 +1,20 @@
-from dagster import repository
+from dagster import repository, with_resources
+from dagster_dbt import dbt_cli_resource
+from project.dbt_config import DBT_PROJECT_PATH
+from project.resources import postgres_resource
 from project.week_4 import (
     get_s3_data_docker,
     process_data_docker,
     put_redis_data_docker,
 )
+from project.week_4_challenge import create_dbt_table, insert_dbt_data
 
 
 @repository
 def repo():
     return [get_s3_data_docker, process_data_docker, put_redis_data_docker]
+
+
+@repository
+def assets_dbt():
+    pass

week_4/project/resources.py

@@ -4,9 +4,26 @@
 
 import boto3
 import redis
+import sqlalchemy
 from dagster import Field, Int, String, resource
 
 
+class Postgres:
+    def __init__(self, host: str, user: str, password: str, database: str):
+        self.host = host
+        self.user = user
+        self.password = password
+        self.database = database
+        self._engine = sqlalchemy.create_engine(self.uri)
+
+    @property
+    def uri(self):
+        return f"postgresql://{self.user}:{self.password}@{self.host}/{self.database}"
+
+    def execute_query(self, query: str):
+        self._engine.execute(query)
+
+
 class S3:
     def __init__(self, bucket: str, access_key: str, secret_key: str, endpoint_url: str = None):
         self.bucket = bucket
@@ -39,6 +56,25 @@ def put_data(self, name: str, value: str):
         self.client.set(name, value)
 
 
+@resource(
+    config_schema={
+        "host": Field(String),
+        "user": Field(String),
+        "password": Field(String),
+        "database": Field(String),
+    },
+    description="A resource that can run Postgres",
+)
+def postgres_resource(context) -> Postgres:
+    """This resource defines a Postgres client"""
+    return Postgres(
+        host=context.resource_config["host"],
+        user=context.resource_config["user"],
+        password=context.resource_config["password"],
+        database=context.resource_config["database"],
+    )
+
+
 @resource
 def mock_s3_resource(context):
     stocks = [

week_4/project/week_4_challenge.py

@@ -0,0 +1,36 @@
+from random import randint
+
+from dagster import AssetIn, asset
+from dagster_dbt import load_assets_from_dbt_project
+from project.dbt_config import DBT_PROJECT_PATH
+
+
+@asset(
+    required_resource_keys={"database"},
+    op_tags={"kind": "postgres"},
+)
+def create_dbt_table(context):
+    sql = "CREATE SCHEMA IF NOT EXISTS analytics;"
+    context.resources.database.execute_query(sql)
+    sql = "CREATE TABLE IF NOT EXISTS analytics.dbt_table (column_1 VARCHAR(100), column_2 VARCHAR(100), column_3 VARCHAR(100));"
+    context.resources.database.execute_query(sql)
+
+
+@asset(
+    required_resource_keys={"database"},
+    op_tags={"kind": "postgres"},
+)
+def insert_dbt_data(context, create_dbt_table):
+    sql = "INSERT INTO analytics.dbt_table (column_1, column_2, column_3) VALUES ('A', 'B', 'C');"
+
+    number_of_rows = randint(1, 10)
+    for _ in range(number_of_rows):
+        context.resources.database.execute_query(sql)
+        context.log.info("Inserted a row")
+
+    context.log.info("Batch inserted")
+
+
+@asset
+def final(context):
+    context.log.info("Week 4 Challenge completed")