Dagster-Slack: Slack message decorator for the successful materialisation of a Pandas DataFrame.

[CHRISCARLON]

Summary

The existing methods in the Dagster-Slack package weren't sufficient for my use case.

What I needed to do:

Send a formatted message containing a basic overview of a Pandas DataFrame to a specified Slack channel upon the successful materialisation of an asset.

The formatted message needs to look like this...

National EV Charge Point Data successfully processed and stored in Silver Bucket.
DataFrame Info:
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 25044 entries, 0 to 25043
Data columns (total 36 columns):
 #   Column                    Non-Null Count  Dtype 
---  ------                    --------------  ----- 
 0   chargedeviceref           25044 non-null  object
 1   chargedevicename          25044 non-null  object
 2   rated_output_kw           25044 non-null  object
 3   connector_type            17252 non-null  object
 4   devicenetworks            25044 non-null  object
 5   chargedevicemanufacturer  16334 non-null  object
 6   chargedevicemodel         4655 non-null   object
 7   publishstatusid           25044 non-null  int64 
 8   datecreated               25044 non-null  object
 9   dateupdated               25041 non-null  object
 10  deviceaccess              220 non-null    object
 11  chargedevicestatus        25044 non-null  object
 12  publishstatus             25044 non-null  object
 13  devicevalidated           25044 non-null  int64 
 14  recordmoderated           25044 non-null  object
 15  recordlastupdated         3242 non-null   object
 16  recordlastupdatedby       3242 non-null   object
 17  paymentrequiredflag       25044 non-null  object
 18  paymentdetails            12202 non-null  object
 19  subscriptionrequiredflag  25044 non-null  object
 20  subscriptiondetails       8068 non-null   object
 21  parkingfeesflag           25044 non-null  object
 22  parkingfeesdetails        4072 non-null   object
 23  parkingfeesurl            1237 non-null   object
 24  accessrestrictionflag     25044 non-null  object
 25  accessrestrictiondetails  2587 non-null   object
 26  physicalrestrictionflag   25044 non-null  object
 27  physicalrestrictiontext   2268 non-null   object
 28  onstreetflag              25044 non-null  object
 29  locationtype              25044 non-null  object
 30  bearing                   53 non-null     object
 31  accessible24hours         25044 non-null  object
 32  latitude                  25044 non-null  object
 33  longitude                 25044 non-null  object
 34  geo_point                 25044 non-null  object
 35  what3words                25044 non-null  object
dtypes: int64(2), object(34)
memory usage: 6.9+ MB

Proposed Feature

Is there a way to create a decorator function to achieve the following:

Send a formatted message containing a basic overview of a Pandas DataFrame to a specified Slack channel upon the successful materialisation of an asset.

I think I need to access the AssetExecutionContext directly and couldn't seem to find a way to access the data I needed with existing HookContext methods.

As a result, I created a workaround that works pretty well so far.

Would it be possible to formally integrate something like this into the Dagster-Slack codebase?

Something that's better than my hacky workaround.

Workaround:

import pandas as pd
from functools import wraps
import io

def send_slack_silver_success_message(context, df, asset_name):
    """
    Function to send a message to Slack following the successful completion of an Asset.

    Sends formatted string containing df.info information.
    
    Leverages the AssetExcecuitonContext to access data directly - in this case a Pandas DataFrame.

    Args:
        context - AssetExecutionContext
        df
        name of the asset

    """
    # Capture DataFrame info in a string buffer
    buffer = io.StringIO()
    df.info(buf=buffer)
    df_info = buffer.getvalue()

    # call the postMessage method. 
    context.resources.slack.get_client().chat_postMessage(
        channel="XXXXX",
        text=f"{asset_name} successfully processed and stored in XXXXX.\n"
             f"Data Overview:\n```\n{df_info}\n```"
    )

def with_slack_notification(asset_name):
    """
    Wrapper to create a slack message decorator for an asset.

    This will only send a message if the return is successful.

    Otherwise the function completes and no message is sent.

    For example (not full code snippet):

    import pandas as pd
    from dagster import asset, AssetIn, AssetExecutionContext
    from ...utils.slack_messages.slack_message import with_slack_notification

        @asset(
            group_name="energy_assets",
            io_manager_key="DeltaLake",
            metadata={"mode": "overwrite"},
            ins={"entsog_gas_uk_data_bronze": AssetIn("entsog_gas_uk_data_bronze")},
            required_resource_keys={"slack"}
        )
        @with_slack_notification("ENTSOG Gas UK data")
        def entsog_gas_uk_data_silver(context: AssetExecutionContext, entsog_gas_uk_data_bronze):
            etc...

    """
    def slack_df_success_decorator(func):
        @wraps(func)
        def wrapper(context, *args, **kwargs):
            result = func(context, *args, **kwargs)
            if isinstance(result, pd.DataFrame):
                send_slack_silver_success_message(context, result, asset_name)
            return result
        return wrapper
    return slack_df_success_decorator

Benefits of the Proposed Feature

In my case, the Pandas DataFrame ends up being processed into a DeltaLake table in a Silver Data Bucket (Medallion Data Lake Structure) for use in downstream data products.

I think integrating a feature like this would be great for Data Engineers who are looking for better/improved ways to communicate with Data Analysts - in particular for communicating what data Analysts can expect to receive further downstream for dashboard data products etc.

Many thanks,
Chris