@@ -1,17 +1,15 @@
 import inspect
 from functools import update_wrapper
-from typing import TYPE_CHECKING, Callable, Generator, List, Optional, Union
+from typing import TYPE_CHECKING, Callable, Generator, List, Optional, Sequence, Union
 
 from dagster import check
 from dagster.core.definitions.sensor import RunRequest, SensorDefinition, SkipReason
 from dagster.core.errors import DagsterInvariantViolationError
 
-from ....seven import funcsigs
-from ...decorator_utils import get_function_params
 from ...errors import DagsterInvariantViolationError
 from ..events import AssetKey
 from ..graph import GraphDefinition
-from ..pipeline import PipelineDefinition
+from ..job import JobDefinition
 from ..sensor import AssetSensorDefinition, RunRequest, SensorDefinition, SkipReason
 
 if TYPE_CHECKING:
import inspect
from functools import update_wrapper
from typing import TYPE_CHECKING, Callable, Generator, List, Optional, Union

from dagster import check
from dagster.core.definitions.sensor import RunRequest, SensorDefinition, SkipReason
from dagster.core.errors import DagsterInvariantViolationError

from ....seven import funcsigs
from ...decorator_utils import get_function_params
from ...errors import DagsterInvariantViolationError
from ..events import AssetKey
from ..graph import GraphDefinition
from ..pipeline import PipelineDefinition
from ..sensor import AssetSensorDefinition, RunRequest, SensorDefinition, SkipReason

if TYPE_CHECKING:
    from ..sensor import SensorEvaluationContext
    from ...events.log import EventLogEntry


def is_context_provided(params: List[funcsigs.Parameter]) -> bool:
    return len(params) == 1


def sensor(
    pipeline_name: Optional[str] = None,
    name: Optional[str] = None,
    solid_selection: Optional[List[str]] = None,
    mode: Optional[str] = None,
    minimum_interval_seconds: Optional[int] = None,
    description: Optional[str] = None,
    job: Optional[Union[PipelineDefinition, GraphDefinition]] = None,
) -> Callable[
    [
        Callable[
            ["SensorEvaluationContext"],
            Union[Generator[Union[RunRequest, SkipReason], None, None], RunRequest, SkipReason],
        ]
    ],
    SensorDefinition,
]:
    """
    Creates a sensor where the decorated function is used as the sensor's evaluation function.  The
    decorated function may:

    1. Return a `RunRequest` object.
    2. Yield multiple of `RunRequest` objects.
    3. Return or yield a `SkipReason` object, providing a descriptive message of why no runs were
       requested.
    4. Return or yield nothing (skipping without providing a reason)

    Takes a :py:class:`~dagster.SensorEvaluationContext`.

    Args:
        pipeline_name (str): Name of the target pipeline
        name (Optional[str]): The name of the sensor. Defaults to the name of the decorated
            function.
        solid_selection (Optional[List[str]]): A list of solid subselection (including single
            solid names) to execute for runs for this sensor e.g.
            ``['*some_solid+', 'other_solid']``
        mode (Optional[str]): The mode to apply when executing runs for this sensor.
            (default: 'default')
        minimum_interval_seconds (Optional[int]): The minimum number of seconds that will elapse
            between sensor evaluations.
        description (Optional[str]): A human-readable description of the sensor.
        job (Optional[PipelineDefinition]): Experimental
    """
    check.opt_str_param(name, "name")

    def inner(
        fn: Callable[
            ["SensorEvaluationContext"],
            Union[Generator[Union[SkipReason, RunRequest], None, None], SkipReason, RunRequest],
        ]
    ) -> SensorDefinition:
        check.callable_param(fn, "fn")
        sensor_name = name or fn.__name__

        def _wrapped_fn(context):
            result = fn(context) if is_context_provided(get_function_params(fn)) else fn()

            if inspect.isgenerator(result):
                for item in result:
                    yield item
            elif isinstance(result, (SkipReason, RunRequest)):
                yield result

            elif result is not None:
                raise DagsterInvariantViolationError(
                    (
                        "Error in sensor {sensor_name}: Sensor unexpectedly returned output "
                        "{result} of type {type_}.  Should only return SkipReason or "
                        "RunRequest objects."
                    ).format(sensor_name=sensor_name, result=result, type_=type(result))
                )

        sensor_def = SensorDefinition(
            name=sensor_name,
            pipeline_name=pipeline_name,
            evaluation_fn=_wrapped_fn,
            solid_selection=solid_selection,
            mode=mode,
            minimum_interval_seconds=minimum_interval_seconds,
            description=description,
            job=job,
            decorated_fn=fn,
        )

        update_wrapper(sensor_def, wrapped=fn)

        return sensor_def

    return inner


def asset_sensor(
    asset_key: AssetKey,
    pipeline_name: Optional[str] = None,
    name: Optional[str] = None,
    solid_selection: Optional[List[str]] = None,
    mode: Optional[str] = None,
    minimum_interval_seconds: Optional[int] = None,
    description: Optional[str] = None,
    job: Optional[Union[PipelineDefinition, GraphDefinition]] = None,
) -> Callable[
    [
        Callable[
            [
                "SensorEvaluationContext",
                "EventLogEntry",
            ],
            Union[Generator[Union[RunRequest, SkipReason], None, None], RunRequest, SkipReason],
        ]
    ],
    AssetSensorDefinition,
]:
    """
    Creates an asset sensor where the decorated function is used as the asset sensor's evaluation
    function.  The decorated function may:

    1. Return a `RunRequest` object.
    2. Yield multiple of `RunRequest` objects.
    3. Return or yield a `SkipReason` object, providing a descriptive message of why no runs were
       requested.
    4. Return or yield nothing (skipping without providing a reason)

    Takes a :py:class:`~dagster.SensorEvaluationContext` and an EventLogEntry corresponding to an
    AssetMaterialization event.

    Args:
        asset_key (AssetKey): The asset_key this sensor monitors.
        pipeline_name (Optional[str]): Name of the target pipeline
        name (Optional[str]): The name of the sensor. Defaults to the name of the decorated
            function.
        solid_selection (Optional[List[str]]): A list of solid subselection (including single
            solid names) to execute for runs for this sensor e.g.
            ``['*some_solid+', 'other_solid']``
        mode (Optional[str]): The mode to apply when executing runs for this sensor.
            (default: 'default')
        minimum_interval_seconds (Optional[int]): The minimum number of seconds that will elapse
            between sensor evaluations.
        description (Optional[str]): A human-readable description of the sensor.
    """

    check.opt_str_param(name, "name")

    def inner(
        fn: Callable[
            [
                "SensorEvaluationContext",
                "EventLogEntry",
            ],
            Union[Generator[Union[SkipReason, RunRequest], None, None], SkipReason, RunRequest],
        ]
    ) -> AssetSensorDefinition:
        check.callable_param(fn, "fn")
        sensor_name = name or fn.__name__

        def _wrapped_fn(context, event):
            result = fn(context, event)

            if inspect.isgenerator(result):
                for item in result:
                    yield item
            elif isinstance(result, (RunRequest, SkipReason)):
                yield result

            elif result is not None:
                raise DagsterInvariantViolationError(
                    (
                        "Error in sensor {sensor_name}: Sensor unexpectedly returned output "
                        "{result} of type {type_}.  Should only return SkipReason or "
                        "RunRequest objects."
                    ).format(sensor_name=sensor_name, result=result, type_=type(result))
                )

        return AssetSensorDefinition(
            name=sensor_name,
            asset_key=asset_key,
            pipeline_name=pipeline_name,
            asset_materialization_fn=_wrapped_fn,
            solid_selection=solid_selection,
            mode=mode,
            minimum_interval_seconds=minimum_interval_seconds,
            description=description,
            job=job,
        )

    return inner

should we asspire to remove `job` if we have `jobs` ? both isnt the worst i guess