[ASSET CHECKS] Building asset checks factories and comments about asset checks documentation

[louistransfer]

While the Dagster documentation has been really helpful and clear so far for core components (assets, jobs, sensors...), I find the asset checks documentation quite confusing. Here is an example : this code sample is a complete anti-pattern as the asset check loads a csv instead of checking data on the output pandas dataframe. There are no examples of asset checks on a pandas dataframe dependency.

In the "Using a factory" section, same issue, the asset check is not checking actual asset data : https://docs.dagster.io/concepts/assets/asset-checks/define-execute-asset-checks#using-a-factory

It is performing a SQL call using a resource rather than actually checking the asset.

In my view, it seems really difficult to deploy asset checks factories in production compared to asset factories which work really well. Most of the time, using the same asset checks for multiple clients would be really useful, however I couldn't find enough documentation to understand where the asset check keys should be declared.

I have been trying this factory pattern. I am using this code in the store_a/assets/sales.py file :

"""Asset checks for null values checking."""

import pandas as pd
from dagster import AssetCheckResult, AssetCheckSeverity, AssetsDefinition, asset_check

@asset(
    io_manager_key="store_a_bq_io_manager",
    compute_kind="bigquery",
    required_resource_keys={"store_a_bq_resource"},
    key_prefix="sales",
)
def raw_sales_store_a(
    context: AssetExecutionContext, process_bronze_table_sales_store_a: pd.DataFrame
) -> pd.DataFrame:
    """Process the sales file from store_a."""
    dfs = process_bronze_table_sales_store_a
    [...]
    return dfs

def check_contains_nulls_factory(target_asset: AssetsDefinition, col: str, severity=AssetCheckSeverity.WARN):
    """Checks if a column contains null."""
    asset_name = target_asset.key[0][-1]

    @asset_check(asset=target_asset, name=f"check_{asset_name}_contains_nulls")
    def check_has_nulls(asset_to_check: pd.DataFrame):
        nb_nulls = asset_to_check[col].isna().sum()
        if nb_nulls > 0:
            passed = False
        else:
            passed = True
        return AssetCheckResult(passed=passed, severity=severity)

    return check_has_nulls


check_raw_sales_store_a_is_yesterday = check_is_yesterday_factory(raw_sales_store_a, "DATE_COL")

Then importing it in my init.py file :

from harvest_flow.dagster_etl.store_a import assets as store_a_assets
all_asset_checks = load_asset_checks_from_package_module(store_a_assets, asset_key_prefix="store_a")
all_assets = load_assets_from_package_module(store_a_assets, group_name="store_a", key_prefix="store_a"),

defs = Definitions(
    assets=with_source_code_references(all_assets),
    resources=all_resources,
    jobs=all_jobs,
    schedules=all_schedulers,
    sensors=all_sensors,
    asset_checks=all_asset_checks,
)

However I get this AssetChecksKey error :

UserWarning: Error loading repository location harvest_flow.dagster_etl:KeyError: AssetCheckKey(asset_key=AssetKey(['store_a', 'sales', 'raw_sales_store_a']), name='check_raw_sales_store_a_contains_nulls')

Should I declare this asset check somewhere in the target asset ?

Being able to perform checks upstream in our data ingestion process would be really nice in addition to dbt asset checks, which are working flawlessly.
In particular, more documentation on about how to handle this part of the factory asset would be helpful (especially the naming and handling of "asset_to_check"):

    @asset_check(asset=target_asset, name=f"check_{asset_name}_contains_nulls")
    def check_has_nulls(asset_to_check: pd.DataFrame):