Add bronze-to-silver Snowflake pipeline template

templates/bronze-silver-snowflake/.bruin.yml

@@ -0,0 +1,14 @@
+default_environment: default
+environments:
+  default:
+    connections:
+      frankfurter:
+        - name: "frankfurter-default"
+      snowflake:
+        - name: "snowflake-default"
+          account: "ACCOUNT-NAME"
+          user: "USERNAME"
+          password: "PASSWORD"
+          database: "SNOWFLAKE_DATABASE"
+          warehouse: "COMPUTE_WAREHOUSE"
+          schema: "PUBLIC"

templates/bronze-silver-snowflake/README.md

@@ -0,0 +1,70 @@
+# Bronze to Silver Snowflake Pipeline
+
+## Overview
+
+This template demonstrates a **bronze-to-silver data pipeline** using Snowflake as the destination.
+
+It ingests exchange rate data from the Frankfurter API into a **bronze layer** (raw data) and performs transformations to generate aggregated insights in a **silver layer**.
+
+The pipeline showcases a complete **ingestion → transformation → validation workflow** using Bruin.
+
+---
+
+## Architecture
+
+Frankfurter API  
+↓  
+Bronze Layer (Raw Ingestion - Snowflake)  
+↓  
+Silver Layer (Aggregations & Metrics - Snowflake)
+
+---
+
+## Features
+
+- Ingestion from a public API (no credentials required)
+- Bronze layer for raw data storage
+- Silver layer with:
+  - Latest exchange rates
+  - 7-day and 30-day averages
+  - Observation counts
+- Data quality checks:
+  - Not null checks
+  - Positive value checks
+  - Freshness validation
+  - Uniqueness constraints
+- Snowflake-native SQL transformations
+
+---
+
+## Project Structure
+bronze-silver-snowflake/
+│
+├── pipeline.yml
+├── .bruin.yml
+├── README.md
+└── assets/
+   ├── bronze_raw_data.asset.yml
+   └── silver_aggregated.sql
+
+
+---
+
+## Snowflake Setup
+
+Before running the pipeline, configure your Snowflake connection in `.bruin.yml`:
+
+```bruin.yaml
+snowflake:
+  - name: "snowflake-default"
+    account: "YOUR_ACCOUNT"
+    user: "YOUR_USERNAME"
+    password: "YOUR_PASSWORD"
+    database: "YOUR_DATABASE"
+    warehouse: "COMPUTE_WAREHOUSE"
+    schema: "PUBLIC"
+```
+Replace placeholders with your Snowflake credentials for local testing.
+⚠️ Do not commit real credentials to GitHub.
+
+---

templates/bronze-silver-snowflake/assets/bronze_raw_data.asset.yml

@@ -0,0 +1,52 @@
+name: public.bronze_exchange_rates
+type: ingestr
+
+description: >
+  Ingests daily FX rates from the Frankfurter API into Snowflake without transformations.
+  The dataset makes the raw bronze layer available for downstream silver modeling.
+
+columns:
+  - name: date
+    type: date
+    description: "Exchange rate date reported by the Frankfurter API."
+    checks:
+      - name: not_null
+  - name: base_currency
+    type: varchar
+    description: "Currency that the quote currency is priced against."
+    checks:
+      - name: not_null
+  - name: currency_code
+    type: varchar
+    description: "Quote currency code returned by Frankfurter."
+    checks:
+      - name: not_null
+  - name: rate
+    type: float
+    description: "Reported daily conversion rate between the base and quote currency."
+    checks:
+      - name: not_null
+      - name: positive
+
+custom_checks:
+  - name: unique currency per day
+    value: 0
+    query: |
+      SELECT COUNT(*)
+      FROM (
+        SELECT date, currency_code, base_currency
+        FROM public.bronze_exchange_rates
+        GROUP BY 1, 2, 3
+        HAVING COUNT(*) > 1
+      ) duplicates
+  - name: data freshness (3 day max lag)
+    value: 0
+    query: |
+      SELECT CASE WHEN MAX(TO_DATE(date)) < DATEADD(day, -3, CURRENT_DATE()) THEN 1 ELSE 0 END AS stale_flag
+      FROM public.bronze_exchange_rates
+
+parameters:
+  source_connection: frankfurter-default
+  source_table: exchange_rates
+  destination: snowflake
+  destination_table: bronze_exchange_rates

templates/bronze-silver-snowflake/assets/silver_aggregated.sql

@@ -0,0 +1,122 @@
+/* @bruin
+
+name: public.silver_exchange_rate_summary
+type: sf.sql
+
+materialization:
+  type: table
+
+description: >
+  Aggregates the bronze Frankfurter exchange rate feed into a silver summary table with
+  fresh rates, rolling averages, and observation counts for downstream analytics.
+
+depends:
+  - public.bronze_exchange_rates
+
+columns:
+  - name: currency_code
+    type: varchar
+    description: "Quoted currency code."
+    checks:
+      - name: not_null
+  - name: base_currency
+    type: varchar
+    description: "Base currency that the quote is relative to."
+    checks:
+      - name: not_null
+  - name: latest_date
+    type: date
+    description: "Most recent exchange rate date captured in the bronze layer."
+    checks:
+      - name: not_null
+  - name: latest_rate
+    type: float
+    description: "Latest available exchange rate between the currency pair."
+    checks:
+      - name: not_null
+      - name: positive
+  - name: avg_rate_7d
+    type: float
+    description: "Seven day rolling average of the conversion rate."
+    checks:
+      - name: positive
+  - name: avg_rate_30d
+    type: float
+    description: "Thirty day rolling average of the conversion rate."
+    checks:
+      - name: positive
+  - name: observations_7d
+    type: integer
+    description: "Number of daily observations in the last seven days."
+    checks:
+      - name: non_negative
+  - name: observations_30d
+    type: integer
+    description: "Number of daily observations in the last thirty days."
+    checks:
+      - name: positive
+
+custom_checks:
+  - name: silver table populated
+    value: 0
+    query: |
+      SELECT CASE WHEN COUNT(*) = 0 THEN 1 ELSE 0 END
+      FROM public.silver_exchange_rate_summary
+  - name: fresh latest date (<= 3 day lag)
+    value: 0
+    query: |
+      SELECT COUNT(*)
+      FROM public.silver_exchange_rate_summary
+      WHERE latest_date < DATEADD(day, -3, CURRENT_DATE())
+
+@bruin */
+
+WITH bronze_data AS (
+  SELECT
+    TO_DATE(date) AS rate_date,
+    base_currency,
+    currency_code,
+    CAST(rate AS FLOAT) AS rate
+  FROM public.bronze_exchange_rates
+),
+recent AS (
+  SELECT
+    currency_code,
+    base_currency,
+    AVG(CASE WHEN rate_date >= DATEADD(day, -7, CURRENT_DATE()) THEN rate END) AS avg_rate_7d,
+    AVG(CASE WHEN rate_date >= DATEADD(day, -30, CURRENT_DATE()) THEN rate END) AS avg_rate_30d,
+    COUNT(CASE WHEN rate_date >= DATEADD(day, -7, CURRENT_DATE()) THEN 1 END) AS observations_7d,
+    COUNT(CASE WHEN rate_date >= DATEADD(day, -30, CURRENT_DATE()) THEN 1 END) AS observations_30d
+  FROM bronze_data
+  WHERE rate_date >= DATEADD(day, -30, CURRENT_DATE())
+  GROUP BY currency_code, base_currency
+),
+
+latest AS (
+  SELECT
+    currency_code,
+    base_currency,
+    rate_date AS latest_date,
+    rate AS latest_rate
+  FROM bronze_data
+  QUALIFY ROW_NUMBER() OVER (
+    PARTITION BY currency_code, base_currency
+    ORDER BY rate_date DESC
+  ) = 1
+)
+
+SELECT
+  r.currency_code,
+  r.base_currency,
+  l.latest_date,
+  l.latest_rate,
+  r.avg_rate_7d,
+  r.avg_rate_30d,
+  r.observations_7d,
+  r.observations_30d
+FROM recent r
+JOIN latest l
+  ON r.currency_code = l.currency_code
+ AND r.base_currency = l.base_currency
+ORDER BY r.currency_code, r.base_currency;
+

templates/bronze-silver-snowflake/pipeline.yml

@@ -0,0 +1,8 @@
+name: bronze-silver-snowflake
+schedule: daily
+start_date: "2024-01-01"
+catchup: false
+
+default_connections:
+  frankfurter: "frankfurter-default"
+  snowflake: "snowflake-default"