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

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/) and this project adheres to [Semantic Versioning](https://semver.org/)
 
+## [Unreleased]
+
+### Added
+
+- Workflow plugins can specify their input and output ports now. 
+
 ## [4.1.0] 2023-07-12
 
 ### Changed

cmem_plugin_base/dataintegration/plugins.py

@@ -4,6 +4,7 @@
 
 from cmem_plugin_base.dataintegration.context import ExecutionContext
 from .entity import Entities
+from .ports import InputPorts, Port
 
 
 class PluginLogger:
@@ -51,6 +52,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]:

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."""