dagster-io / dagster

An orchestration platform for the development, production, and observation of data assets.
https://dagster.io
Apache License 2.0
11.5k stars 1.45k forks source link

Support Union type annotations on op/asset-decorated functions #5115

Open sryza opened 3 years ago

sryza commented 3 years ago

This fails:

from typing import Union

from dagster import pipeline, solid

@solid
def my_solid(current: Union[int, float], previous: Union[int, float]):
    pass

@pipeline
def my_pipeline():
    my_solid()

With this error:

Traceback (most recent call last):
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/input.py", line 264, in _checked_inferred_type
    resolved_type = resolve_dagster_type(inferred.annotation)
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/types/dagster_type.py", line 890, in resolve_dagster_type
    dagster_type=str(dagster_type), additional_msg="."
dagster.core.errors.DagsterInvalidDefinitionError: Invalid type: dagster_type must be an instance of DagsterType or a Python type: got typing.Union[int, float].

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "try-union.py", line 11, in <module>
    def my_solid(current: Union[int, float], previous: Union[int, float]):
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/decorators/solid.py", line 247, in solid
    return _Solid()(name)
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/decorators/solid.py", line 107, in __call__
    exclude_nothing=True,
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/decorators/solid.py", line 380, in resolve_checked_solid_fn_inputs
    for inferred in inferred_props.values()
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/decorators/solid.py", line 381, in <genexpr>
    if inferred.name in inputs_to_infer
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/input.py", line 222, in create_from_inferred
    dagster_type=_checked_inferred_type(inferred),
  File "/Users/sryza/dagster/python_modules/dagster/dagster/core/definitions/input.py", line 270, in _checked_inferred_type
    ) from e
dagster.core.errors.DagsterInvalidDefinitionError: Problem using type 'typing.Union[int, float]' from type annotation for argument 'current', correct the issue or explicitly set the dagster_type on your InputDefinition.
cmpadden commented 8 months ago

@sryza - this may also be handy in having a union of possible resources for an asset. This came up recently for me in:

@asset
def asset_example(database: Union[SnowflakeResource, DuckDBResource]):
    pass
dagster._core.errors.DagsterInvalidDefinitionError: Problem using type 'typing.Union[dagster_snowflake.resources.SnowflakeResource, dagster_duckdb.resource.DuckDBResource]' from type annotation for argument 'database', correct the issue or explicitly set the dagster_type via In().