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})

Returns: Publication — The publication object (an acknowledged publish receipt).

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the publish fails (e.g. not connected).

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})

Returns: Publication — The publication object (an acknowledged publish receipt).

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the publish fails (e.g. not connected).

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})

Returns: Any — The result of the remote procedure call.

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the call fails (e.g. not connected, or the procedure is not registered).

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},
])

Returns: Publication — The publication object (an acknowledged publish receipt).

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the publish fails (e.g. not connected).

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}

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

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the bulk append fails (all-or-nothing — no rows were persisted).

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)

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

Raises: RuntimeError with a descriptive message if the underlying publish/append fails (e.g. not connected).

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)

Returns: Subscription — The subscription object.

Raises: RuntimeError with a descriptive message if the subscription fails (e.g. not connected).

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)

Returns: Subscription — The subscription object.

Raises: RuntimeError with a descriptive message if the subscription fails (e.g. not connected).

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}
    ]
})

# Current value(s) only: add the `latest` marker. The data backend derives
# the latest row per entity in SQL (entity = the table's maintainLatestFlagFor
# columns from the data-template; without one, the single most recent row).
current = await ironflock.getHistory("sensordata", {
    "limit": 100,
    "filterAnd": [{"latest": True}]
})

queryParams fields:

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

IS NULL and IS NOT NULL take no value, and they are the only way to ask about NULL: every other operator yields unknown on a NULL column and excludes the row, exactly as in SQL. LIKE is case-sensitive; use ILIKE to match case-insensitively.

Combining with OR: entries of filterAnd are AND-ed. A group entry combines its own entries with one operator, so the soft-delete pattern becomes:

filterAnd=[
    {
        "combinator": "OR",
        "filters": [
            {"column": "deleted", "operator": "IS NULL"},
            {"column": "deleted", "operator": "=", "value": False},
        ],
    }
]

A data backend that predates groups rejects this shape rather than applying part of it, so it fails loudly instead of quietly returning rows the filter excludes.

Latest values: a {"latest": True} entry in filterAnd is not a WHERE predicate but a mode switch: the data backend returns only the latest row per entity, derived on the fly in SQL (DISTINCT ON over the entity key declared as maintainLatestFlagFor in the table's data-template; a table without an entity key yields the single most recent row). The former physical latest_flag column no longer exists — a legacy {"column": "latest_flag", "operator": "=", "value": True} filter is still accepted and treated as the marker, but new code should use {"latest": True}. Other predicates combine with the marker as expected: entity-key predicates narrow which entities are returned, all other predicates and timeRange filter the resulting latest rows.

Returns: Any — The query result data (typically a list of row dicts). Columns your data template declares secret: true come back as SECRET_PLACEHOLDER — see Secret Columns.

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message when the history procedure is not registered (table not declared / data backend not running) or the call fails.

get_series_history(tablename, params)

Retrieves down-sampled time-series data from a fleet table — numeric columns aggregated into time namespaces (e.g. hourly averages). Ideal for charts over long time ranges. Available for tables (not transforms).

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

params fields:

Returns: Any — The down-sampled series rows.

Raises: ValueError on invalid parameters (including a latest marker in filterAnd); RuntimeError with a descriptive message when the series procedure is not registered or the call fails.

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])

Returns: Any — The result of the remote procedure call.

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the call fails (e.g. not connected, or the procedure is not registered).

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])

Returns: Any — The result of the remote procedure call.

Raises: ValueError on invalid parameters; RuntimeError with a descriptive message if the call fails (e.g. not connected, or the procedure is not registered).

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)

Returns: Registration — The registration object.

Raises: RuntimeError with a descriptive message if the registration fails (e.g. not connected).

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)

Returns: Any — The result of the location update call.

Raises: ValueError on invalid coordinates; RuntimeError with a descriptive message if the location service call fails.

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"

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

Lifecycle Methods

Connection reliability

The connection reconnects on its own. When the socket drops, the SDK retries until the router is back, then restores every subscription and every registered device function — you do not need to re-subscribe after a reconnect. If one topic cannot be restored — a permission that changed, a transient router error — the rest still are, and the failed one is retried on the next reconnect.

A special case is a router that is reachable but has no realm for the app yet. The concurrent-install race boots the app container before the databackend has provisioned its realm, so the first joins are refused with wamp.error.no_such_realm; the SDK keeps retrying every couple of seconds and the app comes up the moment the realm does. A realm still missing after a minute is most likely never going to appear — the app was deleted, or has no databackend for this stage — so from then on the SDK slows to one attempt every two minutes rather than hitting the router every second forever. The fast cadence returns as soon as the realm appears, or the router itself goes away and comes back.

The harder case is a socket that dies silently: an idle NAT or proxy cuts the connection without telling either side, or a router restarts behind a load balancer. Nothing arrives to signal the loss. A connection that only publishes finds out on its next failing write, but a connection that only subscribes never writes at all, so without help it sits dead until the process restarts.

WebSocket keepalive closes that gap. The SDK pings the router every 30 seconds while the link is idle and expects a pong within 10 seconds; if none arrives, the transport is dropped and the normal reconnect path runs. Incoming traffic counts as proof of life, so a busy connection is never probed — the ping only goes out when the link has actually gone quiet.

This is on by default and needs no configuration. Cross-app connections from connect_to_app() are typically subscribe-only, which is exactly the case keepalive protects, and they are covered automatically.

The JavaScript SDK solves the same problem differently: the browser WebSocket API exposes no ping, so it uses a WAMP-level heartbeat with tunable heartbeatIntervalMs / heartbeatTimeoutMs properties instead.

Secret Columns

A data-template column declared secret: true is encrypted by the data backend (AES-256-GCM) before it is written, so what Postgres stores is an ifsec:1:<base64url> envelope, not the value. The flag is allowed on dataType: string columns only, never on tsp, and never on a column used as an entity key (maintainLatestFlagFor).

No read path hands the plaintext back by accident:

None stays None everywhere, so "set but hidden" remains distinguishable from "never written".

Both procedures below are registered per app realm and callable only by your app's own containers — the router denies every browser role and every consuming app, so another app can never decrypt your secrets.

Encryption uses a random IV, so two rows holding the same plaintext hold different ciphertext. Equality filters, GROUP BY, ORDER BY, DISTINCT and entity keys on a secret column are impossible, not merely discouraged — select the rows by their other columns.

reveal_secrets(tablename, query_params={"limit": 10})

Reads rows of your own table with the secret columns decrypted.

# The most recent row, secrets in clear
rows = await ironflock.reveal_secrets("credentials", {
    "limit": 1,
    "filterAnd": [{"latest": True}],
})
print(rows[0]["api_key"])  # the plaintext, not "__secret__"

# Narrow by a NON-secret column — filtering by the secret itself can never match
rows = await ironflock.reveal_secrets("credentials", {
    "limit": 10,
    "filterAnd": [{"column": "gateway_id", "operator": "=", "value": 471}],
})

Returns: Any — The rows, in the same shape getHistory returns, with the secret columns decrypted.

Raises: ValueError on invalid parameters; RuntimeError when the procedure is not registered (the table declares no secret column / data backend not running) or the call fails.

verify_secret(tablename, column, candidate, query_params={"limit": 1})

Checks a candidate against a stored secret without reading it back — the comparison happens inside the data backend, in constant time, and nothing decrypted leaves it. Prefer this whenever you don't actually need the plaintext.

# Is this the current API key? (default query = the most recent row)
ok = await ironflock.verify_secret("credentials", "api_key", received_key)

# Check against a particular entity's rows
ok = await ironflock.verify_secret("credentials", "api_key", received_key, {
    "limit": 1,
    "filterAnd": [
        {"latest": True},
        {"column": "gateway_id", "operator": "=", "value": 471},
    ],
})

Returns: bool — True when any selected row's column decrypts to candidate.

Raises: ValueError on invalid parameters; RuntimeError when the procedure is not registered or the call fails.

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")

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:

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, columns) — including the {"latest": True} marker in filterAnd for reading the provider's current values.

rows = await weather.get_history("forecasts", {"limit": 100})
current = await weather.get_history("forecasts", {
    "limit": 100,
    "filterAnd": [{"latest": True}],
})

Columns the provider marks secret: true (see Secret Columns) are never readable across apps — a non-null value arrives as SECRET_PLACEHOLDER, and only the owning app's own containers can decrypt it. Filtering on or selecting such a column raises CrossAppAccessError with code SECRET_COLUMN client-side rather than returning an empty or masked result.

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):

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:

Managed File Storage

Every app databackend gets private object storage alongside its tables. With no files: section in the data template you still get one namespace named default, so this works against apps released before FleetFiles existed.

# Store an object and get a permanent URL back in the same call
info = await ironflock.files.put("part-1.jpg", jpeg_bytes, content_type="image/jpeg")

# That URL is safe to put in a FleetDB column — a dashboard widget can render
# <img src="{{photo_url}}"> and it just works
await ironflock.publish_to_table("inspections", part_id="1", photo_url=info.url)

data = await ironflock.files.get("part-1.jpg")
async for obj in ironflock.files.iter(prefix="2026/"):
    print(obj.key, obj.size)

The URL never expires, yet stays readable only to an authenticated requestor holding READ on this databackend — an auth proxy re-checks on every request, so it is safe to store but not a public link.

A namespace is a key prefix that carries policy — retention, sharing, allowed content types. It is not a separate S3 bucket; every namespace lives in the app's one bucket. Declare one only when a set of objects needs different rules; otherwise stay in default and organise with key paths.

Declare additional namespaces in data-template.yml:

files:
  # Budget the app suggests for itself; the project user's setting is what gets
  # enforced. catalog() reports both as quota_bytes and suggested_quota_bytes.
  quotaBytes: 5368709120
  namespaces:
    - name: frames
      description: Raw camera frames, one JPEG per inspected part.
      contentTypes: ["image/jpeg"]
      maxObjectBytes: 20971520
      retention: { deleteAfter: 30 days }

The budget is app-wide, not per namespace — a namespace is only a key prefix inside the app's one storage area, so there is nothing for a per-prefix budget to be enforced against. maxObjectBytes is per namespace: it caps a single object, not a total.

files.put(key, data, namespace="default", content_type=None)

Stores bytes. Returns ObjectInfo with key, size, etag, content_type and url.

files.get(key, namespace="default")

Returns bytes.

files.list(prefix="", namespace="default", limit=200, cursor="")

One page: .objects, .prefixes, .is_truncated, .cursor.

files.iter(prefix="", namespace="default")

Async iterator over every object under a prefix, paginating for you.

files.stat(key) / files.exists(key)

Metadata without transferring; exists returns a bool.

files.delete(key) / files.copy(key, to) / files.move(key, to)

move is copy-then-delete on the client (the service has no move verb), so it is not atomic — a failed delete leaves both copies.

files.put_file(key, path) / files.get_to_file(key, path)

Convenience wrappers over a local file.

files.url(key, namespace="default", version=None)

The permanent URL. Returns None where the deployment has no HTTP edge (a plain-HTTP appliance) — the signal to fall back to files.get(). Pass an ETag as version to let browsers cache it immutably.

files.usage(detail=False)

What the object store reports, in one call: size_bytes, object_count, quota_bytes (the enforced budget) and free_bytes. free_bytes is -1 when there is no quota — a quota of 0 means unlimited, and 0 free would read as full.

Pass detail=True for a per_namespace breakdown. That costs one listing per namespace (the store accounts per bucket; a namespace is a prefix), so it is off by default.

files.namespaces() / files.catalog()

What this app may use. catalog() reports quota_bytes (enforced, read from the object store) alongside suggested_quota_bytes (what the app's data template asked for) — the quota is a user setting, so those two can differ. catalog() also reports inline_max_bytes and public_base_url; it is cached after the first call.

files.share_url(key, namespace="default", ttl=900)

An expiring link anyone holding it can fetch. Unlike files.url() this is a bearer capability — nothing re-checks authorization when it is used. Hand it to a person; do not store it in a column. The server clamps ttl.

files.upload_url(key, namespace="default", ttl=3600, content_type=None, size=None)

An expiring URL that accepts a direct upload. Returns url, method, headers and expires_in; send exactly those headers or the signature will not verify.

Large objects

The SDK picks the transport by size, automatically:

put_file() and get_to_file() stream from and to disk on the direct path, so a multi-gigabyte object never has to fit in memory. put() and get() work on bytes and therefore do hold the object in memory — prefer the file variants above a few hundred MB.

Two ceilings remain, and both raise TOO_LARGE with a reason that says which one you hit:

• 5 GiB — S3's single-upload limit. Multipart is not implemented yet.
• inline_max_bytes where there is no direct endpoint — an air-gapped appliance cannot transfer a large object at all. The message says so, because no retry or smaller chunk can help.

The direct path needs the device to reach the object store host, not just the router. Two field failures have their own codes rather than looking like auth problems: PRESIGN_UNREACHABLE (a proxy allowing only the router) and CLOCK_SKEW (S3 rejects requests more than 15 minutes out of step — check NTP).

FileStoreError

Every failure raises FileStoreError with a stable .code and a human-readable .reason. Branch on .code, never on .reason.

from ironflock.filestore import FileStoreError

try:
    await ironflock.files.put("huge.bin", payload)
except FileStoreError as e:
    if e.code == "QUOTA_EXCEEDED":
        ...

A newer server may add codes; unknown ones pass through as .code rather than crashing, so treat anything unrecognised as a generic failure.

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/