[dagster-airlift][jobs 4/n] Definitions.execute_job_in_process

[dpeng817]

Summary & Motivation

Adds a new internal execute_in_process method to Definitions class which contains full repository context. This ends up being really useful for the "monitoring job", which requires full repo context to build its mapping of items.

How I Tested These Changes

Additional execute_in_process test.

[dpeng817]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• [dagster-airlift][jobs 4/n] Definitions.execute_job_in_process #29243 : 2 dependent PRs (#29209 , #29277 ) 👈 (View in Graphite)
• [dagster-airlift][jobs 3/n] make AirflowDefinitionsData aware of jobs #29208
• [dagster-airlift][jobs 2/n] Job builder methods #29207
• [dagster-airlift][jobs 1/n] move assets def construction to common code path; other refactors #29206
• [dagster-airlift][datasets 3/3] dataset filtering #29164
• [dagster-airlift][datasets 2/3] auto-construct assets from datasets #29162
• [dagster-airlift][datasets 1/3] datasets storage layer, AirflowInstance retrieval methods #29155
• [dagster-airlift] Move around classes #29154
• [dagster-airlift] fix prop on AirflowDefinitionsData #29161
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

[OwenKephart]

where does this actually get used upstack? I'm having trouble understanding the precise usecase here -- is it purely for test use cases? If so, I'd prefer to "hide" this code a bit more (i.e. move as much of it as possible into the test code, and have this be a utility function rather than a method on Definitions itself)

[dpeng817]

@OwenKephart purely for test. See the upstack monitoring job tests for how it's used - but the idea is that sometimes you want access to the full repo in scope when executing a job in process.

Fine with making it a utility function; but I'll make that change in an upstack PR to avoid refactor pain.

[OwenKephart]

sounds good fixing upstack, would really prefer stuff like RepoBackedJob to also move into test-land when you do that

[graphite-app]

The line asset_selection=frozenset(asset_selection) will raise a TypeError when asset_selection is None, which is a valid input according to the function signature. Since both this method and core_execute_in_process accept asset_selection as an Optional parameter, this should be changed to:

asset_selection=frozenset(asset_selection) if asset_selection else None

This matches the pattern used in JobDefinition.get_subset on line 743 and will properly handle the None case.

Suggested change

	asset_selection=frozenset(asset_selection),
	asset_selection=frozenset(asset_selection) if asset_selection else None,

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.