Goldsky QA tables (#1822)

* Initial work in progress

* add sqlmesh

* wip

* transactions and blocks all working

* Working missing block number models and many asset fixes

* use op_tags

pyproject.toml

@@ -54,6 +54,7 @@ sqlalchemy = "^2.0.25"
 textual = "^0.52.1"
 redis = "^5.0.7"
 githubkit = "^0.11.6"
+sqlmesh = "^0.113.0"
 
 
 [tool.poetry.scripts]

warehouse/oso_dagster/assets/base.py

@@ -1,48 +1,61 @@
-from ..factories.goldsky import (
-    goldsky_asset,
-    traces_checks,
-    transactions_checks,
-    blocks_checks,
-)
+from ..factories.goldsky import goldsky_network_assets
 
-base_blocks = goldsky_asset(
-    key_prefix="base",
-    name="blocks",
-    source_name="base-blocks",
-    project_id="opensource-observer",
-    destination_table_name="base_blocks",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[blocks_checks()],
-)
+# base_blocks = goldsky_asset(
+#     key_prefix="base",
+#     name="blocks",
+#     source_name="base-blocks",
+#     project_id="opensource-observer",
+#     destination_table_name="base_blocks",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[blocks_checks()],
+#     additional_jobs=[blocks_additional_jobs()],
+# )
 
-base_transactions = goldsky_asset(
-    key_prefix="base",
-    name="transactions",
-    source_name="base-enriched_transactions",
-    project_id="opensource-observer",
-    destination_table_name="base_transactions",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    schema_overrides=[{"name": "value", "field_type": "BYTES"}],
-    checks=[transactions_checks("opensource-observer.superchain.base_blocks")],
-    deps=base_blocks.assets,
-)
+# base_transactions = goldsky_asset(
+#     key_prefix="base",
+#     name="transactions",
+#     source_name="base-enriched_transactions",
+#     project_id="opensource-observer",
+#     destination_table_name="base_transactions",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     schema_overrides=[{"name": "value", "field_type": "BYTES"}],
+#     checks=[transactions_checks("opensource-observer.superchain.base_blocks")],
+#     deps=base_blocks.assets,
+#     additional_jobs=[
+#         transactions_additional_jobs(
+#             blocks_table_fqn="opensource-observer.superchain.base_blocks"
+#         )
+#     ],
+# )
 
-base_traces = goldsky_asset(
-    key_prefix="base",
-    name="traces",
-    source_name="base-traces",
-    project_id="opensource-observer",
-    destination_table_name="base_traces",
-    working_destination_dataset_name="oso_raw_sources",
+# base_traces = goldsky_asset(
+#     key_prefix="base",
+#     name="traces",
+#     source_name="base-traces",
+#     project_id="opensource-observer",
+#     destination_table_name="base_traces",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[traces_checks("opensource-observer.superchain.base_transactions")],
+#     deps=base_transactions.assets,
+#     additional_jobs=[
+#         traces_additional_jobs(
+#             transactions_table_fqn="opensource-observer.superchain.base_transactions"
+#         )
+#     ],
+# )
+
+
+base_network = goldsky_network_assets(
+    network_name="base",
     destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[traces_checks("opensource-observer.superchain.base_transactions")],
-    deps=base_transactions.assets,
+    working_destination_dataset_name="oso_raw_sources",
 )

warehouse/oso_dagster/assets/cyber.py

@@ -0,0 +1,8 @@
+from ..factories.goldsky import goldsky_network_assets
+
+cyber_network = goldsky_network_assets(
+    network_name="cyber",
+    destination_dataset_name="superchain",
+    working_destination_dataset_name="oso_raw_sources",
+    traces_enabled=False,
+)

warehouse/oso_dagster/assets/frax.py

@@ -1,54 +1,67 @@
-from ..factories.goldsky import (
-    goldsky_asset,
-    traces_checks,
-    transactions_checks,
-    blocks_checks,
-)
+from ..factories.goldsky import goldsky_network_assets
 
-frax_blocks = goldsky_asset(
-    key_prefix="frax",
-    name="blocks",
-    source_name="frax-blocks",
-    project_id="opensource-observer",
-    destination_table_name="frax_blocks",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[blocks_checks()],
-    # uncomment the following value to test
-    # max_objects_to_load=1,
-)
+# frax_blocks = goldsky_asset(
+#     key_prefix="frax",
+#     name="blocks",
+#     source_name="frax-blocks",
+#     project_id="opensource-observer",
+#     destination_table_name="frax_blocks",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[blocks_checks()],
+#     # uncomment the following value to test
+#     # max_objects_to_load=1,
+#     additional_jobs=[blocks_additional_jobs()],
+# )
 
-frax_transactions = goldsky_asset(
-    key_prefix="frax",
-    name="transactions",
-    source_name="frax-receipt_transactions",
-    project_id="opensource-observer",
-    destination_table_name="frax_transactions",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    schema_overrides=[{"name": "value", "field_type": "BYTES"}],
-    checks=[transactions_checks("opensource-observer.superchain.frax_blocks")],
-    # uncomment the following value to test
-    # max_objects_to_load=1,
-    deps=frax_blocks.assets,
-)
+# frax_transactions = goldsky_asset(
+#     key_prefix="frax",
+#     name="transactions",
+#     source_name="frax-receipt_transactions",
+#     project_id="opensource-observer",
+#     destination_table_name="frax_transactions",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     schema_overrides=[{"name": "value", "field_type": "BYTES"}],
+#     checks=[transactions_checks("opensource-observer.superchain.frax_blocks")],
+#     # uncomment the following value to test
+#     # max_objects_to_load=1,
+#     deps=frax_blocks.assets,
+#     additional_jobs=[
+#         transactions_additional_jobs(
+#             blocks_table_fqn="opensource-observer.superchain.frax_blocks",
+#         )
+#     ],
+# )
 
-frax_traces = goldsky_asset(
-    key_prefix="frax",
-    name="traces",
-    source_name="frax-traces",
-    project_id="opensource-observer",
-    destination_table_name="frax_traces",
-    working_destination_dataset_name="oso_raw_sources",
+# frax_traces = goldsky_asset(
+#     key_prefix="frax",
+#     name="traces",
+#     source_name="frax-traces",
+#     project_id="opensource-observer",
+#     destination_table_name="frax_traces",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[traces_checks("opensource-observer.superchain.frax_transactions")],
+#     # uncomment the following value to test
+#     # max_objects_to_load=1,
+#     deps=frax_transactions.assets,
+#     additional_jobs=[
+#         traces_additional_jobs(
+#             transactions_table_fqn="opensource-observer.superchain.frax_transactions"
+#         )
+#     ],
+# )
+
+frax_network = goldsky_network_assets(
+    network_name="frax",
     destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[traces_checks("opensource-observer.superchain.frax_transactions")],
-    # uncomment the following value to test
-    # max_objects_to_load=1,
-    deps=frax_transactions.assets,
+    working_destination_dataset_name="oso_raw_sources",
+    transactions_config={"source_name": "frax-receipt_transactions"},
 )

warehouse/oso_dagster/assets/metal.py

@@ -1,49 +1,53 @@
 from ..factories.goldsky import (
-    goldsky_asset,
-    traces_checks,
-    transactions_checks,
-    blocks_checks,
+    goldsky_network_assets,
 )
 
-metal_blocks = goldsky_asset(
-    key_prefix="metal",
-    name="blocks",
-    source_name="metal-blocks",
-    project_id="opensource-observer",
-    destination_table_name="metal_blocks",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[blocks_checks()],
-)
+# metal_blocks = goldsky_asset(
+#     key_prefix="metal",
+#     name="blocks",
+#     source_name="metal-blocks",
+#     project_id="opensource-observer",
+#     destination_table_name="metal_blocks",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[blocks_checks()],
+# )
 
-metal_transactions = goldsky_asset(
-    key_prefix="metal",
-    name="transactions",
-    source_name="metal-receipt_transactions",
-    project_id="opensource-observer",
-    destination_table_name="metal_transactions",
-    working_destination_dataset_name="oso_raw_sources",
-    destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    schema_overrides=[{"name": "value", "field_type": "BYTES"}],
-    checks=[transactions_checks("opensource-observer.superchain.metal_blocks")],
-    deps=metal_blocks.assets,
-)
+# metal_transactions = goldsky_asset(
+#     key_prefix="metal",
+#     name="transactions",
+#     source_name="metal-receipt_transactions",
+#     project_id="opensource-observer",
+#     destination_table_name="metal_transactions",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     schema_overrides=[{"name": "value", "field_type": "BYTES"}],
+#     checks=[transactions_checks("opensource-observer.superchain.metal_blocks")],
+#     deps=metal_blocks.assets,
+# )
 
 
-metal_traces = goldsky_asset(
-    key_prefix="metal",
-    name="traces",
-    source_name="metal-traces",
-    project_id="opensource-observer",
-    destination_table_name="metal_traces",
-    working_destination_dataset_name="oso_raw_sources",
+# metal_traces = goldsky_asset(
+#     key_prefix="metal",
+#     name="traces",
+#     source_name="metal-traces",
+#     project_id="opensource-observer",
+#     destination_table_name="metal_traces",
+#     working_destination_dataset_name="oso_raw_sources",
+#     destination_dataset_name="superchain",
+#     partition_column_name="block_timestamp",
+#     partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
+#     checks=[traces_checks("opensource-observer.superchain.metal_transactions")],
+#     deps=metal_transactions.assets,
+# )
+
+metal_network = goldsky_network_assets(
+    network_name="metal",
     destination_dataset_name="superchain",
-    partition_column_name="block_timestamp",
-    partition_column_transform=lambda c: f"TIMESTAMP_SECONDS(`{c}`)",
-    checks=[traces_checks("opensource-observer.superchain.metal_transactions")],
-    deps=metal_transactions.assets,
+    working_destination_dataset_name="oso_raw_sources",
+    transactions_config={"source_name": "metal-receipt_transactions"},
 )