ironflock 1.3.14

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.

---

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.

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

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/