Allow workflow plugins to specify their input and output ports (CMEM-4791)

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
 
 ### Added
 
+- Workflow plugins can specify their input and output ports now. 
 - `ResourceParameterType` - for selecting DI resource
 - `CodeParameterType` - which supports various different code languages
 

cmem_plugin_base/dataintegration/context.py

@@ -1,7 +1,7 @@
 """Contains classes to pass context information into plugins.
-The classes in this file are only for documentation purposes.
-The actual classes will be injected by DataIntegration and
- will follow the signatures of the classes below.
+
+The classes in this file are only for documentation purposes. The actual classes will
+be injected by DataIntegration and will follow the signatures of the classes below.
 """
 from dataclasses import dataclass, field
 from typing import Optional, Tuple
@@ -62,15 +62,15 @@ class ExecutionReport:
 
     summary: list[Tuple[str, str]] = field(default_factory=list)
     """Generates a short summary of this report.
-       A sequence of key-value pairs representing the summary table """
+       A sequence of key-value pairs representing the summary table."""
 
     warnings: list[str] = field(default_factory=list)
     """If issues occurred during execution, this contains a list of user-friendly
-    messages. """
+    messages."""
 
     error: Optional[str] = None
     """Error message in case a fatal error occurred. If an error is set, the workflow
-    execution will be stopped after the operator has been executed. """
+    execution will be stopped after the operator has been executed."""
 
 
 class ReportContext:

cmem_plugin_base/dataintegration/plugins.py

@@ -3,13 +3,15 @@
 from typing import Sequence, Optional
 
 from cmem_plugin_base.dataintegration.context import ExecutionContext
-from .entity import Entities
+from cmem_plugin_base.dataintegration.entity import Entities
+from cmem_plugin_base.dataintegration.ports import InputPorts, Port
 
 
 class PluginLogger:
     """Logging API for Plugins.
-    If a plugin is running within DataIntegration, this class
-    will be replaced to log into DI using the path: plugins.python.<plugin_id>."""
+
+    If a plugin is running within DataIntegration, this class will be replaced to
+    log into DI using the path: plugins.python.<plugin_id>."""
 
     def debug(self, message: str) -> None:
         """Log a message with severity 'DEBUG'."""
@@ -30,12 +32,13 @@ def error(self, message: str) -> None:
 
 class PluginConfig:
     """Configuration API for Plugins.
-    If a plugin is running within DataIntegration,
-    this class will be replaced to retrieve the DI configuration
-    in the path: plugins.python.<plugin_id>."""
+
+    If a plugin is running within DataIntegration, this class will be replaced to
+    retrieve the DI configuration in the path: plugins.python.<plugin_id>."""
 
     def get(self) -> str:
         """Retrieve plugin configuration as a JSON string.
+
         This test implementation will return an empty string."""
         return ""
 
@@ -51,6 +54,13 @@ class PluginBase:
 class WorkflowPlugin(PluginBase):
     """Base class of all workflow operator plugins."""
 
+    input_ports: InputPorts
+    """Specifies the input ports that this operator allows."""
+
+    output_port: Optional[Port]
+    """Specifies the output port (if any) of this operator.
+    Should be `None`, if this operator does not return any output."""
+
     def execute(
         self, inputs: Sequence[Entities], context: ExecutionContext
     ) -> Optional[Entities]:
@@ -60,6 +70,9 @@ def execute(
             input. Currently, DI sends ALWAYS an input. in case no connected
             input is there, the sequence has a length of 0.
 
+        :param context: An ExecutionContext object which combines context objects
+            that are available during plugin execution.
+
         :return: The entities generated from the inputs. At the moment, only one
             entities objects be returned (means only one outgoing connection)
             or none (no outgoing connection).
@@ -74,8 +87,10 @@ class TransformPlugin(PluginBase):
     def transform(self, inputs: Sequence[Sequence[str]]) -> Sequence[str]:
         """
         Transforms a collection of values.
+
         :param inputs: A sequence which contains as many elements as there are input
             operators for this transformation.
             For each input operator it contains a sequence of values.
+
         :return: The transformed values.
         """

cmem_plugin_base/dataintegration/ports.py

@@ -0,0 +1,45 @@
+"""Workflow operator input and output ports."""
+
+from typing import Sequence
+from cmem_plugin_base.dataintegration.entity import EntitySchema
+
+
+class Port:
+    """Specifies the type of input or output ports."""
+
+
+class FixedSchemaPort(Port):
+    """Input or output port that has a fixed schema."""
+
+    def __init__(self, schema: EntitySchema):
+        self.schema = schema
+
+
+class FlexibleSchemaPort(Port):
+    """Port that does not have a fixed schema, but will adapt its schema to the
+    connected port.
+    Flexible input ports will adapt the schema to the connected output.
+    Flexible output ports will adapt the schema to the connected input.
+    It is not allowed to connect two flexible ports."""
+
+
+class UnknownSchemaPort(Port):
+    """Port for which the schema is not known in advance.
+    This includes output ports with a schema that depends on external factors
+    (e.g., REST requests)."""
+
+
+class InputPorts:
+    """Specifies the input ports of a workflow operator."""
+
+
+class FixedNumberOfInputs(InputPorts):
+    """Operator accepts a fixed number of inputs."""
+
+    def __init__(self, ports: Sequence[Port]):
+        self.ports = ports
+
+
+class FlexibleNumberOfInputs(InputPorts):
+    """Operator accepts a flexible number of inputs.
+    At the moment, each input is a flexible schema port."""