ironflock 1.3.11

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 the publication object, 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

---

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

---

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

---

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.

---

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

Calls a remote procedure registered by another IronFlock device.

result = await ironflock.call("other-device-key", "com.myapp.myprocedure", args=[42])

Parameter	Type	Description
device_key	str	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

---

register_function(topic, endpoint, options=None)

Registers a procedure that can be called by other devices in the fleet.

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

await ironflock.register_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

---

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)

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

---

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/