customer_summary.sql

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_spent
FROM orders
GROUP BY customer_id

defs.yml

type: dagster_open_platform.lib.sql_view_component.SqlViewComponent

attributes:
  spec:
    key: my_database/my_schema/customer_summary
    group_name: views
    kinds:
      - sql
      - view
    tags:
      layer: mart
  sql_path: customer_summary.sql

sql_view_component.py

import hashlib
from pathlib import Path

import dagster as dg


class SqlViewComponent(dg.Component, dg.Resolvable, dg.Model):
    """A component that creates a SQL view and tracks changes via code_version."""

    spec: dg.ResolvedAssetSpec
    sql_path: str

    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
        sql_file = Path(context.path) / self.sql_path
        sql_content = sql_file.read_text()
        sql_hash = hashlib.sha256(sql_content.encode()).hexdigest()[:16]

        # Replace the spec with one that includes the code_version
        spec = self.spec.replace_attributes(code_version=sql_hash)

        @dg.multi_asset(specs=[spec])
        def sql_view_asset(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
            context.log.info(f"Creating view from {sql_file.name} (hash: {sql_hash})")
            # Execute the SQL to create/replace the view
            # e.g.: connection.execute(f"CREATE OR REPLACE VIEW {view_name} AS {sql_content}")
            return dg.MaterializeResult(
                metadata={
                    "sql_hash": dg.MetadataValue.text(sql_hash),
                    "sql_path": dg.MetadataValue.path(str(sql_file)),
                },
            )

        return dg.Definitions(assets=[sql_view_asset])