[sdk] If the pipeline is called from a method, the class is not passed into it.

[CrashCumber]

Environment

• KFP version: 2.5.0
• KFP SDK version: kfp-2.14.4
• All dependencies version:

kfp                      2.14.4
kfp-kubernetes           2.14.4
kfp-pipeline-spec        2.14.0
kfp-server-api           2.14.3

Steps to reproduce

If the pipeline is called from a method, the class is not passed into it. Create pipeline:

class Pipeline:
    @staticmethod
    @dsl.component
    def task()
        ...

    @staticmethod
    @dsl.pipeline(name='test-pipeline')
    def my_pipeline():
        t = Pipeline.task()

Compile python pipelines/test.py

Traceback (most recent call last):
  File "/Users/n/Desktop/code/kubeflow/src/pipelines/test.py", line 53, in <module>
    class Pipeline:
  File "/Users/n/Desktop/code/kubeflow/src/pipelines/test.py", line 62, in Pipeline
    def my_pipeline():
  File "/Users/n/Desktop/code/kubeflow/venv/lib/python3.10/site-packages/kfp/dsl/pipeline_context.py", line 71, in pipeline
    return component_factory.create_graph_component_from_func(
  File "/Users/n/Desktop/code/kubeflow/venv/lib/python3.10/site-packages/kfp/dsl/component_factory.py", line 712, in create_graph_component_from_func
    return graph_component.GraphComponent(
  File "/Users/n/Desktop/code/kubeflow/venv/lib/python3.10/site-packages/kfp/dsl/graph_component.py", line 61, in __init__
    pipeline_outputs = pipeline_func(*args_list)
  File "/Users/n/Desktop/code/kubeflow/src/pipelines/test.py", line 63, in my_pipeline
    t = Pipeline.task(context={})
NameError: name 'Pipeline' is not defined

Compared to regular python code, everything works.

class Pipeline:
    @staticmethod
    def task():
        ...

    @staticmethod
    def my_pipeline():
        t = Pipeline.task()

Expected result

Pipeline created. The class is visible from the method. Behavior is expected as with a regular python method, the method class gets into the visibility of the method.

---

Impacted by this bug? Give it a 👍.

[SeraphVoid]

Hey I would like to work on this issue , kindly assign it to me

[zer-art]

Hi there! I looked into this, and I believe this isn't a bug in KFP but rather an issue with Python's class scoping rules.

When @dsl.pipeline executes, the Pipeline class body is still being defined, so the name Pipeline isn't available in the namespace yet. That is why Pipeline.task() fails with a NameError.

Workaround: You can reference the component function directly within the class scope, or define the pipeline function outside of the class structure.

Example:

@staticmethod
    @dsl.pipeline(name='test-pipeline')
    def my_pipeline():
        # Use task() directly instead of Pipeline.task()
        t = task()

Hope this helps!