hi all! I'm trying to create jobs with logic conditional on an environment variable. I am creating my jobs from a graph (there are multiple ones with the same logic). Essentially I don't want the last step (

dbt_docs_generation_op

) to run if i'm running dagster locally. this didn't seem to work - can someone point me in the right direction?

# base graph for running dbt transformations.
# this will run transformations, run dbt test, and deploy the dbt documentation
@graph
def run_dbt_transformations():
    env = os.getenv('ENVIRONMENT')
    if env == Env.DEV:
        test_dbt_models(after=run_dbt_models())
    else:
        dbt_docs_generation_op(after=test_dbt_models(after=run_dbt_models()))


# dagster jobs for specific model groups. they use the same dagster graph but with different dagster-dbt resources
dbt_run_assist_models = run_dbt_transformations.to_job(resource_defs={"dbt_resource": dbt_resource_assist}, tags=DBT_DOCS_K8S_CONFIG, name="dbt_run_assist_models")

my DBT_DOCS_K8S_CONFIG:

DBT_DOCS_K8S_CONFIG = {
    "dagster-k8s/config": {
        "pod_spec_config": {
            "serviceAccountName": "dbt-data-transformation",
            "volumes": [{"name": "config-secrets", "secret": {"secret_name": "github-app-secret"}}]
        },
        "container_config": {
            "resources": {
                "requests": {"cpu": "250m", "memory": "64Mi"},
                "limits": {"cpu": "500m", "memory": "2560Mi"},
            },
            "volume_mounts": [
                {"name": "config-secrets", "mount_path": "/etc/secrets"}
            ],
            "envFrom": [ # THIS IS WHERE ENVIRONMENT IS PASSED IN 
                {
                    "configMapRef": {
                        "name": "dbt-custom-run-env"
                    }
                }
            ]
        }
    }
}

cc @owen

hi @Salina Wu! When the shape of your graph changes between different environments, this can definitely cause some unexpected issues (although I'm not sure exactly what is causing your example to break). A simpler way to model this would be to just have the dbt_docs_generation_op to do nothing in the case that you're in your dev environment, so something like:

@op(required_resource_keys={"dbt"}, ins={"start_after": In(Nothing)})
def dbt_docs_generation_op(context):
    if env != Env.DEV:
        context.resources.dbt.generate_docs()

ok, sounds good! thanks owen!

ah I see (that's probably what was going wrong with the graph solution as well). Generally, when you want to change how an op functions in different environments, you would specify that using config on the op, so something like

from dagster import Field
@op(config_schema={"should_run": Field(bool, default_value=True)}, required_resource_keys={"dbt"}, ins={"start_after": In(Nothing)})
def dbt_docs_generation_op(context):
    if context.op_config.should_run:
        context.resources.dbt.generate_docs()

then, when you're creating the dev version of this graph, you can supply a False value for this configuration option to "turn off" this op

ohhhh this is neat! this looks simpler than using sensors. can you explain what you mean by "dev version" of the graph and how to pass a False value to the op config?

ah sorry missed this notification -- when you're calling .to_job() on your graph, you can pass in config (so something like

.to_job(..., config={"ops": {"dbt_docs_generation_op": {"config": {"should_run": False}}}})

)