HTTP Read Task
Learn how to acquire data from HTTP endpoints with Synnax.
For the task lifecycle, see Task Basics.
Task Configuration Reference
Endpoint Configuration Reference
Each endpoint defines an HTTP request to poll and a set of fields to extract from its JSON response.
Field Extraction Reference
Fields define how to extract values from a JSON response body using JSON Pointers (RFC 6901). Each field maps a location in the response to a Synnax channel.
For example, given this response:
{ "temperature": 23.5, "sensors": { "pressure": 101.3 } }A pointer of /temperature extracts 23.5, and /sensors/pressure extracts 101.3.
Enum Values
When the HTTP response contains string values that represent discrete states, use
enum_values to map them to numbers. The Driver converts matching strings to their
numeric equivalents before writing to the channel.
http.ReadField(
pointer="/status",
channel=status_ch.key,
data_type="float64",
enum_values=[{"label": "OFF", "value": 0}, {"label": "ON", "value": 1}, {"label": "ERROR", "value": 2}],
)If the response value does not match any key in the map, the Driver reports an error.
Timestamp Handling
By default, the Driver automatically generates timestamps for each poll cycle using
software timing. If your HTTP response includes a timestamp field, you can extract it
directly by setting time_format on a field whose channel data type is TIMESTAMP.
Supported Timestamp Formats
Software vs. Hardware Timing
Software timing (default): The Driver generates a timestamp at each poll cycle. All
fields that share the same index channel are written atomically. No time_format is
needed on any field.
Hardware timing: If your response includes a timestamp, create a TIMESTAMP channel
as the index and add a field that extracts it with a time_format. Other fields using
that index channel will be grouped with the extracted timestamp.
How-To
Console
Python
Configure and run task
import synnax as sy
from synnax import http
client = sy.Synnax()
# Retrieve the HTTP device
dev = client.devices.retrieve(name="My HTTP Server")
# Create an index channel for timestamps
http_time = client.channels.create(
name="http_time",
is_index=True,
data_type=sy.DataType.TIMESTAMP,
retrieve_if_name_exists=True,
)
# Create data channels
temperature = client.channels.create(
name="http_temperature",
index=http_time.key,
data_type=sy.DataType.FLOAT64,
retrieve_if_name_exists=True,
)
pressure = client.channels.create(
name="http_pressure",
index=http_time.key,
data_type=sy.DataType.FLOAT64,
retrieve_if_name_exists=True,
)
# Create the read task
task = http.ReadTask(
name="HTTP Read Task",
device=dev.key,
rate=1, # Poll at 1 Hz
endpoints=[
http.ReadEndpoint(
method="GET",
path="/api/v1/data",
fields=[
http.ReadField(
pointer="/temperature",
channel=temperature.key,
data_type="float64",
name="Temperature",
),
http.ReadField(
pointer="/pressure",
channel=pressure.key,
data_type="float64",
name="Pressure",
),
],
),
],
)
# Configure the task with Synnax
client.tasks.configure(task)
# Start task and read data
with task.run():
with client.open_streamer(
["http_temperature", "http_pressure"]
) as streamer:
for _ in range(10):
frame = streamer.read()
print(frame) Poll multiple endpoints
import synnax as sy
from synnax import http
client = sy.Synnax()
dev = client.devices.retrieve(name="My HTTP Server")
http_time = client.channels.create(
name="http_time",
is_index=True,
data_type=sy.DataType.TIMESTAMP,
retrieve_if_name_exists=True,
)
temperature = client.channels.create(
name="http_temperature",
index=http_time.key,
data_type=sy.DataType.FLOAT64,
retrieve_if_name_exists=True,
)
uptime = client.channels.create(
name="http_uptime",
index=http_time.key,
data_type=sy.DataType.FLOAT64,
retrieve_if_name_exists=True,
)
task = http.ReadTask(
name="Multi-Endpoint Read",
device=dev.key,
rate=1,
endpoints=[
http.ReadEndpoint(
method="GET",
path="/api/v1/data",
fields=[
http.ReadField(
pointer="/temperature",
channel=temperature.key,
data_type="float64",
),
],
),
http.ReadEndpoint(
method="GET",
path="/api/v1/status",
fields=[
http.ReadField(
pointer="/uptime",
channel=uptime.key,
data_type="float64",
),
],
),
],
)
client.tasks.configure(task) Use enum values
import synnax as sy
from synnax import http
client = sy.Synnax()
dev = client.devices.retrieve(name="My HTTP Server")
http_time = client.channels.create(
name="http_time",
is_index=True,
data_type=sy.DataType.TIMESTAMP,
retrieve_if_name_exists=True,
)
power = client.channels.create(
name="http_power",
index=http_time.key,
data_type=sy.DataType.FLOAT64,
retrieve_if_name_exists=True,
)
task = http.ReadTask(
name="Enum Read",
device=dev.key,
rate=1,
endpoints=[
http.ReadEndpoint(
method="GET",
path="/api/v1/device",
fields=[
http.ReadField(
pointer="/power",
channel=power.key,
data_type="float64",
enum_values=[
{"label": "OFF", "value": 0},
{"label": "ON", "value": 1},
],
),
],
),
],
)
client.tasks.configure(task)