https://dagster.io/ logo
#ask-community
Title
# ask-community
a

Akira Renbokoji

03/01/2024, 5:20 PM
Copy code
my_list = {
    "big_list": {
        "list_a": content_a,
        "list_b": content_b,
    },
    "small_list": {
        "list_c": content_c,
        "list_d": content_d,
    }
}

@graph
def foo(my_list):
    for list_content in my_list.values():
        for content_name, content in list_content.items():
            stuff = job.do_stuff(content_name, content)
How do I pass a dictionary to a graph? Should I make it an @op function that returns the dictionary then pass that in as an argument to the graph? Update: I forgot that users shouldn't be trying to pass dictionaries to graphs or having @op call another @op Instead have the @op return data and have @graph pass that data to another @op:
Copy code
from some_file_with_op import job

@graph()
def foo():
    some_dict = job.build_dict()
    list_of_dicts = job.write_data_return_list(some_dict)
    job.run_app(list_of_dicts)
i

Ismael Rodrigues

03/01/2024, 5:31 PM
Hey, try #ask-ai
a

Akira Renbokoji

03/01/2024, 5:42 PM
Didn't help. I'll try rephrasing my question.
dagster spin 1
z

Zach

03/01/2024, 6:45 PM
I'd probably just expose it as a Config class for the op that takes the dictionary as input
a

Akira Renbokoji

03/01/2024, 6:46 PM
Hmm, I have that part done now.
I saw the console log of the dictionary after it was called from within the graph. I'm having trouble iterating through the dictionary since it's not a dictionary to the graph.
z

Zach

03/01/2024, 6:47 PM
Are you still trying to do this?
Copy code
@graph
def foo(my_list):
    for list_content in my_list.values():
        for content_name, content in list_content.items():
            stuff = job.do_stuff(content_name, content)
a

Akira Renbokoji

03/01/2024, 6:47 PM
AttributeError: 'InvokedNodeOutputHandle' object has no attribute 'items'
Yes
z

Zach

03/01/2024, 6:48 PM
Because that won't work.
@graph
and
@job
are executed at the time your code is loaded. They should only contain ops that are being executed and passing the ouputs from ops to each other
a

Akira Renbokoji

03/01/2024, 6:48 PM
ah ok, so maybe run_config is the way to go?
z

Zach

03/01/2024, 6:48 PM
configuration like your dictionary has to be injected via op config
yes
a

Akira Renbokoji

03/01/2024, 6:48 PM
but just config won't work even if i pass it in as op?
wait, let me see if i can find the code
z

Zach

03/01/2024, 6:49 PM
It'd be like
Copy code
from dagster import Config

class ListConfig(Config):
  list_a: List[str]
  list_b: List[str]
  ...

@op
def some_op_that_needs_list_configs(context: OpExecutionContext, config: ListConfig):
  context.log(f"This is list_a: {config.list_a}")
a

Akira Renbokoji

03/01/2024, 6:50 PM
I have something like this:
Copy code
my_graph.to_job(
    config={
        "ops": {
            "load_dict": {
                "config": {
                    "my_dict": my_list
                }
            }
        }
    }
)
my_graph.to_job(config= ... @op def load_dict(context): context.log.info(my_dict0 Something like this works
z

Zach

03/01/2024, 6:51 PM
Then you just need a config schema / config object in the op interface
a

Akira Renbokoji

03/01/2024, 6:51 PM
it's this part that doesn't work because the @op output is not a dict
@graph def my_graph(): for k, v in my_dict.items():
z

Zach

03/01/2024, 6:51 PM
Are you defining a config schema anywhere for your op?
i

Ismael Rodrigues

03/01/2024, 6:52 PM
The OP returns an OutputSomething, so you can't deal with it like a normal list of things
a

Akira Renbokoji

03/01/2024, 6:52 PM
I don't think I am. Let me double check.
i

Ismael Rodrigues

03/01/2024, 6:52 PM
You can't access things inside a graph definition, you only can do that inside OPs
z

Zach

03/01/2024, 6:52 PM
Please read through this
a

Akira Renbokoji

03/01/2024, 6:52 PM
It looks like I did setup a config schema
one moment
i

Ismael Rodrigues

03/01/2024, 6:53 PM
if you need a list of something you should take a look on config-schema with Pythonic classes and DynamicOut for OPs
a

Akira Renbokoji

03/01/2024, 6:53 PM
Copy code
@op(config_schema={"my_dict": Field(dict)})
def load_dict(context):
    stuff = context.op_config["my_dict"]
    <http://context.log.info|context.log.info>(my_dict)
    return my_dict
z

Zach

03/01/2024, 6:53 PM
Yeah here's another relevant doc for what you're trying to do - https://docs.dagster.io/concepts/ops-jobs-graphs/graphs#using-dynamic-outputs
a

Akira Renbokoji

03/01/2024, 6:55 PM
bah, I tried dependencies earlier. I wonder what failed.
AI sent me down a deep rabbit hole that loops
I think AI wants me to really use
Copy code
my_job.execute_in_process()
to set up run_config
z

Zach

03/01/2024, 6:58 PM
Yeah sometimes I feel like AI actually makes things harder for beginners / people new to the concepts, because you don't know whether what the AI is telling you actually makes sense
You can definitely run your job like that and pass run config to it. Or you can launch a UI and pass run config to your job through the launchpad
a

Akira Renbokoji

03/01/2024, 7:00 PM
Thanks. I'll be back in a hour or two. I'll first try the
Copy code
some_job = my_graph.to_job()
some_job.execute_in_process(
    run_config=RunConfig(
        ...
    )
)
i

Ismael Rodrigues

03/01/2024, 7:00 PM
I don't know what's your use-case to use graph decorator, but I would do this steps easily with OPs, JOBs and DynamicOut, specially because you're using for loops
Copy code
class MyConfigClass(Config):
   your_dict: dict

@op(out=DynamicOut())
def my_op_dynamic_out(context, config: MyConfigClass):
   for v, k in config.your_dict:
      yield DynamicOutput(v, mapping_key=k)

@op
def my_op_do_stuff(context, params: dict):
    process stuff

@job
def my_job():
   my_op_dynamic_out(my_op_do_stuff)

OR

@job
def my_job():
  def _for_each(param):
      my_op_do_stuff(param)
  my_op_dynamic_out().map(_for_each)
And you can always use the lambda operator on map() to pass whenever param you want
For me, it's the easist way to deal with list of stuff to process separatedely
Also, the "execute_in_process" is not required, it's just to execute your code some way, it's basically the same as you running dagster dev and lauching your job on the Launchpad
a

Akira Renbokoji

03/01/2024, 7:40 PM
I see. I'll set the list as my default config then.
After reading the code I ended up just doing what I normally do.
Copy code
@graph()
def foo():
    some_dict = job.build_dict()
    list_of_dicts = job.write_data_return_list(some_dict)
    job.run_app(list_of_dicts)
I find it cleaner but I'm not sure if this is best practice and if I should figure out how to pass a dict to @graph directly.
z

Zach

03/01/2024, 10:41 PM
What is the
job
object?
a

Akira Renbokoji

03/01/2024, 10:44 PM
Good question, it's just a python file with
@op
in it.
z

Zach

03/01/2024, 10:45 PM
Ah cool. Yeah that looks pretty much how it should be done, as long as all the
job.*
operations you're calling are ops.
🎉 1