Declare the input port, naming the paths the Jinja templates ask for

CHANGELOG.md

@@ -23,6 +23,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
 - The result always leaves on the output port now, one entity per query execution
 - The task no longer talks to Corporate Memory at all, and no longer needs a deployment
   to be reachable in order to run
+- The task declares its input now: one port when the query or the variables carry Jinja
+  syntax, naming the paths those templates ask for, and no port at all otherwise, since
+  a task without Jinja ignores whatever is connected. A port that names no paths is
+  handed nothing, because DataIntegration reads only the paths a task requests
 - The output schema is derived from **Query** wherever the query describes its own
   response, so the paths are offered to the next task while the workflow is drawn
   instead of only becoming known once the task has run. A query that does not parse
@@ -74,6 +78,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
 - An empty response object counts as a result rather than as a failed entity
 - The execution report calls an anonymous `{ ... }` query a read, where it used to call
   anything not starting with the word `query` a write
+- A run configured for Jinja that receives no entities warns about it in the report,
+  instead of reporting a successful run that queried nothing
 
 ### Breaking Change
 

cmem_plugin_graphql/workflow/graphql.py

@@ -20,6 +20,7 @@
 from cmem_plugin_base.dataintegration.parameter.password import Password, PasswordParameterType
 from cmem_plugin_base.dataintegration.plugins import WorkflowPlugin
 from cmem_plugin_base.dataintegration.ports import (
+    FixedNumberOfInputs,
     FixedSchemaPort,
     UnknownSchemaPort,
 )
@@ -34,6 +35,7 @@
 from cmem_plugin_graphql.workflow.utils import (
     entities_from_payload,
     get_dict,
+    input_schema_from_templates,
     is_jinja_template,
     output_schema_from_query,
     render_template,
@@ -265,7 +267,8 @@ def execute(self, inputs: Sequence[Entities], context: ExecutionContext) -> Enti
         processed_entities: int = 0
         failed_entities: int = 0
         payload = []
-        if (inputs and self.jinja_query) or self.jinja_variable_values:
+        per_entity = bool((inputs and self.jinja_query) or self.jinja_variable_values)
+        if per_entity:
             for entities in inputs:
                 for result in self.process_entities(entities=entities, context=context):
                     if result is None:
@@ -290,9 +293,16 @@ def execute(self, inputs: Sequence[Entities], context: ExecutionContext) -> Enti
             processed_entities += 1
             payload.append(result)
 
-        summary: list[tuple[str, str]] = []
+        summary: list[tuple[str, str]] = [("Failed entities", str(failed_entities))]
         warnings: list[str] = []
-        summary.append(("Failed entities", str(failed_entities)))
+        if per_entity and not processed_entities and not failed_entities:
+            # The run looked successful while doing nothing at all, which is how an input
+            # that never arrives presents itself.
+            warnings.append(
+                f"Jinja syntax is configured, so the endpoint is queried once per arriving"
+                f" entity - but nothing arrived on the input, and no query was sent."
+                f" Received {len(inputs)} input collection(s)."
+            )
         context.report.update(
             ExecutionReport(
                 entity_count=processed_entities,
@@ -406,7 +416,27 @@ def _is_canceled(context: ExecutionContext) -> bool:
         return False
 
     def _set_ports(self) -> None:
-        """Define input/output ports based on the configuration"""
+        """Define input/output ports based on the configuration
+
+        The task declared no input port at all until now, so a dataset wired into it
+        was never read: the per entity loop ran over nothing and the task reported a
+        successful run of zero queries.
+
+        A port is declared exactly when the task reads one - when the query or the
+        variables carry Jinja syntax. Without it the task sends its text once and
+        ignores anything connected, so a handle there would only invite a connection
+        that does nothing.
+
+        The port names the paths the templates ask for, rather than leaving the schema
+        unknown. DataIntegration reads only the paths the consuming task requests, so a
+        port naming none is handed nothing - which is the same empty run, declared.
+        """
+        input_schema = input_schema_from_templates(self.graphql_query, self.graphql_variable_values)
+        self.input_ports = (
+            FixedNumberOfInputs([FixedSchemaPort(schema=input_schema)])
+            if input_schema
+            else FixedNumberOfInputs([])
+        )
         self.output_schema = output_schema_from_query(self.graphql_query)
         if self.output_mode == OUTPUT.file:
             self.output_port = FixedSchemaPort(schema=FileEntitySchema())

cmem_plugin_graphql/workflow/utils.py

@@ -6,6 +6,7 @@
 from uuid import uuid4
 
 import jinja2
+import jinja2.meta
 from cmem_plugin_base.dataintegration.entity import Entities, Entity, EntityPath, EntitySchema
 from cmem_plugin_base.dataintegration.utils.entity_builder import build_entities_from_data
 from gql import gql
@@ -208,6 +209,28 @@ def get_dict(entities: Entities) -> Iterator[dict[str, str]]:
         yield result
 
 
+def input_schema_from_templates(*template_texts: str) -> EntitySchema | None:
+    """Derive the schema the task needs from its input, or None when it needs nothing
+
+    A Jinja template names the values it wants, so the paths the task has to be handed
+    are known before the workflow runs. They have to be declared, too: DataIntegration
+    asks the consuming task which paths it wants and reads only those, so a port that
+    names none is handed nothing - a dataset wired into a task declaring an unknown
+    input schema delivers zero entities, and the task reports a successful run over
+    nothing at all.
+    """
+    environment = jinja2.Environment(autoescape=False)  # noqa: S701
+    names: set[str] = set()
+    for text in template_texts:
+        names |= jinja2.meta.find_undeclared_variables(environment.parse(text))
+    if not names:
+        return None
+    return EntitySchema(
+        type_uri="",
+        paths=[EntityPath(path=name, is_single_value=True) for name in sorted(names)],
+    )
+
+
 def render_template(template_text: str, values: dict[str, str]) -> str:
     """Render a Jinja template that produces GraphQL or JSON, not HTML
 

tests/test_graphql.py

@@ -23,6 +23,7 @@
 from cmem_plugin_graphql.workflow.graphql import OUTPUT, RESULT_FILE_NAME, GraphQLPlugin
 from cmem_plugin_graphql.workflow.utils import (
     entities_from_payload,
+    input_schema_from_templates,
     is_jinja_template,
     output_schema_from_query,
     render_template,
@@ -676,3 +677,32 @@ def test_the_plugin_identifier_never_moves() -> None:
     """
     plugin = next(iter(discover_plugins("cmem_plugin_graphql").plugins))
     assert plugin.plugin_id == "cmem_plugin_graphql-Query"
+
+
+def test_the_input_schema_names_the_jinja_variables() -> None:
+    """Test that the paths the templates ask for are the paths the task requests"""
+    schema = input_schema_from_templates(
+        "query manzana($id: ID!){fruit(id: $id){ {{ field }} }}", '{"id": {{ id }}}'
+    )
+    assert schema is not None
+    assert [(p.path, p.is_single_value) for p in schema.paths] == [("field", True), ("id", True)]
+
+
+def test_a_task_without_jinja_declares_no_input() -> None:
+    """Test that a task ignoring its input offers no handle to connect one to"""
+    plugin = build_plugin(graphql_query=FRUIT_QUERY)
+    assert plugin.input_ports.ports == []
+
+
+def test_a_task_with_jinja_requests_the_paths_it_renders() -> None:
+    """Test that the declared input port names the paths, rather than leaving them unknown.
+
+    A port that names no paths is handed nothing: DataIntegration reads only what the
+    consuming task asks for.
+    """
+    plugin = build_plugin(
+        graphql_query=FRUIT_QUERY_WITH_VARIABLE, graphql_variable_values='{"id" : {{ id }}}'
+    )
+    ports = plugin.input_ports.ports
+    assert len(ports) == 1
+    assert [p.path for p in ports[0].schema.paths] == ["id"]