Jordan Wolinsky
07/17/2023, 11:56 PMfrom typing import Optional
from dagster import Nothing, PythonObjectDagsterType, RunRequest, graph, In, op, sensor
from pathlib import Path
DagsterConfigurablePathType = PythonObjectDagsterType(python_type=(Path, Nothing))
@op()
def no_op(foo: Optional[Path]) -> None:
print(foo)
@graph(ins={"foo": In(DagsterConfigurablePathType)})
def bar(foo: Optional[Path]) -> None:
no_op(foo)
job = bar.to_job(name="job")
@sensor(job_name="job")
def foobar_sensor():
# Following fails due to an error thrown here. This is on 1.2.7 so maybe it was fixed since then.
# <https://github.com/dagster-io/dagster/blob/1.2.7/python_modules/dagster/dagster/_core/execution/plan/plan.py#L437-L445>
# DagsterConfigurablePathType.is_nothing evaluates to False
# this is called downstream which causes the error to be raised.
yield RunRequest(run_key="1", run_config={"inputs": {}})
claire
07/18/2023, 8:50 PMjob = bar.to_job(name="job", input_values={"foo": None})
Another option would be to specify these input values through config. Since you have a custom input type, you may have to define a default value on the input, and override those defaults via config.