feat: add open collective funding data to event table (#2185)

* feat: add open collective funding data to `event` table

* fix: use `for` loop for queries

* fix: filter out non `USD` transactions

* fix: filter out transactions in event `macro`

warehouse/dbt/macros/models/open_collective_events.sql

@@ -0,0 +1,36 @@
+{% macro open_collective_events(source_ref) %}
+
+  {% set query %}
+    (
+      select -- noqa: ST06
+        created_at as `time`,
+        type as event_type,
+        id as event_source_id,
+        "OPEN_COLLECTIVE" as event_source,
+        {{ oso_id(
+          "'OPEN_COLLECTIVE'",
+          "JSON_VALUE(to_account, '$.slug')",
+          "JSON_VALUE(to_account, '$.name')")
+        }} as to_artifact_id,
+        JSON_VALUE(to_account, "$.name") as to_name,
+        JSON_VALUE(to_account, "$.slug") as to_namespace,
+        JSON_VALUE(to_account, "$.type") as to_type,
+        JSON_VALUE(to_account, "$.id") as to_artifact_source_id,
+        {{ oso_id(
+          "'OPEN_COLLECTIVE'",
+          "JSON_VALUE(from_account, '$.slug')",
+          "JSON_VALUE(from_account, '$.name')")
+        }} as from_artifact_id,
+        JSON_VALUE(from_account, "$.name") as from_name,
+        JSON_VALUE(from_account, "$.slug") as from_namespace,
+        JSON_VALUE(from_account, "$.type") as from_type,
+        JSON_VALUE(from_account, "$.id") as from_artifact_source_id,
+        ABS(CAST(JSON_VALUE(amount, "$.value") as FLOAT64)) as amount
+      from {{ ref(source_ref) }}
+      where JSON_VALUE(amount, "$.currency") = "USD"
+    )
+  {% endset %}
+
+  {{ return(query) }}
+
+{% endmacro %}

warehouse/dbt/models/intermediate/events/int_events.sql

@@ -247,6 +247,26 @@ all_events as (
     union all
     select * from github_comments
   )
+  union all
+  select
+    time,
+    event_type,
+    event_source_id,
+    event_source,
+    to_artifact_id,
+    to_name,
+    to_namespace,
+    to_type,
+    to_artifact_source_id,
+    from_artifact_id,
+    from_name,
+    from_namespace,
+    from_type,
+    from_artifact_source_id,
+    amount
+  from (
+    select * from {{ ref("stg_open_collective__events") }}
+  )
 )
 
 select

warehouse/dbt/models/open_collective_sources.yml

@@ -0,0 +1,9 @@
+sources:
+  - name: open_collective
+    database: opensource-observer
+    schema: open_collective
+    tables:
+      - name: expenses
+        identifier: expenses
+      - name: deposits
+        identifier: deposits

warehouse/dbt/models/staging/open-collective/stg_open_collective__deposits.sql

@@ -0,0 +1,24 @@
+{% set columns = [
+    "id", "legacy_id", "group", "type", "kind", "description", "amount", 
+    "amount_in_host_currency", "host_currency_fx_rate", "net_amount", 
+    "net_amount_in_host_currency", "tax_amount", "tax_info", "platform_fee", 
+    "host_fee", "payment_processor_fee", "account", "from_account", 
+    "to_account", "expense", "order", "created_at", "updated_at", 
+    "is_refunded", "is_refund", "is_disputed", "is_in_review", 
+    "payment_method", "payout_method", "is_order_rejected", "merchant_id", 
+    "invoice_template", "host"
+] %}
+
+with source as (
+  select * from {{ source('open_collective', 'deposits') }}
+),
+
+renamed as (
+  select
+  {% for column in columns %}
+    {{ adapter.quote(column) }}{% if not loop.last %},{% endif %}
+  {% endfor %}
+  from source
+)
+
+select * from renamed

warehouse/dbt/models/staging/open-collective/stg_open_collective__events.sql

@@ -0,0 +1,13 @@
+with open_collective_expenses as (
+  select *
+  from {{ open_collective_events("stg_open_collective__expenses") }}
+),
+
+open_collective_deposits as (
+  select *
+  from {{ open_collective_events("stg_open_collective__deposits") }}
+)
+
+select * from open_collective_expenses
+union all
+select * from open_collective_deposits

warehouse/dbt/models/staging/open-collective/stg_open_collective__expenses.sql

@@ -0,0 +1,24 @@
+{% set columns = [
+    "id", "legacy_id", "group", "type", "kind", "description", "amount", 
+    "amount_in_host_currency", "host_currency_fx_rate", "net_amount", 
+    "net_amount_in_host_currency", "tax_amount", "tax_info", "platform_fee", 
+    "host_fee", "payment_processor_fee", "account", "from_account", 
+    "to_account", "expense", "order", "created_at", "updated_at", 
+    "is_refunded", "is_refund", "is_disputed", "is_in_review", 
+    "payment_method", "payout_method", "is_order_rejected", "merchant_id", 
+    "invoice_template", "host"
+] %}
+
+with source as (
+  select * from {{ source('open_collective', 'expenses') }}
+),
+
+renamed as (
+  select
+  {% for column in columns %}
+    {{ adapter.quote(column) }}{% if not loop.last %},{% endif %}
+  {% endfor %}
+  from source
+)
+
+select * from renamed