How to define dependencies between jobs

[AlessioV98]

Hi, as stated in the title, I'd like to achieve a behaviour in which I have 4 ingestion jobs (defined through "define_asset_job" API, all 4 scheduled at the same time and based on independent assets, and then I have a processing job that is supposed to run iff all 4 ingestion jobs above has successfully runned on schedule (could be once a day, could be once a week).
All jobs have been unit tested and work singularly, no need to further define them
What I do need to do is to comprend how this dependecy behaviour can be achieved, for example I've already tried using sensors (with @run_status_sensor decorator) but that doesn't seem to handle my case (i.e: the monitored jobs parameter in its definition seems to be an "or" clause in which if a single one of those jobs succeded than the sensor is triggered, I would like to fire the sensor if ALL 4 monitored jobs succedes as stated before)
Maybe I should define this logic in the sensor function itself but atm im a bit confused on the right approach
Thanks in advance for the support, also feel free to ask for any missing detail.

[aletheavilla]

Not sure if you still need this answered but I'll answer it anyway for the other people who find themselves here.

Within our org, we're able to do this using the @multi_asset_sensor. It's almost exactly as you described. If all the last assets in the monitored jobs successfully run, then the target job is triggered.

Something like this...

@multi_asset_sensor(
    monitored_assets=[ AssetKey(["job1", "last_key"]), AssetKey(["job2", "last_key"]) ],
    job=target_job,
)
def sample_sensor(context: MultiAssetSensorEvaluationContext) -> list:
    run_requests = []
    for (
        partition,
        materializations_by_asset,
    ) in context.latest_materialization_records_by_partition_and_asset().items():
        if set(materializations_by_asset.keys()) == set(context.asset_keys):
            run_requests.append(RunRequest(partition_key=partition))
            for asset_key, materialization in materializations_by_asset.items():
                context.advance_cursor({asset_key: materialization})
    return run_requests

Note that the @multi_asset_sensor has been deprecated in favor of Declarative Automation in the Dagster 1.9 release.