monitor_all_repositories

False

teams_on_run_failure

from dagster import asset, Definitions, load_assets_from_current_module
from dagster_msteams import make_teams_on_run_failure_sensor
from os import getenv

LOCATION = "a"

# Assets
@asset(name=f"error_asset_{LOCATION}")
def error_asset() -> None:
    raise Exception(f"Error from location {LOCATION}")


defs = Definitions(
    assets  = load_assets_from_current_module(key_prefix=LOCATION),
    sensors=[
        make_teams_on_run_failure_sensor(
            hook_url=getenv("MS_TEAMS_WEBHOOK"),
            dagit_base_url=getenv("DAGIT_URL"),
        )
    ],
)