Feature/add vendor data (#285)

.github/workflows/10_feature_dbt_checks.yml

@@ -124,12 +124,12 @@ jobs:
       # There is an issue with --empty and dynamic tables so need to exclude them
       - name: Governance run of dbt with EMPTY models using slim mode
         if: ${{ steps.prod_manifest.outputs.manifest_found == 'true' }}
-        run: "dbt build --fail-fast --defer --state logs --select state:modified+ --empty --exclude config.materialized:dynamic_table ${{ env.FULL_REFRESH_FLAG }}"
+        run: "dbt build --fail-fast --defer --state logs --select state:modified+ --empty --exclude config.materialized:dynamic_table tag:requires_fixture_data ${{ env.FULL_REFRESH_FLAG }}"
 
       # There is an issue with --empty and dynamic tables so need to exclude
       - name: Governance run of dbt with EMPTY models using full run
         if: ${{ steps.prod_manifest.outputs.manifest_found == 'false' }}
-        run: "dbt build --fail-fast --empty --exclude config.materialized:dynamic_table ${{ env.FULL_REFRESH_FLAG }}"
+        run: "dbt build --fail-fast --empty --exclude config.materialized:dynamic_table tag:requires_fixture_data ${{ env.FULL_REFRESH_FLAG }}"
 
       - name: Generate Docs Combining Prod and branch catalog.json
         if: ${{ steps.prod_manifest.outputs.catalog_found == 'true' }}

orchestrate/dags/other_examples/test_oom.py

@@ -21,7 +21,7 @@
     "start_date": datetime(2024, 1, 1),
     "owner": "Fernando Mercado",
     "email_on_failure": False,
-    "retries": 0,
+    "retries": 3,
 }
 
 
@@ -30,7 +30,7 @@
     catchup=False,
     schedule=None,
     default_args=default_args,
-    tags=["test", "oom"],
+    tags=["sample", "maintenance"],
     dag_id="test_oom",
 )
 def test_oom():

secure/snowcap/apply.sh

@@ -75,7 +75,7 @@ if $USE_PII; then
     CONFIG_PATHS="--config $COMBINED_CONFIG"
 else
     ACCOUNT_TO_USE="$SNOWFLAKE_ACCOUNT"
-    EXCLUDE_RESOURCES="--exclude masking_policy,tag,tag_reference,tag_masking_policy_reference,row_access_policy"
+    EXCLUDE_RESOURCES="--exclude masking_policy,tag,tag_reference,tag_masking_policy_reference,row_access_policy,stream"
     USE_ACCOUNT_USAGE=""
     # Standard account - only include base resources
     CONFIG_PATHS="--config resources/"

secure/snowcap/plan.sh

@@ -70,7 +70,7 @@ if $USE_PII; then
     CONFIG_PATHS="--config $COMBINED_CONFIG"
 else
     ACCOUNT_TO_USE="$SNOWFLAKE_ACCOUNT"
-    EXCLUDE_RESOURCES="--exclude masking_policy,tag,tag_reference,tag_masking_policy_reference,row_access_policy"
+    EXCLUDE_RESOURCES="--exclude masking_policy,tag,tag_reference,tag_masking_policy_reference,row_access_policy,stream"
     USE_ACCOUNT_USAGE=""
     # Standard account - only include base resources
     CONFIG_PATHS="--config resources/"

secure/snowcap/resources/object_templates/schema.yml

@@ -6,6 +6,14 @@ schemas:
     owner: "{{ each.value.get('owner', parent.owner) }}"
     managed_access: true
 
+  # BALBOA_QA companion schemas share the BALBOA schema role below.
+  - for_each: var.schemas
+    where: "each.value.name.split('.')[0] == 'BALBOA'"
+    name: "{{ each.value.name.split('.')[1] }}"
+    database: BALBOA_QA
+    owner: "{{ each.value.get('owner', parent.owner) }}"
+    managed_access: true
+
 # Schema roles
 roles:
   - for_each: var.schemas
@@ -17,3 +25,10 @@ grants:
     priv: USAGE
     on: "schema {{ each.value.name }}"
     to: "z_schema__{{ each.value.name.split('.')[1] }}"
+
+  # QA schema grants
+  - for_each: var.schemas
+    where: "each.value.name.split('.')[0] == 'BALBOA'"
+    priv: USAGE
+    on: "schema BALBOA_QA.{{ each.value.name.split('.')[1] }}"
+    to: "z_schema__{{ each.value.name.split('.')[1] }}"

secure/snowcap/resources/roles__functional.yml

@@ -44,6 +44,9 @@ role_grants:
   - to_role: finance_team
     roles:
       - z_db__balboa
+      - z_db__balboa_qa
+      - z_tables_views__select
+      - z_schema__l3_accounts_payable
       - z_schema__l3_loan_analytics
       - z_wh__wh_finance
 
@@ -63,13 +66,18 @@ role_grants:
       - z_schema__l1_google_analytics_4
       - z_schema__l1_loans
       - z_schema__l1_observe
+      - z_schema__l1_erp
       - z_schema__l1_usgs__earthquake_data
       - z_schema__l1_us_population
 
       - z_schema__l2_country_demographics
       - z_schema__l2_covid_observations
+      - z_schema__l2_invoices
+      - z_schema__l2_purchase_orders
       - z_schema__l2_snowflake_usage
+      - z_schema__l2_vendors
 
+      - z_schema__l3_accounts_payable
       - z_schema__l3_covid_analytics
       - z_schema__l3_earthquake_analytics
       - z_schema__l3_loan_analytics

secure/snowcap/resources/schemas.yml

@@ -23,13 +23,18 @@ vars:
       - name: BALBOA.L1_GOOGLE_ANALYTICS_4
       - name: BALBOA.L1_LOANS
       - name: BALBOA.L1_OBSERVE
+      - name: BALBOA.L1_ERP
       - name: BALBOA.L1_USGS__EARTHQUAKE_DATA
       - name: BALBOA.L1_US_POPULATION
 
       - name: BALBOA.L2_COUNTRY_DEMOGRAPHICS
       - name: BALBOA.L2_COVID_OBSERVATIONS
+      - name: BALBOA.L2_INVOICES
+      - name: BALBOA.L2_PURCHASE_ORDERS
       - name: BALBOA.L2_SNOWFLAKE_USAGE
+      - name: BALBOA.L2_VENDORS
 
+      - name: BALBOA.L3_ACCOUNTS_PAYABLE
       - name: BALBOA.L3_COVID_ANALYTICS
       - name: BALBOA.L3_EARTHQUAKE_ANALYTICS
       - name: BALBOA.L3_LOAN_ANALYTICS
@@ -42,23 +47,3 @@ vars:
       # GREAT_BAY DB
       - name: GREAT_BAY.COVE_MARKETING
 
-grants:
-  # BALBOA_QA is a clone of BALBOA (transform/macros/tooling/blue-green).
-  # CREATE DATABASE ... CLONE copies grants on child objects, so every
-  # z_schema__<name> role already reaches the QA copy of its schema -- these
-  # grants exist in Snowflake whether or not we declare them. Declaring them
-  # keeps sync from revoking them.
-  #
-  # Deliberately per-schema rather than "all schemas in database balboa_qa":
-  # roles are scoped by layer (finance_team sees only L3_LOAN_ANALYTICS), and a
-  # database-wide grant would hand every such role the whole QA database.
-  #
-  # The QA schemas themselves stay undeclared -- the clone creates them, and
-  # the `where` keeps this off schemas in RAW, GOVERNANCE and the rest, which
-  # have no QA counterpart.
-  - for_each: var.schemas
-    where: "each.value.name.split('.')[0] == 'BALBOA'"
-    priv: "USAGE"
-    on: "schema BALBOA_QA.{{ each.value.name.split('.')[1] }}"
-    to: "z_schema__{{ each.value.name.split('.')[1] }}"
-

transform/dbt_project.yml

@@ -74,6 +74,8 @@ models:
         +schema: L1_COVID19_EPIDEMIOLOGICAL_DATA
       loans:
         +schema: L1_LOANS
+      erp:
+        +schema: L1_ERP
       observe:
         +schema: L1_OBSERVE
       us_population:
@@ -91,6 +93,12 @@ models:
         +schema: L2_COVID_OBSERVATIONS
       snowflake_usage:
         +schema: L2_SNOWFLAKE_USAGE
+      vendors:
+        +schema: L2_VENDORS
+      invoices:
+        +schema: L2_INVOICES
+      purchase_orders:
+        +schema: L2_PURCHASE_ORDERS
 
     L3_coves:
       +group: marketing
@@ -103,6 +111,8 @@ models:
         +schema: L3_EARTHQUAKE_ANALYTICS
       loan_analytics:
         +schema: L3_LOAN_ANALYTICS
+      accounts_payable:
+        +schema: L3_ACCOUNTS_PAYABLE
                 # cannot persist docs on dynamic tables
                 # +persist_docs:
                 #     relation: false
@@ -153,6 +163,8 @@ data_tests:
 
 vars:
   'dbt_date:time_zone': 'America/Los_Angeles'
+  authored_history_through: '2025-12-01'
+  as_of: '2026-09-08'
 
   # Snowcap governance macros
   snowcap_tag_database: "GOVERNANCE"

transform/macros/spine.sql

@@ -0,0 +1,15 @@
+{% macro trailing_edge_month() %}
+    {% set as_of = var('as_of') %}
+    date_trunc('month', dateadd(month, -1, to_date('{{ as_of }}')))
+{% endmacro %}
+
+{% macro procurement_month_spine() %}
+    {% set trailing_edge = trailing_edge_month() %}
+
+    select
+        cast(
+            dateadd(month, seq4(), dateadd(month, -23, {{ trailing_edge }}))
+            as date
+        ) as month_start
+    from table(generator(rowcount => 24))
+{% endmacro %}

transform/models/L1_inlets/erp/stg_invoices.sql

@@ -0,0 +1,70 @@
+with purchase_orders as (
+
+    select
+        po_id,
+        vendor_id,
+        po_date,
+        po_amount
+    from {{ ref('stg_purchase_orders') }}
+
+),
+
+vendors as (
+
+    select
+        vendor_id,
+        vendor_seq,
+        activity_bucket,
+        invoice_po_through_month,
+        invoice_lag_months
+    from {{ ref('stg_vendors') }}
+
+),
+
+candidate_invoices as (
+
+    select
+        purchase_orders.po_id,
+        purchase_orders.vendor_id,
+        purchase_orders.po_amount,
+        vendors.vendor_seq,
+        cast(dateadd(month, vendors.invoice_lag_months, purchase_orders.po_date) as date) as invoice_date
+    from purchase_orders
+    inner join vendors
+        on purchase_orders.vendor_id = vendors.vendor_id
+    where purchase_orders.po_date <= vendors.invoice_po_through_month
+        and (
+            vendors.activity_bucket <> 'C'
+            or purchase_orders.po_date = vendors.invoice_po_through_month
+        )
+
+),
+
+invoices_in_span as (
+
+    select
+        po_id,
+        vendor_id,
+        po_amount,
+        vendor_seq,
+        invoice_date
+    from candidate_invoices
+    where invoice_date <= {{ trailing_edge_month() }}
+
+)
+
+select
+    replace(po_id, '-PO-', '-INV-')::varchar as invoice_id,
+    po_id,
+    vendor_id,
+    invoice_date,
+    po_amount::bigint as invoice_amount,
+    case
+        when mod(vendor_seq + month(invoice_date), 5) = 0 then null
+        else dateadd(day, 15, invoice_date)::date
+    end as paid_date,
+    case
+        when mod(vendor_seq + month(invoice_date), 5) = 0 then 'open'
+        else 'paid'
+    end::varchar as invoice_status
+from invoices_in_span

transform/models/L1_inlets/erp/stg_invoices.yml

@@ -0,0 +1,46 @@
+version: 2
+
+models:
+  - name: stg_invoices
+    description: Deterministic ERP invoices, including delayed and missing invoice scenarios.
+    columns:
+      - name: invoice_id
+        description: Invoice primary key.
+        data_tests: [not_null, unique]
+      - name: po_id
+        description: Billed purchase order foreign key.
+        data_tests:
+          - not_null
+          - unique
+          - relationships:
+              arguments:
+                to: ref('stg_purchase_orders')
+                field: po_id
+      - name: vendor_id
+        description: Vendor foreign key.
+        data_tests:
+          - not_null
+          - relationships:
+              arguments:
+                to: ref('stg_vendors')
+                field: vendor_id
+      - name: invoice_date
+        description: Invoice activity date.
+        data_tests: [not_null]
+      - name: invoice_amount
+        description: Positive whole-currency invoice amount equal to the PO amount.
+        data_tests:
+          - not_null
+          - dbt_utils.expression_is_true:
+              arguments:
+                expression: '> 0'
+      - name: paid_date
+        description: Payment date; null while the invoice remains open.
+
+      - name: invoice_status
+        description: Payment state for the invoice.
+        data_tests:
+          - not_null
+          - accepted_values:
+              arguments:
+                values: ['paid', 'open']

transform/models/L1_inlets/erp/stg_purchase_orders.sql

@@ -0,0 +1,41 @@
+with month_spine as (
+
+    {{ procurement_month_spine() }}
+
+),
+
+vendors as (
+
+    select
+        vendor_id,
+        category,
+        onboarded_date,
+        po_active_through_month,
+        vendor_seq
+    from {{ ref('stg_vendors') }}
+
+)
+
+select
+    concat(vendors.vendor_id, '-PO-', to_char(month_spine.month_start, 'YYYYMM'))::varchar as po_id,
+    vendors.vendor_id,
+    month_spine.month_start::date as po_date,
+    vendors.category,
+    (
+        case vendors.category
+            when 'CRO Services' then 24000
+            when 'Clinical Reagents' then 18000
+            when 'IT / Software' then 12000
+            when 'Logistics' then 9000
+            when 'Lab Supplies' then 15000
+            when 'Packaging' then 7000
+            when 'Consulting' then 16000
+            when 'Facilities' then 11000
+        end
+        + mod(vendors.vendor_seq, 4) * 125
+    )::bigint as po_amount
+from vendors
+inner join month_spine
+    on month_spine.month_start >= vendors.onboarded_date
+        and month_spine.month_start <= vendors.po_active_through_month
+where mod(datediff(month, vendors.onboarded_date, month_spine.month_start), 3) = 0

transform/models/L1_inlets/erp/stg_purchase_orders.yml

@@ -0,0 +1,30 @@
+version: 2
+
+models:
+  - name: stg_purchase_orders
+    description: Deterministic quarterly ERP purchase orders at monthly date grain.
+    columns:
+      - name: po_id
+        description: Purchase order primary key.
+        data_tests: [not_null, unique]
+      - name: vendor_id
+        description: Vendor foreign key.
+        data_tests:
+          - not_null
+          - relationships:
+              arguments:
+                to: ref('stg_vendors')
+                field: vendor_id
+      - name: po_date
+        description: First day of the purchase order month.
+        data_tests: [not_null]
+      - name: category
+        description: Vendor category carried onto the purchase order.
+        data_tests: [not_null]
+      - name: po_amount
+        description: Positive committed spend in whole currency units.
+        data_tests:
+          - not_null
+          - dbt_utils.expression_is_true:
+              arguments:
+                expression: '> 0'