Hey guys. How do I access the context/catalog during a node execution?

hi @Zemeio, I'm wondering what are you trying to achieve?

@Juan Luis Allow me to “jump in” / piggy back on this question: In our case, we needed access to the

catalog

in the

pipeline.py

to create multiple pipelines dynamically, based on values specified in the parameter files. So far, I had “hacked my way around” by adding this in

pipeline.py

project_path = paths.PROJECT_ROOT_DIR.as_posix()
session = KedroSession.create(project_path=project_path)
context = session.load_context()
catalog = context.catalog

It worked just fine. However… for better “customer segregation” I have recently separated the conf for our different customers in this way:

conf/customer_{x,y,z}/{base,local}

Thanks (or because) to this,

--conf-source

is now a mandatory argument to

kedro run

(which is what we wanted) However the “hack” above (in

pipeline.py

) does not work anymore… So, instead of re-instantiating a session / context / catalog, is there away to access the “actual runtime” session / context / catalog ? Thx in advance, M In our case, this very convenient / re-assuring since

However the “hack” above (in

pipeline.py

) does not work anymore…

could you elaborate? is it something that stopped working because of a change we did in Kedro recently?

Since

conf_source

is left unspecified in

KedroSession.create

it raises an error… (i.e the desire “customer conf isolation”)

looping my colleague @Nok Lam Chan

Hey @Marc Gris, me again! Love all these questions and please keep asking! In short,

after_context_created

is a replacement for

get_current_session

, a long thread discussion happened last year. @Zemeio The

after_context_created

hook is where you can access the

context

directly. If you find that you need the catalog within the node it’s a sign that you may deviating from the Kedro way of doing things. Could you elaborate what you are trying to do @Zemeio?

My use case is similar to @Marc Gris, and I agree that it deviates a bit from the Kedro way, by making nodes a bit more dynamic.

(I’ll be to this thread, in a meeting now)

I have lots of "small variations" that can happen according to parameters, and it feels like the best way to accomplish this is by having the parameters be accessed in the node execution instead of creating hundreds of different static pipelines

Thx @Nok Lam Chan will look into the

after_context_created

hook 👍🏼 (glad to hear that my flood of questions is not annoying 😉 )

What do you need to access? Maybe it helps if you can give me a code snippets even if such things doesn’t exist. Dynamic pipeline is something that I would like to improve, I don’t know if Kedro would ever support a full dynamic execution but it will help us to design a better solution. https://github.com/kedro-org/kedro/issues/2627

In my “current” case, I need to access “current”

"params:schema.tables"

to dynamically generate the different nodes for all those tables. Will try to do so with the hook you suggested.

Preliminary results here: I added the context to my Hooks class at the ``after_context_created`` hook, and then I can change the value of the node slightly depending on context values. I have to find possible side effects now, since this is not expected by the framework. One side effect I can think of, is that the order of running the nodes is defined before this change, so that is one thing to watch out for.

before_pipeline_run

- could you share how you make the change?

@Nok Lam Chan

@hook_impl
    def before_pipeline_run(self, pipeline: Pipeline):
        multi_nodes = []
        for node in pipeline.nodes:
            if [i for i in node.inputs if "{sln}" in i] or [i for i in node.outputs if "{sln}" in i]:
                multi_nodes.append(node)
        for node in multi_nodes:
            for sln in self.SLNS:
                n = Node(
                    node.func,
                    [i.format(sln=sln) for i in node.inputs],
                    [i.format(sln=sln) for i in node.outputs],
                    name=node.name + f"_{sln}",
                    tags=node.tags,
                    namespace=node.namespace
                )
                pipeline.nodes.append(n)
            pipeline.nodes.remove(node)
        print(pipeline)

This is my before_pipeline run. I add nodes based on how many slns I need to run for (I changed a few of the names here), but the pipeline ends up being the same. Probably a copy is being created and passed to the hooks? Should I, instead, put this functionality in the session and use my own managed session?