https://dagster.io/ logo
Join Slack
Powered by
```my_list = { "big_list": { "list_a":...
# ask-community
a
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
Hey, try #ask-ai
a
Didn't help. I'll try rephrasing my question.
dagster spin 1
z
I'd probably just expose it as a Config class for the op that takes the dictionary as input
a
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
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
AttributeError: 'InvokedNodeOutputHandle' object has no attribute 'items'
Yes
z
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
ah ok, so maybe run_config is the way to go?
z
configuration like your dictionary has to be injected via op config
yes
a
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
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
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
Then you just need a config schema / config object in the op interface
a
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
Are you defining a config schema anywhere for your op?
i
The OP returns an OutputSomething, so you can't deal with it like a normal list of things
a
I don't think I am. Let me double check.
i
You can't access things inside a graph definition, you only can do that inside OPs
z
Please read through this
a
It looks like I did setup a config schema
one moment
i
if you need a list of something you should take a look on config-schema with Pythonic classes and DynamicOut for OPs
a
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
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
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
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
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
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
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
What is the 
job
object?
a
Good question, it's just a python file with 
@op
in it.
z
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
Open in Slack
PreviousNext
Powered by