testing Snowflake identity resolution automation

CREATE DATABASE DEMO;

USE DATABASE DEMO;

CREATE or REPLACE TABLE customers(
 "id" VARCHAR,
 "first_name" VARCHAR,
 "last_name" VARCHAR,
 "email" VARCHAR
);

ALTER TABLE customers SET CHANGE_TRACKING = TRUE;

INSERT INTO customers VALUES
    ('1', 'John', 'Smith', 'john.smith@example.com'),
    ('2', 'Jessica', 'Davis', 'jessica.davis@example.com'),
    ('3', 'John Robert', 'Smith', 'js@example.com'),
    ('4', 'Jane', 'Doe', 'jane@example.com'),
    ('5', 'J. Robert', 'Smith', 'js+1@example.com'),
    ('6', 'Jessika', 'Davis', 'jessd@example.com');

SELECT * from customers;

USE DATABASE DEMO;

ALTER TABLE customers SET CHANGE_TRACKING = TRUE;

CREATE or REPLACE TABLE customers_entities(
    "entity_id" VARCHAR,
    "customer_id" VARCHAR
);

CREATE OR REPLACE TABLE TILORES_CONFIG (
    name varchar(255) NOT NULL,
    value varchar(255) default NULL
);

INSERT INTO TILORES_CONFIG VALUES
    ('last_sync', CURRENT_TIMESTAMP()::string),
    ('fetch_on_next_run', 'true');

SELECT * FROM TILORES_CONFIG;
      
CREATE OR REPLACE TASK tilores_sync
    SCHEDULE = '1 MINUTE'
    AS
    DECLARE
        now STRING;
        last_sync STRING;
        fetch_on_next_run STRING;
    BEGIN 
        now := (SELECT CURRENT_TIMESTAMP());
        last_sync := (SELECT value FROM TILORES_CONFIG WHERE name='last_sync');
        fetch_on_next_run := (SELECT value FROM TILORES_CONFIG WHERE name='fetch_on_next_run');
        
        SELECT tilores_ingest(OBJECT_CONSTRUCT(*)) AS ingested 
        FROM (
            SELECT * EXCLUDE(METADATA$ACTION, METADATA$ISUPDATE, METADATA$ROW_ID)
            FROM customers
            CHANGES(INFORMATION => DEFAULT)
                AT(TIMESTAMP => to_timestamp_tz(:last_sync))
            WHERE METADATA$ACTION='INSERT');

        IF (fetch_on_next_run = 'true') THEN
            MERGE INTO customers_entities TARGET USING (SELECT tilores_entity_by_record_id("id"):id AS "entity_id", "id" FROM customers) SOURCE
            ON TARGET."customer_id" = SOURCE."id"
            WHEN MATCHED THEN
              UPDATE SET
                  TARGET."entity_id" = SOURCE."entity_id"
            WHEN NOT MATCHED THEN
              INSERT ("entity_id", "customer_id")
                  VALUES (SOURCE."entity_id", SOURCE."id");
        END IF;
            
        UPDATE TILORES_CONFIG
            SET value = :now
            WHERE name = 'last_sync';
        UPDATE TILORES_CONFIG
            SET value = (SELECT 
                            CASE
                                WHEN count(*) > 0 THEN 'true'
                                ELSE 'false'
                            END AS fetch_on_next_run
                        FROM customers
                        CHANGES(INFORMATION => DEFAULT)
                            AT(TIMESTAMP => to_timestamp_tz(:last_sync))
                        WHERE METADATA$ACTION='INSERT')
            WHERE name = 'fetch_on_next_run';
    END;

ALTER TASK tilores_sync RESUME;

SELECT *
FROM TABLE(information_schema.task_history())
ORDER BY scheduled_time;

SELECT * EXCLUDE("customer_id")
FROM customers
LEFT JOIN customers_entities ON "id"="customer_id"
ORDER BY "id";

INSERT INTO customers VALUES
  ('7', 'Johnn', 'Smeth', 'john.smith@otherDomain.com');

SELECT * EXCLUDE("customer_id")
FROM customers
LEFT JOIN customers_entities ON "id"="customer_id"
ORDER BY "id";

## Using snowflake connector 
import snowflake.connector
from configparser import ConfigParser
import pandas as pd

config = ConfigParser()
config.read("config.ini")

user=config["SNOWFLAKE"]["USER"]
password=config["SNOWFLAKE"]["PASSWORD"]
database=config["SNOWFLAKE"]["DATABASE"]
schema=config["SNOWFLAKE"]["SCHEMA"]
warehouse=config["SNOWFLAKE"]["WAREHOUSE"]
role=config["SNOWFLAKE"]["ROLE"]



## snowflake connector method
connection = snowflake.connector.connect(
    user=user,
    password=password,
    account='hostelworld.eu-west-1',
    database = database,
    warehouse = warehouse,
    role = role)
print("snowflake connector connected...")

query = '''select * from production.ppc.campaign_performance_daily limit 1000 ;'''
df = pd.read_sql_query(sql=query, con=connection)
print('DF shape = ', df.shape)

connection.close()
print("connection closed")



### Using sql alchemy
from sqlalchemy import create_engine
from configparser import ConfigParser
import pandas as pd

config = ConfigParser()
config.read("config.ini")

user=config["SNOWFLAKE"]["USER"]
password=config["SNOWFLAKE"]["PASSWORD"]
database=config["SNOWFLAKE"]["DATABASE"]
schema=config["SNOWFLAKE"]["SCHEMA"]
warehouse=config["SNOWFLAKE"]["WAREHOUSE"]
role=config["SNOWFLAKE"]["ROLE"]

# sql alchemy method
engine = create_engine(
            f"""snowflake://{user}:{password}@hostelworld.eu-west-1/{database}/{schema}?warehouse={warehouse}&role={role}"""
)

connection = engine.connect()
print('sql alchemy connected...')
            
query = '''select * from production.ppc.campaign_performance_daily limit 1000 ;'''
df = pd.read_sql_query(sql=query, con=connection)
print('DF shape = ', df.shape)

connection.close()
print("connection closed")

create or replace transient table analytics.dbt_klee.mock_orders(
    order_id integer,
    status varchar(100),
    created_at date,
    updated_at date
)

insert into analytics.dbt_klee.mock_orders (order_id, status, created_at, updated_at) 
values
    (1, 'delivered', '2020-01-01', '2020-01-04'),
    (2, 'shipped', '2020-01-02', '2020-01-04'),
    (3, 'shipped', '2020-01-03', '2020-01-04'),
    (4, 'processed', '2020-01-04', '2020-01-04');