Use AssetExecutionContext to correctly pass in StravaAPIResource

[jairus-m]

Summary

I created a Dagster resource to abstract the Strava API in #5. However, it looks like I didn't implement its use correctly. This PR updates the implementation so that Dagster actually recognizes its use as a resource

Details

Before:

• The StravaAPIResource I built it is not being recognized as a utilized asset

• This is because I literally used it as a class object rather than using it as a Dagster resource through the AssetExecutionContext after passing the resource into the Definitions object.

  from ..resources import StravaAPIResource, strava_api_resouce
  
  @asset(key=["strava", "activities"], group_name="dltHub")
  def load_strava_activities():
      """
      dlt EL pipeline based off declarative Rest API Config
      to load raw Strava activities into DuckDB
      """
      duckdb_database_path = EnvVar("DUCKDB_DATABASE").get_value()
      logger.info(f"Dagster Env: {EnvVar('DAGSTER_ENVIRONMENT').get_value()}")
      logger.info(f"Writing to {duckdb_database_path}..")
      pipeline = dlt.pipeline(
          pipeline_name="strava_rest_config",
          destination=dlt.destinations.duckdb(duckdb_database_path),
          dataset_name="activities",
          progress="log",
      )
  
      source = strava_rest_api_config(strava_api_resouce) # this works but is not recognized by Dagster
  
      load_info = pipeline.run(source)
      logger.info(load_info)

Note below that 'strava' has 0 uses.

After

• The StravaAPIResource is now being recognized

• AssetExecutionContext allows an asset to use any resource that is defined in the Definitions object and call it through context.resource

  from dagster import AssetExecutionContext
  
  @asset(
  key=["strava", "activities"],
  group_name="dltHub",
  required_resource_keys={"strava"},
  )
  def load_strava_activities(context: AssetExecutionContext):
      """
      dlt EL pipeline based off declarative Rest API Config
      to load raw Strava activities into DuckDB
      """
      duckdb_database_path = EnvVar("DUCKDB_DATABASE").get_value()
      logger.info(f"Dagster Env: {EnvVar('DAGSTER_ENVIRONMENT').get_value()}")
      logger.info(f"Writing to {duckdb_database_path}..")
      pipeline = dlt.pipeline(
          pipeline_name="strava_rest_config",
          destination=dlt.destinations.duckdb(duckdb_database_path),
          dataset_name="activities",
          progress="log",
      )
  
      source = strava_rest_api_config(context.resources.strava) # this can now be recognized 
  
      load_info = pipeline.run(source)
      logger.info(load_info)

Note below that 'strava' has 2 uses.

• 1 asset, 1 job