@Scout i have a lot of assets and want to log the execution time, name of the assets, their partition keys and if the materialization failed or not. how could i achieve this?

To log the execution time, name of the assets, their partition keys, and whether the materialization failed or not, you can follow these steps: 1. Logging Execution Time: You can log the execution time of the step that materialized a particular asset by using the

DagsterInstance

to retrieve run step statistics. Here's an example function that calculates the duration of a step that materialized an asset:

python     from dagster import DagsterInstance, AssetKey      def get_asset_step_duration_seconds(run_id: str, asset_key: AssetKey, instance: DagsterInstance) -> float:         run_step_stats = instance.get_run_step_stats(run_id)         for step_stats in run_step_stats:             if asset_key in [                 event.asset_materialization.asset_key for event in step_stats.materialization_events             ]:                 return step_stats.end_time - step_stats.start_time          raise ValueError(f"None of the steps in run {run_id} materialized asset {asset_key}")

You would need to call this function for each asset and pass the appropriate

run_id

,

asset_key

, and

instance

. 2. Logging Asset Name and Partition Keys: When defining your assets, you can include logging within the asset computation itself. For example:

python     from dagster import asset      @asset     def my_asset(context):         # Log the asset name         <http://context.log.info|context.log.info>(f"Asset name: {context.asset_key.path[-1]}")         # Log the partition key if it exists         if context.has_partition_key:             <http://context.log.info|context.log.info>(f"Partition key: {context.partition_key}")         # Perform your computation here

3. Logging Materialization Failures: To log materialization failures, you can use the

Failure

event within your asset computation. If an exception occurs or a condition for failure is met, you can raise a

Failure

event:

python     from dagster import asset, Failure      @asset     def my_asset(context):         try:             # Perform your computation here             pass         except Exception as e:             # Log the failure and raise a Failure event             context.log.error(f"Materialization failed for asset: {context.asset_key.path[-1]}")             raise Failure(description=str(e))

4. Using Sensors or Hooks: You can also use sensors or hooks to perform additional logging or actions based on the materialization events. Sensors can be used to trigger alerts or additional processes based on the materialization status of assets. Remember to replace the placeholders and example code with your actual asset definitions and logic. The provided examples are based on the documentation and should be adapted to fit your specific use case and environment.