Skip to content

Commit c9c35b8

Browse files
committed
chore: Source component/pipeline name from decorator params
- Prefer value from metadata.yaml but fall back to decorator param - If both of these don't exist, instead use function name Signed-off-by: Giulio Frasca <gfrasca@redhat.com>
1 parent 6ce2988 commit c9c35b8

File tree

2 files changed

+154
-1
lines changed

2 files changed

+154
-1
lines changed

scripts/generate_readme/metadata_parser.py

Lines changed: 52 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,53 @@ def _get_type_string(self, annotation: Any) -> str:
7777
# and clean up the typing module prefix
7878
return str(annotation).replace('typing.', '')
7979

80+
def _extract_decorator_name(self, decorator: ast.AST) -> Optional[str]:
81+
"""Extract the 'name' parameter from a decorator if present.
82+
83+
Args:
84+
decorator: AST node representing the decorator.
85+
86+
Returns:
87+
The name parameter value if found, None otherwise.
88+
"""
89+
# Check if decorator is a Call node (has arguments)
90+
if isinstance(decorator, ast.Call):
91+
# Look for name parameter in keyword arguments
92+
for keyword in decorator.keywords:
93+
if keyword.arg == 'name':
94+
# Extract the string value
95+
if isinstance(keyword.value, ast.Constant):
96+
return keyword.value.value
97+
return None
98+
99+
def _get_name_from_decorator_if_exists(self, function_name: str) -> Optional[str]:
100+
"""Get the decorator's name parameter for a specific function.
101+
102+
Args:
103+
function_name: Name of the function to find.
104+
105+
Returns:
106+
The name parameter from the decorator if found, None otherwise.
107+
"""
108+
try:
109+
with open(self.file_path, 'r', encoding='utf-8') as f:
110+
source = f.read()
111+
112+
tree = ast.parse(source)
113+
114+
for node in ast.walk(tree):
115+
if isinstance(node, ast.FunctionDef) and node.name == function_name:
116+
# Check decorators for name parameter
117+
for decorator in node.decorator_list:
118+
decorator_name = self._extract_decorator_name(decorator)
119+
if decorator_name:
120+
return decorator_name
121+
122+
return None
123+
except Exception as e:
124+
logger.debug(f"Could not extract decorator name from AST: {e}")
125+
return None
126+
80127
def _extract_function_metadata(self, function_name: str, module_name: str = "module") -> Dict[str, Any]:
81128
"""Extract metadata from a KFP function (component or pipeline).
82129
@@ -105,9 +152,13 @@ def _extract_function_metadata(self, function_name: str, module_name: str = "mod
105152
# Fallback to the object itself if neither attribute is available
106153
func = func_obj
107154

155+
# Try to get name from decorator, fall back to function name
156+
decorator_name = self._get_name_from_decorator_if_exists(function_name)
157+
component_name = decorator_name if decorator_name else function_name
158+
108159
# Extract basic function information
109160
metadata = {
110-
'name': function_name,
161+
'name': component_name,
111162
'docstring': inspect.getdoc(func) or '',
112163
'signature': inspect.signature(func),
113164
'parameters': {},

scripts/generate_readme/tests/test_metadata_parser.py

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,57 @@ def test_is_component_decorator_wrong_decorator(self):
201201
decorator = tree.body[0].decorator_list[0]
202202

203203
assert parser._is_component_decorator(decorator) is False
204+
205+
def test_extract_decorator_name_component(self, temp_dir):
206+
"""Test extracting name parameter from @dsl.component decorator."""
207+
component_file = temp_dir / "component.py"
208+
component_file.write_text("""
209+
from kfp import dsl
210+
211+
@dsl.component(name='custom-component-name', base_image='python:3.10')
212+
def my_component(param: str) -> str:
213+
'''A component with custom name in decorator.
214+
215+
Args:
216+
param: Input parameter.
217+
218+
Returns:
219+
Output value.
220+
'''
221+
return param
222+
""")
223+
224+
parser = ComponentMetadataParser(component_file)
225+
226+
# Test finding the function
227+
function_name = parser.find_function()
228+
assert function_name == 'my_component'
229+
230+
# Test extracting decorator name from AST (without executing the code)
231+
decorator_name = parser._get_name_from_decorator_if_exists('my_component')
232+
233+
# Should extract the decorator name
234+
assert decorator_name == 'custom-component-name'
235+
236+
def test_extract_decorator_name_component_no_name(self, temp_dir):
237+
"""Test extracting name when decorator has no name parameter."""
238+
component_file = temp_dir / "component.py"
239+
component_file.write_text("""
240+
from kfp import dsl
241+
242+
@dsl.component(base_image='python:3.10')
243+
def my_component(param: str) -> str:
244+
'''A component without custom name in decorator.'''
245+
return param
246+
""")
247+
248+
parser = ComponentMetadataParser(component_file)
249+
250+
# Test extracting decorator name (should return None)
251+
decorator_name = parser._get_name_from_decorator_if_exists('my_component')
252+
253+
# Should return None when no name in decorator
254+
assert decorator_name is None
204255

205256

206257
class TestPipelineMetadataParser:
@@ -283,4 +334,55 @@ def test_is_pipeline_decorator_wrong_decorator(self):
283334
decorator = tree.body[0].decorator_list[0]
284335

285336
assert parser._is_pipeline_decorator(decorator) is False
337+
338+
def test_extract_decorator_name_pipeline(self, temp_dir):
339+
"""Test extracting name parameter from @dsl.pipeline decorator."""
340+
pipeline_file = temp_dir / "pipeline.py"
341+
pipeline_file.write_text("""
342+
from kfp import dsl
343+
344+
@dsl.pipeline(name='custom-pipeline-name', description='A test pipeline')
345+
def my_pipeline(input_data: str) -> str:
346+
'''A pipeline with custom name in decorator.
347+
348+
Args:
349+
input_data: Input data path.
350+
351+
Returns:
352+
Output data path.
353+
'''
354+
return input_data
355+
""")
356+
357+
parser = PipelineMetadataParser(pipeline_file)
358+
359+
# Test finding the function
360+
function_name = parser.find_function()
361+
assert function_name == 'my_pipeline'
362+
363+
# Test extracting decorator name from AST (without executing the code)
364+
decorator_name = parser._get_name_from_decorator_if_exists('my_pipeline')
365+
366+
# Should extract the decorator name
367+
assert decorator_name == 'custom-pipeline-name'
368+
369+
def test_extract_decorator_name_pipeline_no_name(self, temp_dir):
370+
"""Test extracting name when decorator has no name parameter."""
371+
pipeline_file = temp_dir / "pipeline.py"
372+
pipeline_file.write_text("""
373+
from kfp import dsl
374+
375+
@dsl.pipeline(description='A test pipeline')
376+
def my_pipeline(input_data: str) -> str:
377+
'''A pipeline without custom name in decorator.'''
378+
return input_data
379+
""")
380+
381+
parser = PipelineMetadataParser(pipeline_file)
382+
383+
# Test extracting decorator name (should return None)
384+
decorator_name = parser._get_name_from_decorator_if_exists('my_pipeline')
385+
386+
# Should return None when no name in decorator
387+
assert decorator_name is None
286388

0 commit comments

Comments
 (0)