diff --git a/models/gold/nft/nft__dim_topshot_metadata.sql b/models/gold/nft/nft__dim_topshot_metadata.sql index 3a999b8..232ce88 100644 --- a/models/gold/nft/nft__dim_topshot_metadata.sql +++ b/models/gold/nft/nft__dim_topshot_metadata.sql @@ -1,17 +1,46 @@ -{{ config ( - materialized = 'view', - meta={ - 'database_tags':{ - 'table': { - 'PURPOSE': 'NFT, TOPSHOT' - } - } - }, - tag = ['scheduled'] +{{ config( + materialized = 'incremental', + incremental_strategy = 'merge', + merge_exclude_columns = ['inserted_timestamp'], + cluster_by = ['left(season,4)'], + unique_key = "nft_id", + post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION ON EQUALITY(nft_id,nbatopshot_id);", + tags = ['scheduled_non_core'], + meta ={ 'database_tags':{ 'table':{ 'PURPOSE': 'NFT, TOPSHOT' }} } ) }} +-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }} +WITH -WITH topshot AS ( +{% if is_incremental() %} +{% else %} + topshot_old AS ( + SELECT + nft_id, + nft_collection, + nbatopshot_id, + serial_number, + total_circulation, + moment_description, + player, + team, + season, + play_category, + play_type, + moment_date, + set_name, + set_series_number, + video_urls, + moment_stats_full, + player_stats_game, + player_stats_season_to_date, + nft_moment_metadata_topshot_id AS dim_topshot_metadata_id + FROM + {{ ref('silver__nft_topshot_metadata_view') }} + ), +{% endif %} + +topshot AS ( SELECT nft_id, nft_collection, @@ -31,13 +60,74 @@ WITH topshot AS ( moment_stats_full, player_stats_game, player_stats_season_to_date, - nft_moment_metadata_topshot_id as dim_topshot_metadata_id, - inserted_timestamp, - modified_timestamp + nft_topshot_metadata_v2_id AS dim_topshot_metadata_id FROM - {{ ref('silver__nft_topshot_metadata') }} + {{ ref('silver__nft_topshot_metadata_v2') }} + +{% if is_incremental() %} +WHERE + modified_timestamp >= ( + SELECT + MAX(modified_timestamp) modified_timestamp + FROM + {{ this }} + ) +{% endif %} +), +ua AS ( + SELECT + nft_id, + nft_collection, + nbatopshot_id, + serial_number, + total_circulation, + moment_description, + player, + team, + season, + play_category, + play_type, + moment_date, + set_name, + set_series_number, + video_urls, + moment_stats_full, + player_stats_game, + player_stats_season_to_date, + dim_topshot_metadata_id + FROM + topshot + +{% if is_incremental() %} +{% else %} + UNION ALL + SELECT + nft_id, + nft_collection, + nbatopshot_id, + serial_number, + total_circulation, + moment_description, + player, + team, + season, + play_category, + play_type, + moment_date, + set_name, + set_series_number, + video_urls, + moment_stats_full, + player_stats_game, + player_stats_season_to_date, + dim_topshot_metadata_id + FROM + topshot_old + {% endif %} ) SELECT - * + *, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp FROM - topshot + ua diff --git a/models/gold/nft/nft__fact_topshot_buybacks.sql b/models/gold/nft/nft__fact_topshot_buybacks.sql new file mode 100644 index 0000000..ca25235 --- /dev/null +++ b/models/gold/nft/nft__fact_topshot_buybacks.sql @@ -0,0 +1,124 @@ +{{ config( + materialized = 'incremental', + incremental_strategy = 'merge', + merge_exclude_columns = ['inserted_timestamp'], + incremental_predicates = ["COALESCE(DBT_INTERNAL_DEST.block_timestamp::DATE,'2099-12-31') >= (select min(block_timestamp::DATE) from " ~ generate_tmp_view_name(this) ~ ")"], + cluster_by = ['block_timestamp::date', 'modified_timestamp::date'], + unique_key = "topshot_buyback_id", + tags = ['nft', 'topshot', 'scheduled'], + meta = { 'database_tags': { 'table': { 'PURPOSE': 'NFT, TOPSHOT' } } } +) }} + +WITH flowty_sales AS ( + SELECT + tx_id AS tx_id, + block_timestamp, + EVENT_DATA:buyer :: string AS buyer, + event_data:storefrontAddress :: string AS seller, + CAST(EVENT_DATA:"salePrice" AS DECIMAL(18, 2)) AS price, + event_data:salePaymentVaultType as currency, + 'A.3cdbb3d569211ff3.NFTStorefrontV2' as marketplace, + 'FLOWTY' as sale_type, + EVENT_DATA:nftType :: string as nft_collection, + EVENT_DATA:nftID :: string as nft_id, + modified_timestamp + FROM + {{ ref('core__fact_events') }} AS events + WHERE + EVENT_CONTRACT IN ('A.3cdbb3d569211ff3.NFTStorefrontV2') + AND EVENT_TYPE = 'ListingCompleted' + AND TX_SUCCEEDED = TRUE + AND EVENT_DATA:purchased :: string = 'true' + AND CAST(EVENT_DATA:"salePrice" AS DECIMAL(18, 2)) > 0 + AND EVENT_DATA:nftType :: string = 'A.0b2a3299cc857e29.TopShot' + + {% if is_incremental() %} + AND modified_timestamp >= ( + SELECT + MAX(modified_timestamp) + FROM + {{ this }} + ) + {% endif %} +), + +all_sales AS ( + SELECT + tx_id AS tx_id, + CONVERT_TIMEZONE('UTC', 'America/New_York', BLOCK_TIMESTAMP) as block_timestamp, + buyer AS buyer, + seller AS seller, + price AS price, + nft_collection AS nft_collection, + nft_id AS nft_id, + modified_timestamp + FROM {{ ref('nft__ez_nft_sales') }} AS sales + WHERE nft_collection = 'A.0b2a3299cc857e29.TopShot' + AND TX_SUCCEEDED = TRUE + + {% if is_incremental() %} + AND modified_timestamp >= ( + SELECT + MAX(modified_timestamp) + FROM + {{ this }} + ) + {% endif %} + + UNION ALL + + SELECT + tx_id AS tx_id, + block_timestamp AS block_timestamp, + buyer AS buyer, + seller AS seller, + price AS price, + nft_collection AS nft_collection, + nft_id AS nft_id, + modified_timestamp + FROM flowty_sales AS fs +), + + sales_with_running_total AS ( + SELECT + block_timestamp AS block_timestamp, + DATE_TRUNC('DAY', block_timestamp) as block_day, + tx_id AS tx_id, + nft_id AS nft_id, + buyer AS buyer, + seller AS seller, + price AS price, + 1 as sale_count, + ROW_NUMBER() OVER (PARTITION BY tx_id, nft_id ORDER BY block_timestamp) as rn, + SUM(price) OVER (ORDER BY block_timestamp ROWS UNBOUNDED PRECEDING) as running_total, + modified_timestamp + FROM all_sales AS asales + WHERE buyer = '0xe1f2a091f7bb5245' -- Filtering for TopShot buyback wallet + ) + + SELECT + s.block_timestamp AS block_timestamp, + s.block_height AS block_height, + s.tx_id AS tx_id, + s.nft_id AS nft_id, + COALESCE(ts.player, mm.metadata:player::string) as player, + COALESCE(ts.team, mm.metadata:team::string) as team, + COALESCE(ts.season, mm.metadata:season::string) as season, + COALESCE(ts.set_name, mm.set_name) as set_name, + s.buyer AS buyer, + s.seller AS seller, + s.price AS price, + s.sale_count as sale, + s.running_total as total, + CONCAT($$https://nbatopshot.com/moment/$$, s.nft_id) AS URL, + {{ dbt_utils.generate_surrogate_key(['s.tx_id', 's.nft_id']) }} AS topshot_buyback_id, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp, + '{{ invocation_id }}' AS _invocation_id + FROM sales_with_running_total s + LEFT JOIN {{ ref('nft__dim_topshot_metadata') }} ts + ON s.nft_id = ts.nft_id + LEFT JOIN {{ ref('nft__dim_moment_metadata') }} mm + ON s.nft_id = mm.nft_id + AND ts.player IS NULL -- Only pull from moment_metadata if topshot_metadata is empty + WHERE s.rn = 1 -- Deduplicate if needed \ No newline at end of file diff --git a/models/gold/nft/nft__fact_topshot_buybacks.yml b/models/gold/nft/nft__fact_topshot_buybacks.yml new file mode 100644 index 0000000..962d2a7 --- /dev/null +++ b/models/gold/nft/nft__fact_topshot_buybacks.yml @@ -0,0 +1,155 @@ +version: 2 + +models: + - name: nft__fact_topshot_buybacks + description: |- + This table captures NBA TopShot buyback activity - transactions where the TopShot buyback wallet purchases moments from users. + The model combines sales from both standard marketplace and Flowty, and includes a running total of amount spent over time. + tests: + - dbt_utils.recency: + datepart: day + field: block_timestamp + interval: 3 + + columns: + - name: block_timestamp + description: "Timestamp of the block in Eastern time (America/New_York)" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - TIMESTAMP_NTZ + + - name: block_height + description: "Height of the block in the blockchain" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - NUMBER + - INTEGER + + - name: tx_id + description: "{{ doc('tx_id') }}" + tests: + - not_null + + - name: nft_id + description: "{{ doc('nft_id') }}" + tests: + - not_null + + - name: player + description: "Player name from TopShot metadata or moment metadata" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: team + description: "Team name from TopShot metadata or moment metadata" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: season + description: "Season from TopShot metadata or moment metadata" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: set_name + description: "Set name from TopShot metadata or moment metadata" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: buyer + description: "Buyer address (TopShot buyback wallet)" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + - accepted_values: + values: ['0xe1f2a091f7bb5245'] + + - name: seller + description: "Seller address" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: price + description: "Price of the moment in USD" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - NUMBER + - FLOAT + - DECIMAL + + - name: sale + description: "Counter field, always 1 representing a single sale transaction" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - NUMBER + - INTEGER + + - name: total + description: "Running total of all buyback purchases up to this transaction" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - NUMBER + - FLOAT + - DECIMAL + + - name: URL + description: "Direct link to the TopShot moment on the website" + tests: + - not_null + - dbt_expectations.expect_column_values_to_be_in_type_list: + column_type_list: + - STRING + - VARCHAR + + - name: topshot_buyback_id + description: "Surrogate key generated from tx_id and nft_id" + tests: + - not_null + - unique + + - name: inserted_timestamp + description: "Timestamp when the record was inserted" + tests: + - not_null + + - name: modified_timestamp + description: "Timestamp when the record was last modified" + tests: + - not_null + + - name: _invocation_id + description: "Unique identifier for the dbt run that created this record" + tests: + - not_null \ No newline at end of file diff --git a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql index c59f2c5..af5a8e1 100644 --- a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql +++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata.sql @@ -4,22 +4,23 @@ cluster_by = ['_inserted_timestamp::DATE'], unique_key = 'nft_id', tags = ['livequery', 'topshot'], - post_hook = "ALTER TABLE {{ this }} ADD SEARCH OPTIMIZATION ON EQUALITY(nft_id,nbatopshot_id);", - full_refresh = False + full_refresh = False, + enabled = false ) }} -{# NFT Metadata from legacy process lives in external table, deleted CTE and set FR=False -to limit / avoid unnecessary table scans #} +{# NFT Metadata from legacy process lives in external table, deleted CTE and set FR=False +TO -WITH metadata_lq AS ( - - SELECT - _res_id, - 'A.0b2a3299cc857e29.TopShot' AS contract, - moment_id, - DATA :data :data :: variant AS DATA, - _inserted_timestamp - FROM - {{ ref('livequery__request_topshot_metadata') }} +LIMIT + / avoid unnecessary TABLE scans #} + WITH metadata_lq AS ( + SELECT + _res_id, + 'A.0b2a3299cc857e29.TopShot' AS contract, + moment_id, + DATA :data :data :: variant AS DATA, + _inserted_timestamp + FROM + {{ ref('livequery__request_topshot_metadata') }} {% if is_incremental() %} WHERE @@ -33,8 +34,8 @@ WHERE ), lq_final AS ( SELECT - moment_id AS nft_id, - contract AS nft_collection, + moment_id :: STRING AS nft_id, + contract :: STRING AS nft_collection, DATA :getMintedMoment :data :id :: STRING AS nbatopshot_id, DATA :getMintedMoment :data :flowSerialNumber :: NUMBER AS serial_number, DATA :getMintedMoment :data :setPlay :circulationCount :: NUMBER AS total_circulation, @@ -60,11 +61,11 @@ lq_final AS ( SELECT *, {{ dbt_utils.generate_surrogate_key( - ['nft_id'] - ) }} AS nft_moment_metadata_topshot_id, + ['nft_id'] + ) }} AS nft_moment_metadata_topshot_id, SYSDATE() AS inserted_timestamp, SYSDATE() AS modified_timestamp, - '{{ invocation_id }}' AS _invocation_id + '{{ invocation_id }}' AS _invocation_id FROM lq_final qualify ROW_NUMBER() over ( PARTITION BY nft_id diff --git a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_v2.sql b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_v2.sql new file mode 100644 index 0000000..184b3be --- /dev/null +++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_v2.sql @@ -0,0 +1,64 @@ +{{ config( + materialized = 'incremental', + incremental_strategy = 'delete+insert', + cluster_by = ['_inserted_timestamp::DATE'], + unique_key = 'nft_id', + tags = ['streamline', 'topshot'] +) }} +-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }} +WITH metadata_from_streamline AS ( + + SELECT + VALUE :CONTRACT AS contract, + VALUE :ID AS moment_id, + DATA, + _inserted_timestamp + FROM + +{% if is_incremental() %} +{{ ref('bronze__streamline_topshot_metadata') }} +WHERE + _inserted_timestamp >= ( + SELECT + MAX(_inserted_timestamp) _inserted_timestamp + FROM + {{ this }} + ) +{% else %} + {{ ref('bronze__streamline_topshot_metadata_FR') }} +{% endif %} +) +SELECT + moment_id :: STRING AS nft_id, + contract :: STRING AS nft_collection, + DATA :data :getMintedMoment :data :id :: STRING AS nbatopshot_id, + DATA :data :getMintedMoment :data :flowSerialNumber :: NUMBER AS serial_number, + DATA :data :getMintedMoment :data :setPlay :circulationCount :: NUMBER AS total_circulation, + DATA :data :getMintedMoment :data :play :description :: VARCHAR AS moment_description, + DATA :data :getMintedMoment :data :play :stats :playerName :: STRING AS player, + DATA :data :getMintedMoment :data :play :stats :teamAtMoment :: STRING AS team, + DATA :data :getMintedMoment :data :play :stats :nbaSeason :: STRING AS season, + DATA :data :getMintedMoment :data :play :stats :playCategory :: STRING AS play_category, + DATA :data :getMintedMoment :data :play :stats :playType :: STRING AS play_type, + DATA :data :getMintedMoment :data :play :stats :dateOfMoment :: TIMESTAMP AS moment_date, + DATA :data :getMintedMoment :data :set :flowName :: STRING AS set_name, + DATA :data :getMintedMoment :data :set :flowSeriesNumber :: NUMBER AS set_series_number, + DATA :data :getMintedMoment :data :play :assets :videos :: ARRAY AS video_urls, + DATA :data :getMintedMoment :data :play :stats :: OBJECT AS moment_stats_full, + DATA :data :getMintedMoment :data :play :statsPlayerGameScores :: OBJECT AS player_stats_game, + DATA :data :getMintedMoment :data :play :statsPlayerSeasonAverageScores :: OBJECT AS player_stats_season_to_date, + _inserted_timestamp, + {{ dbt_utils.generate_surrogate_key( + ['nft_id'] + ) }} AS nft_topshot_metadata_v2_id, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp, + '{{ invocation_id }}' AS _invocation_id +FROM + metadata_from_streamline +WHERE + DATA :errors IS NULL qualify ROW_NUMBER() over ( + PARTITION BY nft_id + ORDER BY + _inserted_timestamp DESC + ) = 1 diff --git a/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_view.sql b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_view.sql new file mode 100644 index 0000000..9ff1a38 --- /dev/null +++ b/models/silver/nft/metadata/livequery/topshot/silver__nft_topshot_metadata_view.sql @@ -0,0 +1,33 @@ +{{ config( + materialized = 'view' +) }} + +SELECT + nft_id, + nft_collection, + nbatopshot_id, + serial_number, + total_circulation, + moment_description, + player, + team, + season, + play_category, + play_type, + moment_date, + set_name, + set_series_number, + video_urls, + moment_stats_full, + player_stats_game, + player_stats_season_to_date, + _INSERTED_TIMESTAMP, + nft_moment_metadata_topshot_id, + inserted_timestamp, + modified_timestamp, + _INVOCATION_ID +FROM + {{ source( + 'silver', + 'nft_topshot_metadata' + ) }} diff --git a/models/sources.yml b/models/sources.yml index 0e1692d..fc8e9de 100644 --- a/models/sources.yml +++ b/models/sources.yml @@ -132,6 +132,7 @@ sources: - name: contract_abis - name: evm_traces_v2 - name: evm_decoded_logs + - name: topshot_metadata - name: crosschain_silver database: crosschain @@ -184,4 +185,10 @@ sources: - name: evm_known_event_sigs - name: evm_known_event_names - name: evm_event_sigs - - name: dates \ No newline at end of file + - name: dates + + + - name: silver + schema: silver + tables: + - name: nft_topshot_metadata \ No newline at end of file diff --git a/models/streamline/external/topshot/bronze__streamline_topshot_metadata.sql b/models/streamline/external/topshot/bronze__streamline_topshot_metadata.sql new file mode 100644 index 0000000..e49f772 --- /dev/null +++ b/models/streamline/external/topshot/bronze__streamline_topshot_metadata.sql @@ -0,0 +1,8 @@ +{{ config ( + materialized = 'view', + tags = ['streamline_non_core'] +) }} +{{ streamline_external_table_query_v2( + model = "topshot_metadata", + partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )" +) }} diff --git a/models/streamline/external/topshot/bronze__streamline_topshot_metadata_FR.sql b/models/streamline/external/topshot/bronze__streamline_topshot_metadata_FR.sql new file mode 100644 index 0000000..a4b56af --- /dev/null +++ b/models/streamline/external/topshot/bronze__streamline_topshot_metadata_FR.sql @@ -0,0 +1,8 @@ +{{ config ( + materialized = 'view', + tags = ['streamline_non_core'] +) }} +{{ streamline_external_table_FR_query_v2( + model = "topshot_metadata", + partition_function = "CAST(SPLIT_PART(SPLIT_PART(file_name, '/', 3), '_', 1) AS INTEGER )" +) }} diff --git a/models/streamline/external/topshot/streamline__topshot_metadata_complete.sql b/models/streamline/external/topshot/streamline__topshot_metadata_complete.sql new file mode 100644 index 0000000..e1912bf --- /dev/null +++ b/models/streamline/external/topshot/streamline__topshot_metadata_complete.sql @@ -0,0 +1,68 @@ +-- depends_on: {{ ref('bronze__streamline_topshot_metadata') }} +-- depends_on: {{ ref('bronze__streamline_topshot_metadata_FR') }} +{{ config ( + materialized = "incremental", + unique_key = "topshot_metadata_complete_id", + merge_exclude_columns = ["inserted_timestamp"], + tags = ['streamline_complete'] +) }} + +SELECT + VALUE :CONTRACT :: STRING AS event_contract, + VALUE :id :: STRING AS moment_id, + DATA :errors :extensions :error_reason :: STRING AS error_reason, + partition_key :: STRING AS partition_key, + _inserted_timestamp, + {{ dbt_utils.generate_surrogate_key( + ['event_contract','moment_id'] + ) }} AS topshot_metadata_complete_id, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp, + '{{ invocation_id }}' AS _invocation_id +FROM + +{% if is_incremental() %} +{{ ref('bronze__streamline_topshot_metadata') }} +WHERE + _inserted_timestamp >= COALESCE( + ( + SELECT + MAX(_inserted_timestamp) _inserted_timestamp + FROM + {{ this }} + ), + '1900-01-01' :: timestamp_ntz + ) +{% else %} + {{ ref('bronze__streamline_topshot_metadata_FR') }} +{% endif %} + +{% if is_incremental() %} +{% else %} + UNION ALL + SELECT + contract AS event_contract, + id AS moment_id, + CASE + WHEN len( + DATA :getMintedMoment + ) IS NULL THEN 'null data' + END error_reason, + to_char(TO_TIMESTAMP_NTZ(SYSDATE()), 'YYYYMMDD') AS partition_key, + _INSERTED_DATE AS _inserted_timestamp, + {{ dbt_utils.generate_surrogate_key( + ['event_contract','moment_id'] + ) }} AS topshot_metadata_complete_id, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp, + '{{ invocation_id }}' AS _invocation_id + FROM + {{ source( + 'bronze_streamline', + 'moments_minted_metadata_api' + ) }} + {% endif %} + + qualify(ROW_NUMBER() over (PARTITION BY topshot_metadata_complete_id +ORDER BY + _inserted_timestamp DESC)) = 1 diff --git a/models/streamline/external/topshot/streamline__topshot_metadata_realtime.sql b/models/streamline/external/topshot/streamline__topshot_metadata_realtime.sql new file mode 100644 index 0000000..4f2aeea --- /dev/null +++ b/models/streamline/external/topshot/streamline__topshot_metadata_realtime.sql @@ -0,0 +1,71 @@ +{{ config( + materialized = 'view', + tags = ['streamline', 'topshot', 'moments_metadata', 'backfill'], + post_hook = fsc_utils.if_data_call_function_v2( + func = '{{this.schema}}.udf_bulk_rest_api_v2', + target = "{{this.schema}}.{{this.identifier}}", + params ={ "external_table": "topshot_metadata", + "sql_limit": "500", + "producer_batch_size": "100", + "worker_batch_size": "100", + "sql_source": "{{this.identifier}}", + "async_concurrent_requests": "10" } + ) +) }} + +WITH api_parameters AS ( + -- Use the same parameters as for realtime + + SELECT + base_url, + query + FROM + {{ ref('streamline__topshot_parameters') }} + WHERE + contract = 'A.0b2a3299cc857e29.TopShot' +), +work_todo AS ( + SELECT + event_contract, + moment_id + FROM + {{ ref('streamline__topshot_moments') }} + EXCEPT + SELECT + event_contract, + moment_id + FROM + {{ ref('streamline__topshot_metadata_complete') }} +) +SELECT + to_char(TO_TIMESTAMP_NTZ(SYSDATE()), 'YYYYMMDD') AS partition_key, + m.event_contract AS contract, + m.moment_id AS id, + {{ target.database }}.live.udf_api( + 'POST', + p.base_url, + OBJECT_CONSTRUCT( + 'Accept', + 'application/json', + 'Accept-Encoding', + 'gzip', + 'Connection', + 'keep-alive', + 'Content-Type', + 'application/json', + 'User-Agent', + 'Flipside_Flow_metadata/0.1' + ), + OBJECT_CONSTRUCT( + 'query', + p.query, + 'variables', + OBJECT_CONSTRUCT( + 'momentId', + m.moment_id + ) + ) + ) AS request +FROM + work_todo m + CROSS JOIN api_parameters p diff --git a/models/streamline/external/topshot/streamline__topshot_moments.sql b/models/streamline/external/topshot/streamline__topshot_moments.sql new file mode 100644 index 0000000..b087812 --- /dev/null +++ b/models/streamline/external/topshot/streamline__topshot_moments.sql @@ -0,0 +1,85 @@ +{{ config ( + materialized = "incremental", + unique_key = ["moment_id","event_contract"], + merge_exclude_columns = ["inserted_timestamp"], + tags = ['streamline_complete_evm'] +) }} + +WITH mints AS ( + + SELECT + event_contract, + event_data :momentID :: STRING AS moment_id, + _inserted_timestamp + FROM + {{ ref('silver__nft_moments_s') }} + WHERE + event_contract = 'A.0b2a3299cc857e29.TopShot' + AND event_type = 'MomentMinted' + +{% if is_incremental() %} +AND _inserted_timestamp >= COALESCE( + ( + SELECT + MAX(_inserted_timestamp) _inserted_timestamp + FROM + {{ this }} + ), + '1900-01-01' :: timestamp_ntz +) +{% endif %} +), +sales AS ( + SELECT + nft_collection AS event_contract, + nft_id AS moment_id, + _inserted_timestamp + FROM + {{ ref('silver__nft_sales_s') }} + WHERE + nft_collection ILIKE '%topshot%' + +{% if is_incremental() %} +AND _inserted_timestamp >= COALESCE( + ( + SELECT + MAX(_inserted_timestamp) _inserted_timestamp + FROM + {{ this }} + ), + '1900-01-01' :: timestamp_ntz +) +{% endif %} +), +all_topshots AS ( + SELECT + event_contract, + moment_id, + _inserted_timestamp + FROM + mints + UNION ALL + SELECT + event_contract, + moment_id, + _inserted_timestamp + FROM + sales +) +SELECT + event_contract, + moment_id, + _inserted_timestamp, + {{ dbt_utils.generate_surrogate_key( + ['event_contract','moment_id'] + ) }} AS topshot_moments_id, + SYSDATE() AS inserted_timestamp, + SYSDATE() AS modified_timestamp, + '{{ invocation_id }}' AS _invocation_id +FROM + all_topshots qualify ROW_NUMBER() over ( + PARTITION BY event_contract, + moment_id + ORDER BY + _inserted_timestamp DESC + ) = 1 diff --git a/models/streamline/external/topshot/streamline__topshot_parameters.sql b/models/streamline/external/topshot/streamline__topshot_parameters.sql new file mode 100644 index 0000000..c6676cd --- /dev/null +++ b/models/streamline/external/topshot/streamline__topshot_parameters.sql @@ -0,0 +1,207 @@ + {{ config( + materialized = 'table', + unique_key = 'contract', + tags = ['livequery', 'topshot', 'allday', 'moment_metadata'] + ) }} + + SELECT + 'A.0b2a3299cc857e29.TopShot' AS contract, + 'https://public-api.nbatopshot.com/graphql' as base_url, + 'query getMintedMoment($momentId: ID!) { + getMintedMoment(momentId: $momentId) { + data { + id + version + sortID + set { + id + sortID + version + flowId + flowName + flowSeriesNumber + flowLocked + setVisualId + assetPath + assets { + images { + type + url + } + } + } + play { + id + version + description + flowID + sortID + status + assets { + videos { + type + url + videoLength + } + videoLengthInMilliseconds + } + stats { + playerID + playerName + firstName + lastName + jerseyNumber + teamAtMoment + awayTeamName + awayTeamScore + homeTeamName + homeTeamScore + dateOfMoment + totalYearsExperience + teamAtMomentNbaId + height + weight + currentTeam + currentTeamId + primaryPosition + homeTeamNbaId + awayTeamNbaId + nbaSeason + draftYear + draftSelection + draftRound + birthplace + birthdate + draftTeam + draftTeamNbaId + playCategory + playType + quarter + } + statsPlayerGameScores { + blocks + points + steals + assists + minutes + rebounds + turnovers + plusMinus + flagrantFouls + personalFouls + technicalFouls + twoPointsMade + blockedAttempts + fieldGoalsMade + freeThrowsMade + threePointsMade + defensiveRebounds + offensiveRebounds + pointsOffTurnovers + twoPointsAttempted + assistTurnoverRatio + fieldGoalsAttempted + freeThrowsAttempted + twoPointsPercentage + fieldGoalsPercentage + freeThrowsPercentage + threePointsAttempted + threePointsPercentage + playerPosition + } + statsPlayerSeasonAverageScores { + minutes + blocks + points + steals + assists + rebounds + turnovers + plusMinus + flagrantFouls + personalFouls + technicalFouls + twoPointsMade + blockedAttempts + fieldGoalsMade + freeThrowsMade + threePointsMade + defensiveRebounds + offensiveRebounds + pointsOffTurnovers + twoPointsAttempted + assistTurnoverRatio + fieldGoalsAttempted + freeThrowsAttempted + twoPointsPercentage + fieldGoalsPercentage + freeThrowsPercentage + threePointsAttempted + threePointsPercentage + efficiency + true_shooting_attempts + points_in_paint_made + points_in_paint_attempted + points_in_paint + fouls_drawn + offensive_fouls + fast_break_points + fast_break_points_attempted + fast_break_points_made + second_chance_points + second_chance_points_attempted + second_chance_points_made + } + tags { + id + name + title + visible + hardcourt + level + } + } + flowId + flowSerialNumber + price + forSale + listingOrderID + owner { + dapperID + email + flowAddress + username + profileImageUrl + twitterHandle + segmentID + } + assetPathPrefix + setPlay { + id: ID + setID + playID + flowRetired + circulationCount + tags { + id + name + title + visible + hardcourt + level + } + } + createdAt + acquiredAt + packListingID + tags { + id + name + title + visible + hardcourt + level + } + } + } + }' AS query