We have a job which is comprised of dbt assets. We are using Dagster Asset Checks on these assets. In dbt you can set a specific test to be a warning. Is it possible in Dagster to send a slack alert for dbt Asset Checka with a severity of warning, but to not fail the asset materialization?

In Dagster, asset checks can be used to validate the data produced by assets, and you can configure alerts on asset checks using asset-based alert policies in Dagster Cloud. However, the documentation indicates that checks are currently only supported per-asset, not per-partition, and there are some limitations to be aware of, such as the UI being tested with a maximum of 1000 checks per asset and the expectation that most assets will have fewer than 50 checks. Regarding the specific behavior of asset checks in relation to dbt tests and the ability to send Slack alerts without failing the asset materialization, the documentation does not provide explicit details on configuring asset checks to have different severities (such as warning) that would allow the asset materialization to continue while still sending alerts. However, there is a mention of a discussion around decoupling severity from control flow, which suggests that there might be interest in the community for having more granular control over asset checks and their impact on downstream execution. This indicates that there may be future developments in this area, but as of the current state of the documentation, it does not provide a built-in way to set a dbt asset check with a severity of warning that sends a Slack alert without failing the asset materialization. For alerting, you can configure alerts on asset checks in Dagster Cloud, and there are also examples of creating sensors for sending Slack messages on run failure. If you are looking to implement custom behavior, such as sending Slack alerts for warnings, you might need to write custom logic within your asset or sensor definitions to handle this. If you have a use case that doesn't fit within the current limitations or if you need more specific behavior, you are encouraged to reach out to discuss your requirements. Additionally, you can keep an eye on the relevant GitHub issues and discussions for updates on this topic.

@Scout Is it possible to implement this functionality in Dagster without the use of Asset Checks?

Yes, it is possible to implement custom alerting functionality in Dagster without using asset checks. You can achieve this by writing custom logic within your assets or by using sensors and hooks to monitor asset materializations and send notifications based on your own criteria. Here's a high-level approach to implement such functionality: 1. Custom Asset Logic: Within your asset definitions, you can include custom validation logic that assesses the data and determines whether it meets your 'warning' criteria. Instead of failing the asset, you can emit an event or log a message that can be used to trigger a notification. 2. Sensors: You can create sensors that monitor the completion of asset materializations. These sensors can check for specific conditions or metadata in the materialization events and send alerts accordingly. 3. Hooks: Hooks in Dagster allow you to perform custom actions in response to different lifecycle events of an op or an asset. You can create a custom hook that sends a Slack message when certain conditions are met. 4. Slack Integration: For sending Slack alerts, you can use the Slack API or a Slack client library in Python to send messages from within your custom logic, sensors, or hooks. 5. External Systems: If you have an external monitoring or alerting system, you can integrate it with Dagster by emitting custom events or metrics that the external system can consume to trigger alerts. It's important to note that while you can implement custom alerting logic, this approach may require more development and maintenance effort compared to using built-in features like asset checks. Additionally, you should ensure that your custom logic does not interfere with the normal operation and materialization of assets. Remember to consider the scalability and performance implications of your custom alerting logic, especially if you are monitoring a large number of assets or if your validation logic is complex. If you need further guidance on implementing custom alerting in Dagster, you can refer to the Dagster documentation for more details on sensors, hooks, and logging, or reach out to the community for support.

@Scout can you provide an example of parsing the dbt test results so that I can alert on warning severity dbt alerts, but fail on full test failures?

As of my knowledge cutoff date in early 2023, Dagster does not provide a built-in mechanism to parse dbt test results and conditionally alert on warnings versus errors directly within the asset materialization process. However, you can implement custom logic to parse dbt test results and handle alerts accordingly. Here's a conceptual example of how you might approach this: 1. Run dbt tests separately: Execute your dbt tests separately from your asset materialization process, capturing the output of the dbt test command. 2. Parse dbt test results: Parse the output of the dbt test command to determine the severity of any test failures. dbt typically outputs test results in a machine-readable format like JSON when using the

--output

flag. 3. Custom logic for alerts: Implement custom logic that sends alerts based on the parsed test results. For example, you could send a Slack message for warnings and raise an exception for errors. 4. Integrate with Dagster: Use Dagster's sensors or hooks to integrate this custom logic with your asset materialization process. Here's a simplified example of how you might write a Python function to parse dbt test results and send Slack alerts:

python
import subprocess
import json
import requests

def run_dbt_tests_and_alert():
    # Run dbt tests and capture the output
    result = subprocess.run(['dbt', 'test', '--output', 'json'], capture_output=True)
    test_output = result.stdout.decode()

    # Parse the JSON output
    test_results = json.loads(test_output)

    # Check for warnings and errors
    warnings = []
    errors = []
    for test in test_results['results']:
        if test['status'] == 'warn':
            warnings.append(test['name'])
        elif test['status'] == 'fail':
            errors.append(test['name'])

    # Send Slack alerts for warnings
    if warnings:
        message = f"dbt test warnings: {', '.join(warnings)}"
        send_slack_alert(message)

    # Raise an exception for errors to fail the asset materialization
    if errors:
        raise Exception(f"dbt test failures: {', '.join(errors)}")

def send_slack_alert(message):
    webhook_url = 'YOUR_SLACK_WEBHOOK_URL'
    slack_data = {'text': message}
    response = <http://requests.post|requests.post>(
        webhook_url, json=slack_data,
        headers={'Content-Type': 'application/json'}
    )
    if response.status_code != 200:
        raise ValueError(f"Request to Slack returned an error {response.status_code}, the response is:\n{response.text}")

# Example usage within a Dagster asset or op
@asset
def my_asset():
    run_dbt_tests_and_alert()
    # Proceed with asset materialization if no errors
    # ...

Please note that this is a high-level example and does not include error handling or the specifics of integrating with

@Scout rewrite this so that it is implemented with the @dbt_assets decorator.

As of my knowledge cutoff date, the

@dbt_assets

decorator in Dagster is used to define a collection of assets that correspond to dbt models, tests, and snapshots. However, the

@dbt_assets

decorator itself does not provide direct functionality for parsing dbt test results and conditionally sending alerts based on those results. To implement custom logic for parsing dbt test results and sending Slack alerts within the context of

@dbt_assets

, you would need to manually handle the dbt test command execution and result parsing within the decorated function. Here's a conceptual example of how you might do this:

python
from dagster import AssetExecutionContext
from dagster_dbt import dbt_assets, DbtCliResource
import json
import requests

@dbt_assets
def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    # Run dbt tests and capture the output
    dbt_cli_invocation = dbt.cli(["test", "--output", "json"], context=context)

    # Parse the JSON output
    test_output = dbt_cli_invocation.stdout.decode()
    test_results = json.loads(test_output)

    # Check for warnings and errors
    warnings = []
    errors = []
    for test in test_results['results']:
        if test['status'] == 'warn':
            warnings.append(test['name'])
        elif test['status'] == 'fail':
            errors.append(test['name'])

    # Send Slack alerts for warnings
    if warnings:
        message = f"dbt test warnings: {', '.join(warnings)}"
        send_slack_alert(message)

    # Raise an exception for errors to fail the asset materialization
    if errors:
        raise Exception(f"dbt test failures: {', '.join(errors)}")

    # If no errors, continue to yield asset events from dbt run
    yield from dbt.cli(["run"], context=context).stream()

def send_slack_alert(message):
    webhook_url = 'YOUR_SLACK_WEBHOOK_URL'
    slack_data = {'text': message}
    response = <http://requests.post|requests.post>(
        webhook_url, json=slack_data,
        headers={'Content-Type': 'application/json'}
    )
    if response.status_code != 200:
        raise ValueError(f"Request to Slack returned an error {response.status_code}, the response is:\n{response.text}")

In this example, the

my_dbt_assets

function is decorated with

@dbt_assets

, which means it is responsible for running dbt commands and yielding the resulting asset events. The function first runs the

dbt test

command and captures the output in JSON format. It then parses the test results to identify warnings and errors. If there are any warnings, it sends a Slack alert using the

send_slack_alert

function. If there are any errors, it raises an exception to fail the asset materialization. If there are no errors, it proceeds to run the

dbt run

command and yields the resulting asset events. Please note th