It seems I've hit a limitation of Dagstermill and the new API. I'm using

pytest-postgresql

to spin up a test database for each of my jobs. I can access it as a fixture in a pytest function, where I convert it to a resource like so:

def test_foo(postgresql): 
    engine = create_engine(
        f"postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}",
        echo=False,
        poolclass=NullPool,
    )
    @resource
    def test_data_warehouse(context):
        return Postgres(engine=engine)
    my_job = my_graph.to_job(resource_defs={"data_warehouse": test_data_warehouse})

This works until I added a Dagstermill solid: `dagster.core.errors.DagsterInvariantViolationError: Reconstructable target was not a function returning a job definition, or a job definition produced by a decorated function. If your job was constructed using ``GraphDefinition.to_job``, you must wrap the ``to_job`` call in a function at module scope, ie not within any other functions. To learn more, check out the docs on ``reconstructable``: https://docs.dagster.io/_apidocs/execution#dagster.reconstructable` But I cannot define the job at the module level, because the postgres engine is only available to me as a fixture. Any workarounds?

hi @Jeremy Fisher ! What you want here is build_reconstructable_pipeline (which aliases to

build_reconstructable_target

if you don't want to use the legacy name for jobs 🙂). Using that, you can define a function to produce your job at module scope, as long as you can serialize the arguments to that function using JSON. I think for your use case, you could just pass in the connection string, rather than the whole postgresql object. A simple example looks like:

# my_module/test_foo.py
import pytest
from dagster import op, graph, ResourceDefinition
from dagster.core.definitions.reconstructable import build_reconstructable_target


@pytest.fixture
def something():
    return 123


def get_my_job(val):
    @op(required_resource_keys={"xyz"})
    def foo(context):
        return <http://context.resources.xyz|context.resources.xyz>

    @graph
    def bar():
        foo()

    return bar.to_job(resource_defs={"xyz": ResourceDefinition.hardcoded_resource(val)})


def test_x(something):

    recon_job = build_reconstructable_target(
        "my_module.test_foo", "get_my_job", reconstructable_args=(something,)
    )
    # you now have a reconstructable job using a fixture

Thanks! I wouldn't have guessed that this API existed.

yeah it's pretty obscure -- we haven't even migrated the names / docs over yet :blobl-grimace:

I wonder if it makes sense to add something to this effect in the dagstermill integration?

what would you imagine that interface looking like?

I mean, in the docs

ohhh yep we definitely should

Created issue at: https://github.com/dagster-io/dagster/issues/6022

?