http sender drops reply keys no output port declares

A sender without outputs no longer fails on a {"status": "ok"} reply.
This commit is contained in:
2026-10-15 09:57:51 +02:00
parent e6afc23952
commit 0562753d9c
3 changed files with 34 additions and 5 deletions
+8 -3
View File
@@ -301,7 +301,7 @@ class HttpNode(Node):
:type params: dict
:param kwargs: Pipeline data to send (from required messages).
:type kwargs: Any
:returns: Response data if the endpoint returns JSON, None otherwise.
:returns: The reply's values for the declared output ports, or None.
:rtype: dict | None
"""
client = shared_client()
@@ -330,8 +330,13 @@ class HttpNode(Node):
except Exception:
return None
# Outputs are keyed by port, so only an object can be one. A bare
# scalar or list is a valid reply, just not something to publish.
return body if isinstance(body, dict) else None
# scalar or list is a valid reply, just not something to publish,
# and neither is a key no port declares: the reply is the
# endpoint's, and a `status` beside the values is common.
if not isinstance(body, dict):
return None
ports = {spec.port for spec in self.output_ports}
return {key: value for key, value in body.items() if key in ports} or None
except httpx.HTTPStatusError as e:
logger.error(
+25 -1
View File
@@ -2,8 +2,10 @@
import asyncio
import httpx
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes import MqttNode
from fluksio.flow.nodes import HttpNode, MqttNode
from fluksio.flow.nodes.http import close_shared_client, shared_client
@@ -20,6 +22,28 @@ def test_http_senders_share_one_pooled_client():
close_shared_client()
def test_an_http_reply_keeps_only_the_declared_ports(monkeypatch):
"""A `status` beside the value is the endpoint's business, not an error."""
reply = {"temp": 21.5, "status": "ok"}
client = httpx.Client(
transport=httpx.MockTransport(lambda _: httpx.Response(200, json=reply))
)
monkeypatch.setattr("fluksio.flow.nodes.http.shared_client", lambda: client)
def sender(*provides: MessageSpec) -> HttpNode:
node = HttpNode(
requires=[MessageSpec(name="setpoint", dtype=DType.FLOAT)],
provides=provides,
params={"url": "http://thermostat.test/set"},
)
node.assign_flow("heating", "thermostat")
return node
temp = MessageSpec(name="temp", dtype=DType.FLOAT)
assert sender(temp).execute({"heating.setpoint": 21.0}) == {"heating.temp": 21.5}
assert sender().execute({"heating.setpoint": 21.0}) is None
def _sender() -> MqttNode:
node = MqttNode(
requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
+1 -1
View File
@@ -63,7 +63,7 @@ The ports pick the direction: outputs only make a **listener**, inputs a
**sender**. A listener is a **webhook**: the engine mounts a route at
`/hooks/{flow}/{url}/{secret}` while the flow runs. A sender requests `url`
and publishes a JSON object response on its outputs, one key per output port.
A key no port declares fails the call, and a response that is not an object
Keys no port declares are dropped, and a response that is not an object
publishes nothing.
| Setting | Default | Notes |