dsl: add set_operation() method for setting TOSCA operations directly…

… on template objects.

docs/api.rst

@@ -39,6 +39,8 @@ TOSCA Types
 
   .. automethod:: set_to_property_source
 
+  .. automethod:: set_operation
+
 .. autoclass:: Node
 
   .. automethod:: find_required_by

tests/test_dsl.py

@@ -494,9 +494,10 @@ class Inputs(TopologyInputs):
     example_template_python
     + """
 wordpress_db.db_content = tosca.artifacts.File(file="files/wordpress_db_content.txt")
+
 def create(self):
     return self.find_artifact("db_create.sh").execute(db_data=self.db_content)
-wordpress_db.create = create
+wordpress_db.set_operation(create)
 """
 )
 

tosca-package/tosca/_tosca.py

@@ -14,7 +14,6 @@
 import re
 from typing import (
     Any,
-    Callable,
     ClassVar,
     Dict,
     ForwardRef,
@@ -37,6 +36,8 @@
 import types
 from typing_extensions import (
     Protocol,
+    Callable,
+    Concatenate,
     dataclass_transform,
     get_args,
     get_origin,
@@ -1579,7 +1580,7 @@ def Artifact(
 
 _make_field_doc(Artifact)
 
-_EvalDataExpr = Union["EvalData", str, None, Dict[str, Any], List[Any]]
+_EvalDataExpr = Union[str, None, Dict[str, Any], List[Any]]
 
 
 class _GetName:
@@ -1596,7 +1597,9 @@ class EvalData:
     "A wrapper around JSON/YAML data that may contain TOSCA functions or eval expressions and should be evaluated at runtime."
 
     def __init__(
-        self, expr: _EvalDataExpr, path: Optional[List[Union[str, _GetName]]] = None
+        self,
+        expr: Union["EvalData", _EvalDataExpr],
+        path: Optional[List[Union[str, _GetName]]] = None,
     ):
         if isinstance(expr, EvalData):
             expr = expr.expr
@@ -1661,7 +1664,7 @@ def __str__(self) -> str:
             return "{{ " + jinja + " }}"
         elif isinstance(expr, list):
             return "{{ " + str(expr) + "| map_value }}"
-        return expr or ""  # type: ignore   # unreachable
+        return str(expr or "")
 
     def __repr__(self):
         return f"EvalData({self.expr})"
@@ -1903,7 +1906,7 @@ def is_data_field(obj) -> bool:
     return (
         not callable(obj)
         and not inspect.ismethoddescriptor(obj)
-        and not inspect.isdatadescriptor(object)
+        and not inspect.isdatadescriptor(obj)
     )
 
 
@@ -2396,6 +2399,13 @@ class B(Node):
     return cast(_T, _search(field_name, ".hosted_on", cls_or_obj))
 
 
+# XXX remove when type_extensions 4.13 is released
+if sys.version_info >= (3, 11):
+    _OperationFunc = Callable[Concatenate[Self, ...], Any]
+else:
+    _OperationFunc = Callable
+
+
 class ToscaType(_ToscaType):
     "Base class for TOSCA type definitions."
 
@@ -2428,16 +2438,33 @@ def __post_init__(self):
                     setattr(val, name, value)
         self._initialized = True
 
-    def _enforce_required_fields(self):
+    def _enforce_required_fields(self) -> bool:
         return True
 
     # XXX version (type and template?)
 
-    def register_template(self, current_module, name):
+    def register_template(self, current_module, name) -> None:
         self._all_templates.setdefault(self._template_section, {})[
             (current_module, self._name or name)
         ] = self
 
+    def set_operation(self, op: _OperationFunc, name: Optional[str] = None) -> None:
+        """
+        Assign the given :std:ref:`TOSCA operation<operation>` to this TOSCA object.
+        TOSCA allows operations to be defined directly on templates.
+
+        Args:
+          op: A function implements the operation. It should looks like a method, i.e. accepts ``Self`` as the first argument.
+              Using the `tosca.operation` function decorator is recommended but not required.
+          name: The TOSCA operation name. If omitted, ``op``'s :py:func:`operation_name<tosca.operation>` or function name is used.
+        """
+
+        if not name:
+            name = cast(str, getattr(op, "operation_name", op.__name__))
+        # we invoke methods through a proxy during yaml generation and at runtime so we don't need to worry
+        # that this function will not receive self because are assigning it directly to the object here.
+        setattr(self, name, op)
+
     @classmethod
     def tosca_bases(cls, section=None) -> Iterator[Type["ToscaType"]]:
         for c in cls.__bases__:
@@ -3103,8 +3130,10 @@ def find_all_required_by(
     else:
         find_all_required_by = anymethod(find_all_required_by, keyword="cls_or_obj")
 
+
 NodeType = Node
 
+
 class _OwnedToscaType(ToscaType):
     _local_name: Optional[str] = field(default=None)
     _node: Optional[Node] = field(default=None)
@@ -3172,8 +3201,11 @@ def to_yaml(self, dict_cls=dict):
         for field, value in self.get_instance_fields().values():
             body[field.tosca_name] = to_tosca_value(value, dict_cls)
         return body
+
+
 DataType = DataEntity  # deprecated
 
+
 class OpenDataEntity(DataEntity):
     "Properties don't need to be declared with TOSCA data types derived from this class."
 
@@ -3189,7 +3221,10 @@ def extend(self, **kw) -> Self:
         "Add undeclared properties to the data type."
         self.__dict__.update(kw)
         return self
-OpenDataType = OpenDataEntity # deprecated
+
+
+OpenDataType = OpenDataEntity  # deprecated
+
 
 class CapabilityEntity(_OwnedToscaType):
     _type_section: ClassVar[str] = "capability_types"
@@ -3202,8 +3237,11 @@ def to_template_yaml(self, converter: "PythonToYaml") -> dict:
         tpl = super().to_template_yaml(converter)
         del tpl["type"]
         return tpl
+
+
 CapabilityType = CapabilityEntity
 
+
 class Relationship(_OwnedToscaType):
     # the "owner" of the relationship is its source node
     _type_section: ClassVar[str] = "relationship_types"
@@ -3237,8 +3275,11 @@ def __getitem__(self, target: Node) -> Self:
             return dataclasses.replace(self, _target=target)  # type: ignore
         self._target = target
         return self
+
+
 RelationshipType = Relationship
 
+
 class ArtifactEntity(_OwnedToscaType):
     _type_section: ClassVar[str] = "artifact_types"
     _mime_type: ClassVar[Optional[str]] = None
@@ -3288,8 +3329,11 @@ def to_template_yaml(self, converter: "PythonToYaml") -> dict:
     def execute(self, *args: ToscaInputs, **kw):
         self.inputs = ToscaInputs._get_inputs(*args, **kw)
         return self
+
+
 ArtifactType = ArtifactEntity  # deprecated
 
+
 class Interface(ToscaType):
     # "Note: Interface types are not derived from ToscaType"
     _type_section: ClassVar[str] = "interface_types"
@@ -3318,26 +3362,35 @@ def _cls_to_yaml(cls, converter: "PythonToYaml") -> dict:
             yaml[tosca_name].pop("interfaces", None)
             yaml[tosca_name].update(body)
         return yaml
+
+
 InterfaceType = Interface  # deprecated
 
+
 class Policy(ToscaType):
     _type_section: ClassVar[str] = "policy_types"
     _template_section: ClassVar[str] = "policies"
 
     @classmethod
     def _cls_to_yaml(cls, converter: "PythonToYaml") -> dict:
         return cls._shared_cls_to_yaml(converter)
+
+
 PolicyType = Policy  # deprecated
 
+
 class Group(ToscaType):
     _type_section: ClassVar[str] = "group_types"
     _template_section: ClassVar[str] = "groups"
 
     @classmethod
     def _cls_to_yaml(cls, converter: "PythonToYaml") -> dict:
         return cls._shared_cls_to_yaml(converter)
+
+
 GroupType = Group  # deprecated
 
+
 class _ArtifactProxy:
     def __init__(self, name_or_tpl):
         self.name_or_tpl = name_or_tpl

tosca-package/tosca/yaml2python.py

@@ -1362,7 +1362,7 @@ def operation2func(
         # XXX add arguments declared on the interface definition
         # XXX declare configurator/artifact as the return value
         type_anno = ": Any" if not self.concise else ""
-        args = f"{'' if template_name else 'self, '}**kw{type_anno}"
+        args = f"self, **kw{type_anno}"
         type_anno = " -> Any" if not self.concise else ""
         src += f"{indent}def {func_name}({args}){type_anno}:\n"
         indent += "   "
@@ -1711,10 +1711,7 @@ def add_template_interfaces(self, node_template, indent, name) -> str:
         if names:
             self._pending_defs.append(ops_src)
             for op_name in names:
-                if self.assign_attr:
-                    src += f"{indent}{name}.{op_name} = {name}_{op_name}  # type: ignore[attr-defined]\n"
-                else:
-                    src += f"{indent}setattr({name}, '{op_name}', {name}_{op_name})\n"
+                src += f'{indent}{name}.set_operation({name}_{op_name}, "op_name")\n'
         return src
 
     def _get_req_assignment(self, req):