Hi all How do I access a resource used by an asset from the dagster #ask-community

Hi all. How do I access a resource used by an asse...

Noud Jaspers

05/25/2023, 3:19 PM

Hi all. How do I access a resource used by an asset from the IOmanager? I've specified a

ResourceDependency

in the IOmanager's class but it says that it is of type None.

owen

05/25/2023, 4:23 PM

hi @Noud Jaspers! do you mind sharing the code you're using?

Noud Jaspers

05/25/2023, 4:31 PM

Copy code

class PartitionedParquetObjectGCSIOManager(ConfigurableIOManager):
    postgres: ResourceDependency[PostgresResource]

    def load_input(self, context: InputContext) -> Any:
        ...

    def handle_output(self, context: OutputContext, tables: Any) -> None:
        ...


@asset()
def polaris(postgres: PostgresResource):
    return {k: postgres.get_table_data(k) for k in postgres.get_tables()}


defs = Definitions(
    assets=[polaris],
    resources={
        "io_manager": PartitionedParquetObjectGCSIOManager(),
        "postgres": PostgresResource(
            connection_string=get_connection_string(config["polaris_db"])
        ),
    },
)

Noud Jaspers

05/25/2023, 4:33 PM

@owen sure! See the message above. When I try to access postgres in the IOmanager it's None. Any idea what I am doing wrong? Thanks already.

Noud Jaspers

05/25/2023, 5:38 PM

I took this example and adapted it a little bit, see the code here. Same error, gcs is None

Copy code

from dagster import (
    Definitions,
    asset,
    ConfigurableResource,
    ResourceDependency,
    ConfigurableIOManager,
    InputContext,
    OutputContext,
)
from typing import Any


class PostgresResource(ConfigurableResource):
    project: str

    def get_test():
        return [1, 2, 3, 45]


class PartitionedParquetObjectGCSIOManager(ConfigurableIOManager):
    gcs: ResourceDependency[PostgresResource]

    def load_input(self, context: InputContext) -> Any:
        ...

    def handle_output(self, context: OutputContext, tables: Any) -> None:
        print(self.gcs.get_test())


@asset
def asset1():
    # create df ...
    return [1, 2, 3, 4, 5, 6]


@asset
def asset2(asset1):
    return asset1[:5]


defs = Definitions(
    assets=[asset1, asset2],
    resources={
        "io_manager": PartitionedParquetObjectGCSIOManager(),
        "gcs": PostgresResource(project="my-cool-project"),
    },
)

Noud Jaspers

05/25/2023, 5:41 PM

The full error is here:

Copy code

dagster._core.errors.DagsterExecutionHandleOutputError: Error occurred while handling output "result" of step "asset1":

  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 262, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 375, in core_dagster_event_sequence_for_step
    for evt in _type_check_and_store_output(step_context, user_event):
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 428, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output):
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 603, in _store_output
    for elt in iterate_with_context(
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 443, in iterate_with_context
    with context_fn():
  File "/usr/lib/python3.10/contextlib.py", line 153, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 84, in op_execution_error_boundary
    raise error_cls(

The above exception was caused by the following exception:
AttributeError: 'NoneType' object has no attribute 'get_test'

  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 54, in op_execution_error_boundary
    yield
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 445, in iterate_with_context
    next_output = next(iterator)
  File "/home/noud/Documents/Projects/parthenon/.venv/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 593, in _gen_fn
    gen_output = output_manager.handle_output(output_context, output.value)
  File "/home/noud/Documents/Projects/parthenon/dagster/prod/prod/__init__.py", line 54, in handle_output
    print(self.gcs.get_test())

👀 1

ben

05/26/2023, 11:46 AM

Hi Noud, the IO manager in this case needs to be explicitly provided with the resource in question (as a param when constructed)

ben

05/26/2023, 11:47 AM

The following should work:

Copy code

gcs = PostgresResource(project="my-cool-project")

defs = Definitions(
    assets=[asset1, asset2],
    resources={
        "io_manager": PartitionedParquetObjectGCSIOManager(gcs=gcs),
        "gcs": gcs
    },
)

ben

05/26/2023, 11:48 AM

At present these dependencies need to be explicit. There’s a world where we might be able to enable something like this instead if that would substantially improve your experience:

Copy code

defs = Definitions(
    assets=[asset1, asset2],
    resources={
        "io_manager": PartitionedParquetObjectGCSIOManager(gcs=ResourceByKey("gcs")),
        "gcs": PostgresResource(project="my-cool-project")
    },
)

Noud Jaspers

05/26/2023, 2:40 PM

Hey @ben thanks!! I'll provide it explicitly, which is fine!

13 Views

Open in Slack

Previous Next