ironflock 1.5.1

IronFlock Python SDK for connecting to the IronFlock Platform

ironflock

About

With this library you can publish data from your apps on your IoT edge hardware to the fleet data storage of the IronFlock devops platform. When this library is used on a certain device the library automatically uses the private messaging realm (Unified Name Space) of the device's fleet and the data is collected in the respective fleet database.

So if you use the library in your app, the data collection will always be private to the app user's fleet.

For more information on the IronFlock IoT Devops Platform for engineers and developers visit our IronFlock home page.

Requirements

• Python 3.8 or higher

Installation

Install from PyPI:

pip install ironflock

Usage

import asyncio
from ironflock import IronFlock

# create an IronFlock instance to connect to the IronFlock platform data infrastructure.
# The IronFlock instance handles authentication when run on a device registered in IronFlock.
ironflock = IronFlock()

async def main():
    while True:
        # publish an event (if connection is not established the publish is skipped)
        publication = await ironflock.publish("test.publish.example", {"temperature": 20})
        print(publication)
        await asyncio.sleep(3)


if __name__ == "__main__":
    ironflock = IronFlock(mainFunc=main)
    ironflock.run()

Options

The IronFlock __init__ function can be configured with the following options:

{
    serial_number: string;
}

serial_number: Used to set the serial_number of the device if the DEVICE_SERIAL_NUMBER environment variable does not exist. It can also be used if the user wishes to authenticate as another device.

API Reference

publish(topic, *args, **kwargs)

Publishes an event to a topic on the IronFlock message router.

publication = await ironflock.publish("com.myapp.mytopic", {"temperature": 20})

Parameter	Type	Description
topic	str	The URI of the topic to publish to
*args	positional	Payload arguments
**kwargs	keyword	Payload keyword arguments

Returns: Publication | None — The publication object (an acknowledged publish receipt), or None if the publish failed.

---

publish_to_table(tablename, *args, **kwargs)

Convenience function to publish data to a fleet table in the IronFlock platform. Automatically constructs the correct topic using the SWARM_KEY and APP_KEY environment variables.

await ironflock.publish_to_table("sensordata", {"temperature": 22.5, "humidity": 60})

Parameter	Type	Description
tablename	str	The name of the table, e.g. "sensordata"
*args	positional	Row data to publish
**kwargs	keyword	Row data as keyword arguments

Returns: Publication | None — The publication object (an acknowledged publish receipt), or None if the publish failed.

---

append_to_table(tablename, *args, **kwargs)

Appends data to a fleet table by calling the registered append procedure at append.<SWARM_KEY>.<APP_KEY>.<tablename>. Unlike publish_to_table, this uses a remote procedure call rather than a pub/sub event.

await ironflock.append_to_table("sensordata", {"temperature": 22.5, "humidity": 60})

Parameter	Type	Description
tablename	str	The name of the table, e.g. "sensordata"
*args	positional	Row data to append
**kwargs	keyword	Row data as keyword arguments

Returns: Any | None — The result of the remote procedure call, or None if the call failed.

---

publish_rows_to_table(tablename, rows, **kwargs)

Publishes many rows in a single message (bulk insert) to the dedicated topic bulk.<SWARM_KEY>.<APP_KEY>.<tablename>. The platform inserts the whole batch atomically (all-or-nothing) in one operation. Use this for high-frequency data where one round-trip per row is too costly. Like publish_to_table, this is fire-and-forget — the ack confirms delivery to the router, not the DB insert.

await ironflock.publish_rows_to_table("sensordata", [
    {"tsp": "2024-01-15T10:30:00.000Z", "temperature": 22.5},
    {"tsp": "2024-01-15T10:30:01.000Z", "temperature": 22.7},
])

Parameter	Type	Description
tablename	str	The name of the table, e.g. "sensordata"
rows	List[dict]	Non-empty list of row dicts to insert
**kwargs	keyword	Extra arguments shared by the whole batch

Returns: Publication | None — The publication object (an acknowledged publish receipt), or None if the publish failed.

---

append_rows_to_table(tablename, rows, **kwargs)

Appends many rows in a single RPC (bulk insert) by calling the dedicated procedure appendBulk.<SWARM_KEY>.<APP_KEY>.<tablename>. The platform inserts the whole batch atomically (all-or-nothing): if any row is invalid the entire batch is rejected and nothing is persisted. Prefer this over publish_rows_to_table when you need the insert outcome.

result = await ironflock.append_rows_to_table("sensordata", [
    {"tsp": "2024-01-15T10:30:00.000Z", "temperature": 22.5},
    {"tsp": "2024-01-15T10:30:01.000Z", "temperature": 22.7},
])
# result -> {"success": True, "count": 2}

Parameter	Type	Description
tablename	str	The name of the table, e.g. "sensordata"
rows	List[dict]	Non-empty list of row dicts to insert
**kwargs	keyword	Extra arguments shared by the whole batch

Returns: Any | None — The result of the remote procedure call (e.g. {"success": True, "count": N}), or None if the call failed.

---

report_error(error, level="error", append=False, tsp=None)

Reports an application error into the fleet's error-logs table. This is a convenience wrapper over publish_to_table / append_to_table: it stamps the row with source="app", a severity level, and a timestamp, then writes it like any normal table row. The error lands in the same per-databackend error-logs table that fleetdb system errors use (tagged source="system"), so it is queryable with getHistory, streamable with subscribe_to_table, usable in board-templates, and delivered in realtime on transformed.error-logs — without firing the platform's system-error toast.

# Fire-and-forget (default): publishes to the error-logs table
await ironflock.report_error("Sensor read timed out", level="warn")

# Pass an Exception to capture its traceback (falls back to the message)
try:
    risky_operation()
except Exception as err:
    await ironflock.report_error(err)

# Use the append RPC when you want to await the insert outcome
await ironflock.report_error("Calibration failed", level="error", append=True)

Parameter	Type	Description
error	str | BaseException	The error message, or an exception whose traceback (or message) is recorded
level	str, optional	Severity: "error", "warn", "info" or "debug". Defaults to "error"
append	bool, optional	When True, use the append RPC (returns the insert outcome). Defaults to False (fire-and-forget publish)
tsp	str, optional	ISO-8601 timestamp override. Defaults to the current time

Returns: Publication | Any | None — The publication object (or, with append=True, the RPC result), or None if the call failed.

---

subscribe(topic, handler, options=None)

Subscribes to a topic on the IronFlock message router.

def on_message(*args, **kwargs):
    print("Received:", args, kwargs)

subscription = await ironflock.subscribe("com.myapp.mytopic", on_message)

Parameter	Type	Description
topic	str	The URI of the topic to subscribe to
handler	callable	Function called when a message is received
options	SubscribeOptions, optional	Subscription options

Returns: Subscription | None — The subscription object, or None if the subscription failed.

---

subscribe_to_table(tablename, handler, options=None)

Convenience function to subscribe to a fleet table. Automatically constructs the correct topic using the SWARM_KEY and APP_KEY environment variables. Receives rows written via both the single-row and the bulk insert paths — rows from a bulk insert are delivered to your handler one at a time, so handler code stays the same.

def on_table_data(*args, **kwargs):
    print("New row:", args, kwargs)

await ironflock.subscribe_to_table("sensordata", on_table_data)

Parameter	Type	Description
tablename	str	The name of the table to subscribe to
handler	callable	Function called when new data arrives
options	SubscribeOptions, optional	Subscription options

Returns: Subscription | None — The subscription object, or None if the subscription failed.

---

getHistory(tablename, queryParams)

Retrieves historical data from a fleet table.

# Simple query with limit
data = await ironflock.getHistory("sensordata", {"limit": 100})

# Query with time range and filters
data = await ironflock.getHistory("sensordata", {
    "limit": 500,
    "offset": 0,
    "timeRange": {
        "start": "2026-01-01T00:00:00Z",
        "end": "2026-03-01T00:00:00Z"
    },
    "filterAnd": [
        {"column": "temperature", "operator": ">", "value": 20},
        {"column": "humidity", "operator": "<=", "value": 80}
    ]
})

Parameter	Type	Description
tablename	str	The name of the table to query
queryParams	dict or TableQueryParams	Query parameters (see below)

queryParams fields:

Field	Type	Description
limit	int	Maximum number of rows to return (1–10000, required)
offset	int, optional	Offset for pagination
timeRange	dict, optional	{"start": "<ISO datetime>", "end": "<ISO datetime>"}
filterAnd	list, optional	List of AND filter conditions: {"column": str, "operator": str, "value": ...}

Supported filter operators: =, !=, >, <, >=, <=, LIKE, ILIKE, IN, NOT IN, IS, IS NOT.

Returns: Any | None — The query result data (typically a list of row dicts), or None if the call failed.

---

call(topic, args=None, kwargs=None, options=None)

Calls a remote procedure on the IronFlock message router using a full WAMP topic URI.

result = await ironflock.call("some.full.wamp.topic", args=[42])

Parameter	Type	Description
topic	str	The full WAMP URI of the procedure to call
args	list, optional	Positional arguments
kwargs	dict, optional	Keyword arguments
options	CallOptions, optional	Call options

Returns: Any | None — The result of the remote procedure call, or None if the call failed.

---

call_device_function(device_key, topic, args=None, kwargs=None, options=None)

Calls a remote procedure registered by another IronFlock device. Automatically assembles the full WAMP topic as {swarm_key}.{device_key}.{app_key}.{env}.{topic}.

result = await ironflock.call_device_function(42, "com.myapp.myprocedure", args=[42])

Parameter	Type	Description
device_key	int	The device key of the target device
topic	str	The URI of the procedure to call
args	list, optional	Positional arguments
kwargs	dict, optional	Keyword arguments
options	CallOptions, optional	Call options

Returns: Any | None — The result of the remote procedure call, or None if the call failed.

call_function() is a deprecated alias for call_device_function().

---

register_device_function(topic, endpoint, options=None)

Registers a procedure that can be called by other devices in the fleet. Automatically constructs the full WAMP topic as {swarm_key}.{device_key}.{app_key}.{env}.{topic}.

def add(a, b):
    return a + b

await ironflock.register_device_function("com.myapp.add", add)

Parameter	Type	Description
topic	str	The URI of the procedure to register
endpoint	callable	The function to register
options	RegisterOptions, optional	Registration options

Returns: Registration | None — The registration object, or None if the registration failed.

register() is an alias for register_device_function(). register_function() is a deprecated alias.

---

set_device_location(long, lat)

Updates the device's location in the platform master data. The maps in device or group overviews will reflect the new location in realtime.

await ironflock.set_device_location(long=8.6821, lat=50.1109)

Parameter	Type	Description
long	float	Longitude (-180 to 180)
lat	float	Latitude (-90 to 90)

Returns: Any | None — The result of the location update call, or None if the call failed.

Note: Location history is not stored. If you need location history, create a dedicated table and use publish_to_table.

---

getRemoteAccessUrlForPort(port)

Returns the remote access URL for a given port on the device.

url = ironflock.getRemoteAccessUrlForPort(8080)
# e.g. "https://<device_key>-<app_name>-8080.app.ironflock.com"

Parameter	Type	Description
port	int	The port number

Returns: str | None — The remote access URL string (e.g. "https://<device_key>-<app_name>-8080.app.ironflock.com"), or None if the device key or app name is not available.

---

Properties

Property	Type	Description
is_connected	bool	True if the connection to the platform is established
connection	CrossbarConnection	The underlying connection instance (for advanced use)

Lifecycle Methods

Method	Description
run()	Starts the connection and runs the main function (blocking, synchronous)
await start()	Starts the connection asynchronously
await stop()	Stops the connection and cancels the main task
await run_async()	Starts the connection and keeps it running asynchronously

Cross-App Data Access

Read another app's fleet data from within your app, in the same project and fleet. The provider app must list your app in its data-template consumes: section, and the project user must grant access. Access is read-only: you can query history and subscribe to realtime rows of the tables and transforms (views) the provider shares — you cannot write to them.

from ironflock import IronFlock, CrossAppAccessError

ironflock = IronFlock()
await ironflock.start()

# Open a read-only handle on another app's data backend
weather = await ironflock.connect_to_app("weather-app")

# Inspect what the provider shares (non-private tables / transforms)
print([t["tablename"] for t in weather.tables])

# Query history, just like your own tables
rows = await weather.get_history("forecasts", {"limit": 100})

# Subscribe to realtime rows
def on_forecast(*args, **kwargs):
    print("New forecast:", args)

await weather.subscribe_to_table("forecasts", on_forecast)

# Access errors carry a machine-readable code
try:
    await ironflock.connect_to_app("unshared-app")
except CrossAppAccessError as err:
    print(err.code)  # e.g. "NO_GRANT"

Consumed-app connections are cached per app + stage and are closed automatically by ironflock.stop().

connect_to_app(app_name, stage=None, on_error=None)

Opens a read-only connection to another app's data backend in the same project and returns a ConsumedApp handle. Resolves the provider and connects to its realm using this device's credentials.

weather = await ironflock.connect_to_app("weather-app", stage="prod")

Parameter	Type	Description
app_name	str	Provider app name, as declared in your consumes: section
stage	str, optional	Provider stage to connect to ("dev" or "prod"). Defaults to this app's own stage (ENV)
on_error	callable, optional	Called with a CrossAppAccessError when the connection is fatally denied after connect_to_app resolved (e.g. the grant is later revoked)

Returns: ConsumedApp — A read-only handle on the provider's data backend.

Raises: CrossAppAccessError (code: NO_GRANT, PROVIDER_NOT_INSTALLED, UNKNOWN_APP, or NOT_AUTHORIZED).

---

ConsumedApp handle

Returned by connect_to_app. A read-only view of a provider app's shared tables and transforms.

Properties:

Property	Type	Description
app	str	Provider app name
stage	str	Provider stage this handle is connected to ("dev" or "prod")
tables	List[dict]	Non-private tables the provider shares (dicts with tablename, optional description/columns)
transforms	List[dict]	Non-private transforms (views) the provider shares (same shape as tables)
is_connected	bool	True while the connection to the provider is open
connection	CrossbarConnection	The underlying connection (advanced use)

consumed_app.get_history(tablename, query_params=None)

Queries history rows of a shared table or transform. Takes the same query parameters as getHistory (limit, offset, timeRange, filterAnd).

rows = await weather.get_history("forecasts", {"limit": 100})

Returns: Any — The query result rows.

consumed_app.subscribe_to_table(tablename, handler, options=None)

Subscribes to realtime rows of a shared table or transform. Bulk-inserted rows are delivered one at a time, exactly like subscribe_to_table on your own app.

def on_forecast(*args, **kwargs):
    print("New forecast:", args)

await weather.subscribe_to_table("forecasts", on_forecast)

Returns: Any — The subscription object.

consumed_app.get_series_history(tablename, params)

Queries down-sampled time-series history of a shared table (not available for transforms).

series = await weather.get_series_history("forecasts", {
    "metrics": ["temperature"],
    "method": "AVG",
    "limit": 500,
    "timeRange": ["2026-01-01T00:00:00Z", "2026-03-01T00:00:00Z"],
})

params fields (SeriesQueryParams):

Field	Type	Description
metrics	List[str]	Numeric columns to down-sample
method	str	Aggregation per bucket: "AVG", "SUM", "COUNT", "MIN", "MAX", "FIRST" or "LAST"
limit	int	Maximum number of rows (1–10000)
timeRange	list	[start, end] — ISO strings or epoch-ms numbers; None = open end (required)
groupBy	List[str], optional	Columns to group the series by
filterAnd	list, optional	AND filter conditions

Returns: Any — The down-sampled series rows.

consumed_app.close()

Closes this handle's connection to the provider. (All consumed-app connections are also closed by ironflock.stop().)

Returns: None

---

CrossAppAccessError

Raised by connect_to_app and the ConsumedApp methods when cross-app access is denied or misused. Exposes a machine-readable code:

code	Meaning
NO_GRANT	The project user has not granted your app access to the provider
PROVIDER_NOT_INSTALLED	The provider app has no data backend for that stage in this project
UNKNOWN_APP	No app by that name
PRIVATE_TABLE	The requested table/transform is not in the provider's shared catalog
NOT_AUTHORIZED	The router/provider denied access (e.g. the grant was revoked)

Advanced Usage

If you need more control, e.g. acting on lifecycle events (onJoin, onLeave) take a look at the examples folder.

Development

This project uses uv for dependency management and building.

Install uv if you don't have it:

curl -LsSf https://astral.sh/uv/install.sh | sh

Install dependencies (including dev dependencies):

uv sync --extra dev

Run tests:

just test-unit    # Run unit tests only
just test         # Run all tests
just test-docker  # Run integration tests with Docker

Build and publish a new pypi package:

just publish

Or manually:

# Clean previous builds
rm -rf dist

# Build the package
uv build

# Upload to PyPI
uv publish

Check the package at https://pypi.org/project/ironflock/.

Test Deployment

To test the package before deploying to PyPI you can use test.pypi.

just publish-test

Or manually:

uv build
uv publish --publish-url https://test.pypi.org/legacy/

Once the package is published you can install it from TestPyPI:

pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple/ ironflock

Once the package is published you can use it in other code by putting these lines at the top of the requirements.txt

--index-url https://test.pypi.org/simple/
--extra-index-url https://pypi.org/simple/