This is what I did. I created a

@job

first:

@job
def fetch_and_upload_mfrs_job():
    upload_mfrs_to_snowflake(fetch_manufacturers())

Why I created a job? Well, I tried a few different things and this is the only way I was able to make this pipeline work because upload_mfrs_to_snowflake has configuration, I couldn't find a way to run or execute it otherwise. Then defined my @repository like so:

@repository
def nhtsa_repo():
    return [
        fetch_manufacturers,
        upload_mfrs_to_snowflake,
        define_asset_job(
            name="fetch_and_upload_mfrs_job",
            selection=["fetch_manufacturers", "upload_mfrs_to_snowflake"],
            config={
                "ops": {
                    "upload_mfrs_to_snowflake": {
                        "config": {
                            'sf_account': SF_ACCOUNT,
                            'sf_warehouse': SF_WAREHOUSE,
                            'sf_database': SF_DATABASE,
                            'sf_schema': 'nhtsa',
                            'sf_role': SF_ROLE,
                            'sf_authenticator': SF_AUTHENTICATOR,
                            'sf_username': SF_USERNAME,
                            'sf_password': SF_PASSWORD,
                            'table_name': 'manufacturers',
                        }
                    }
                }
            }
        )
    ]

So the above works as intended. But I feel like there must be another way where I didn't need to define a job first and then have to utilize the

define_asset_job()

function. Furthermore, I also have two artifacts in dagit: the "fetch_and_upload_mfrs_job" job and a "nhtsa_mfr" asset group. Not sure why I need or have these 2 artifacts since it seems redundant, but I guess that is because of my repository definition. I feel like the above is verbose. I was thinking I could instead do something along the lines of:

@repository
def nhtsa_repo():
    return [
        fetch_manufacturers,
        upload_mfrs_to_snowflake.op.configured(
            "ops": {
                "upload_mfrs_to_snowflake": {
                    "config": {
                        'sf_account': SF_ACCOUNT,
                        'sf_warehouse': SF_WAREHOUSE,
                        'sf_database': SF_DATABASE,
                        'sf_schema': 'nhtsa',
                        'sf_role': SF_ROLE,
                        'sf_authenticator': SF_AUTHENTICATOR,
                        'sf_username': SF_USERNAME,
                        'sf_password': SF_PASSWORD,
                        'table_name': 'manufacturers',
                    }
                }
            }
        )
    ]

But I got an error as expected as it isn't a return type allowed for repository definition:

Bad return value from repository construction function: all elements of list must be of type JobDefinition, GraphDefinition, PipelineDefinition, PartitionSetDefinition, ScheduleDefinition, SensorDefinition, AssetsDefinition, or SourceAsset.Got value of type <class 'dagster.core.definitions.solid_definition.SolidDefinition'> at index 1.

So I would appreciate it if someone can point me to a less verbose way to do this. Thanks in advance!

Oooops nevermind! I realized the

name=

argument in

define_asset_job()

refers to just an arbitrary name for the job, not a pre-existing job 🤭