Need help correctly importing from other files in my dagster project

[dagsir[bot]]

Hello everyone,
I have the following job defined under jobs/master_inventory.py

from dagster import AssetSelection, define_asset_job
selection = AssetSelection.keys("report_validator").upstream()
master_inventory_list_job = define_asset_job(
    name="master_inventory_list_job",
    selection=selection,
)

I need to import the mentioned job in a sensor definition file under sensors/master_inventory_ready.py as follows:

from dagster import (sensor,
                     SensorResult,
                     AssetExecutionContext)

from ..jobs import master_inventory

@sensor(job=master_inventory)
def inventory_file_ready_sensor(context: AssetExecutionContext):

  current_state = {}
  runs_to_request = []
  
  return SensorResult(
    run_requests=runs_to_request,
  )

when I try to use the above sensor in the init.py file as follows I get a syntax error:

File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/auction_derisking/__init__.py", line 13, in <module>
    from .sensors.master_inventory_ready import inventory_file_ready_sensor
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/auction_derisking/sensors/master_inventory_ready.py", line 8, in <module>
    def inventory_file_ready_sensor(context: AssetExecutionContext):
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_core/definitions/decorators/sensor_decorator.py", line 73, in inner
    sensor_def = SensorDefinition.dagster_internal_init(
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_core/definitions/sensor_definition.py", line 631, in dagster_internal_init
    return SensorDefinition(
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_core/definitions/sensor_definition.py", line 570, in __init__
    targets = [DirectTarget(job)]
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_core/definitions/target.py", line 41, in __new__
    check.inst_param(
  File "/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_check/__init__.py", line 640, in inst_param
    raise _param_type_mismatch_exception(
dagster._check.ParameterCheckError: Param "target" is not one of ['GraphDefinition', 'JobDefinition', 'UnresolvedAssetJobDefinition']. Got <module 'auction_derisking.jobs.master_inventory' from '/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/auction_derisking/jobs/master_inventory.py'> which is type <class 'module'>.
/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/.env/lib/python3.10/site-packages/dagster/_core/workspace/context.py:616: UserWarning: Error loading repository location auction_derisking:dagster._check.ParameterCheckError: Param "target" is not one of ['GraphDefinition', 'JobDefinition', 'UnresolvedAssetJobDefinition']. Got <module 'auction_derisking.jobs.master_inventory' from '/Users/rezarazavipour/sandbox/dagster/auction-derisking-project/auction-derisking/auction_derisking/jobs/master_inventory.py'> which is type <class 'module'>.

Source code is:

from . import assets
from .jobs.master_inventory import master_inventory_list_job
from .jobs.create_auction_folder import create_auction_root_folder
from .sensors.master_inventory_ready import inventory_file_ready_sensor

my_jobs = [master_inventory_list_job, create_auction_root_folder]
all_sensors = [inventory_file_ready_sensor]

defs = Definitions(
    assets=load_assets_from_package_module(assets),
    schedules=[
        ScheduleDefinition(
            job=master_inventory_list_job,
            cron_schedule="*/3 * * * *",
            # default_status=DefaultScheduleStatus.RUNNING,
            execution_timezone = "America/Los_Angeles",
        ),
        ScheduleDefinition(
            job=create_auction_root_folder,
            cron_schedule="*/1 * * * *",
            # default_status=DefaultScheduleStatus.RUNNING,
            execution_timezone = "America/Los_Angeles",
        ),
    ],
    jobs=my_jobs,
    sensors=all_sensors,
)

I must be defining something in a non Dagster acceptable type. Any ideas?

---

The question was originally asked in Dagster Slack.

[zyd14]

It looks like your sensor definition

@sensor(job=master_inventory)
def inventory_file_ready_sensor(context: AssetExecutionContext):

is pointing to master_inventory, which looks like a module (you import a job from it in your definitions). Did you perhaps mean to point it to master_inventory_list_job? So it'd look like this:

from ..jobs.master_inventory import master_inventory_list_job
@sensor(job=master_inventory_list_job)
def inventory_file_ready_sensor(context: AssetExecutionContext):

[rrazavipour]

that is correct, thanks for the answer