Merge pull request #11 from FlipsideCrypto/add/contract-reads

add contracts

models/bronze/api_udf/bronze_api__token_reads.sql

@@ -0,0 +1,126 @@
+{{ config(
+    materialized = 'incremental',
+    unique_key = ['contract_address', 'function_sig'],
+    full_refresh = false,
+    tags = ['non_realtime']
+) }}
+
+WITH base AS (
+
+    SELECT
+        contract_address,
+        latest_event_block AS latest_block
+    FROM
+        {{ ref('silver__relevant_contracts') }}
+    WHERE
+        total_event_count >= 25
+
+{% if is_incremental() %}
+AND contract_address NOT IN (
+    SELECT
+        contract_address
+    FROM
+        {{ this }}
+)
+{% endif %}
+ORDER BY
+    total_event_count DESC
+LIMIT
+    200
+), function_sigs AS (
+    SELECT
+        '0x313ce567' AS function_sig,
+        'decimals' AS function_name
+    UNION
+    SELECT
+        '0x06fdde03',
+        'name'
+    UNION
+    SELECT
+        '0x95d89b41',
+        'symbol'
+),
+all_reads AS (
+    SELECT
+        *
+    FROM
+        base
+        JOIN function_sigs
+        ON 1 = 1
+),
+ready_reads AS (
+    SELECT
+        contract_address,
+        latest_block,
+        function_sig,
+        RPAD(
+            function_sig,
+            64,
+            '0'
+        ) AS input,
+        utils.udf_json_rpc_call(
+            'eth_call',
+            [{'to': contract_address, 'from': null, 'data': input}, utils.udf_int_to_hex(latest_block)],
+            concat_ws(
+                '-',
+                contract_address,
+                input,
+                latest_block
+            )
+        ) AS rpc_request
+    FROM
+        all_reads
+),
+batch_reads AS (
+    SELECT
+        ARRAY_AGG(rpc_request) AS batch_rpc_request
+    FROM
+        ready_reads
+),
+node_call AS (
+    SELECT
+        *,
+        {{ target.database }}.live.udf_api(
+            'POST',
+            '{Service}',{},
+            batch_rpc_request,
+            'Vault/prod/klaytn/blockjoy/mainnet'
+        ) AS response
+    FROM
+        batch_reads
+    WHERE
+        EXISTS (
+            SELECT
+                1
+            FROM
+                ready_reads
+            LIMIT
+                1
+        )
+), flat_responses AS (
+    SELECT
+        VALUE :id :: STRING AS call_id,
+        VALUE :result :: STRING AS read_result
+    FROM
+        node_call,
+        LATERAL FLATTEN (
+            input => response :data
+        )
+)
+SELECT
+    SPLIT_PART(
+        call_id,
+        '-',
+        1
+    ) AS contract_address,
+    SPLIT_PART(
+        call_id,
+        '-',
+        3
+    ) AS block_number,
+    LEFT(SPLIT_PART(call_id, '-', 2), 10) AS function_sig,
+    NULL AS function_input,
+    read_result,
+    SYSDATE() :: TIMESTAMP AS _inserted_timestamp
+FROM
+    flat_responses

models/bronze/api_udf/bronze_api__token_reads.yml

@@ -0,0 +1,18 @@
+version: 2
+models:
+  - name: bronze_api__token_reads
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns:
+            - CONTRACT_ADDRESS
+            - FUNCTION_SIG
+    columns:
+      - name: _INSERTED_TIMESTAMP
+        tests:
+          - not_null
+          - dbt_expectations.expect_row_values_to_have_recent_data:
+              datepart: day
+              interval: 1
+          - dbt_expectations.expect_column_values_to_be_in_type_list:
+              column_type_list:
+                - TIMESTAMP_NTZ

models/silver/core/silver__contracts.sql

@@ -0,0 +1,109 @@
+{{ config(
+    materialized = 'incremental',
+    unique_key = 'contract_address',
+    merge_exclude_columns = ["inserted_timestamp"],
+    tags = ['non_realtime']
+) }}
+
+WITH base_metadata AS (
+
+    SELECT
+        contract_address,
+        block_number,
+        function_sig AS function_signature,
+        read_result AS read_output,
+        _inserted_timestamp
+    FROM
+        {{ ref('bronze_api__token_reads') }}
+    WHERE
+        read_result IS NOT NULL
+        AND read_result <> '0x'
+
+{% if is_incremental() %}
+AND _inserted_timestamp >= (
+    SELECT
+        MAX(
+            _inserted_timestamp
+        )
+    FROM
+        {{ this }}
+)
+{% endif %}
+),
+token_names AS (
+    SELECT
+        contract_address,
+        block_number,
+        function_signature,
+        read_output,
+        utils.udf_hex_to_string(
+            SUBSTR(read_output,(64 * 2 + 3), len(read_output))) AS token_name
+            FROM
+                base_metadata
+            WHERE
+                function_signature = '0x06fdde03'
+                AND token_name IS NOT NULL
+        ),
+        token_symbols AS (
+            SELECT
+                contract_address,
+                block_number,
+                function_signature,
+                read_output,
+                utils.udf_hex_to_string(
+                    SUBSTR(read_output,(64 * 2 + 3), len(read_output))) AS token_symbol
+                    FROM
+                        base_metadata
+                    WHERE
+                        function_signature = '0x95d89b41'
+                        AND token_symbol IS NOT NULL
+                ),
+                token_decimals AS (
+                    SELECT
+                        contract_address,
+                        CASE
+                            WHEN read_output IS NOT NULL THEN utils.udf_hex_to_int(
+                                read_output :: STRING
+                            )
+                            ELSE NULL
+                        END AS token_decimals,
+                        LENGTH(token_decimals) AS dec_length
+                    FROM
+                        base_metadata
+                    WHERE
+                        function_signature = '0x313ce567'
+                        AND read_output IS NOT NULL
+                        AND read_output <> '0x'
+                ),
+                contracts AS (
+                    SELECT
+                        contract_address,
+                        MAX(_inserted_timestamp) AS _inserted_timestamp
+                    FROM
+                        base_metadata
+                    GROUP BY
+                        1
+                )
+            SELECT
+                c1.contract_address :: STRING AS contract_address,
+                token_name,
+                TRY_TO_NUMBER(token_decimals) AS token_decimals,
+                token_symbol,
+                _inserted_timestamp,
+                {{ dbt_utils.generate_surrogate_key(
+                    ['c1.contract_address']
+                ) }} AS contracts_id,
+                SYSDATE() AS inserted_timestamp,
+                SYSDATE() AS modified_timestamp,
+                '{{ invocation_id }}' AS _invocation_id
+            FROM
+                contracts c1
+                LEFT JOIN token_names
+                ON c1.contract_address = token_names.contract_address
+                LEFT JOIN token_symbols
+                ON c1.contract_address = token_symbols.contract_address
+                LEFT JOIN token_decimals
+                ON c1.contract_address = token_decimals.contract_address
+                AND dec_length < 3 qualify(ROW_NUMBER() over(PARTITION BY c1.contract_address
+            ORDER BY
+                _inserted_timestamp DESC)) = 1