Complete API reference for the PyColonies SDK.
from pycolonies import Colonies
client = Colonies(host, port, tls=False, native_crypto=True)| Parameter | Type | Description |
|---|---|---|
| host | str | Server hostname |
| port | int | Server port |
| tls | bool | Enable TLS (default: False) |
| native_crypto | bool | Use native crypto library (default: True) |
List all colonies on the server.
colonies = client.list_colonies(server_prvkey)| Parameter | Type | Description |
|---|---|---|
| server_prvkey | str | Server private key |
Returns: List of colonies
Create a new colony.
colony = client.add_colony(colonyid, colonyname, server_prvkey)| Parameter | Type | Description |
|---|---|---|
| colonyid | str | Colony ID (derived from colony private key) |
| colonyname | str | Colony name |
| server_prvkey | str | Server private key |
Get colony details.
colony = client.get_colony(colonyname, prvkey)Delete a colony.
client.del_colony(colonyname, server_prvkey)Get colony statistics.
stats = client.stats(colonyname, prvkey)Returns: Colony statistics (process counts, etc.)
Register a new executor.
executor = {
"executorname": "my-executor",
"executorid": executorid,
"colonyname": colonyname,
"executortype": "python-executor"
}
client.add_executor(executor, colony_prvkey)Approve a registered executor.
client.approve_executor(colonyname, executorname, colony_prvkey)Reject a registered executor.
client.reject_executor(colonyname, executorname, colony_prvkey)Remove an executor.
client.remove_executor(colonyname, executorname, colony_prvkey)List all executors in a colony.
executors = client.list_executors(colonyname, prvkey)Returns: List of executor objects
Get details about a specific executor.
executor = client.get_executor(colonyname, executorname, prvkey)Submit a function specification for execution.
from pycolonies import func_spec
spec = func_spec(
func="my_function",
args=["arg1", "arg2"],
colonyname="my_colony",
executortype="python-executor",
maxexectime=60,
maxwaittime=60,
maxretries=3
)
process = client.submit_func_spec(spec, prvkey)Returns: Process object with processid
Assign a waiting process to an executor.
process = client.assign(colonyname, timeout, executor_prvkey)| Parameter | Type | Description |
|---|---|---|
| colonyname | str | Colony name |
| timeout | int | Timeout in seconds (blocks until process available) |
| executor_prvkey | str | Executor private key |
Returns: Process object or None if timeout
Get process details by ID.
process = client.get_process(processid, prvkey)List processes by state.
processes = client.list_processes(colonyname, count, state, prvkey)| Parameter | Type | Description |
|---|---|---|
| state | int | 0=waiting, 1=running, 2=success, 3=failed |
Close a process as successful.
client.close(processid, output, prvkey)| Parameter | Type | Description |
|---|---|---|
| output | list | Output values |
Close a process as failed.
client.fail(processid, errors, prvkey)| Parameter | Type | Description |
|---|---|---|
| errors | list | Error messages |
Set process output without closing.
client.set_output(processid, output, prvkey)Wait for a process to complete.
completed_process = client.wait(process, timeout, prvkey)Remove a process.
client.remove_process(processid, prvkey)Remove all processes in a colony.
client.remove_all_processes(colonyname, prvkey, state=-1)| Parameter | Type | Description |
|---|---|---|
| state | int | -1=all, 0=waiting, 1=running, 2=success, 3=failed |
Submit a workflow (process graph).
from pycolonies import Workflow, func_spec
wf = Workflow(colonyname="my_colony")
f1 = func_spec(func="step1", args=[], colonyname="my_colony",
executortype="python-executor", maxexectime=60, maxwaittime=60)
wf.functionspecs.append(f1)
f2 = func_spec(func="step2", args=[], colonyname="my_colony",
executortype="python-executor", maxexectime=60, maxwaittime=60)
f2.conditions.dependencies.append("step1")
wf.functionspecs.append(f2)
processgraph = client.submit_workflow(wf, prvkey)Get a process graph by ID.
graph = client.get_processgraph(processgraphid, prvkey)List process graphs in a colony.
graphs = client.get_processgraphs(colonyname, count, prvkey, state=None)Get all processes in a workflow.
processes = client.get_processes_for_workflow(processgraphid, colonyname, prvkey, count=100)Remove a process graph.
client.remove_processgraph(processgraphid, prvkey)Remove all process graphs.
client.remove_all_processgraphs(colonyname, prvkey, state=None)Dynamically add a child process to a workflow.
client.add_child(processgraphid, parentprocessid, childprocessid, funcspec, nodename, insert, prvkey)Find a process by node name in a workflow.
process = client.find_process(nodename, processids, prvkey)Append a message to a channel.
client.channel_append(
processid,
channel_name,
sequence,
payload,
prvkey,
in_reply_to=0,
payload_type=""
)| Parameter | Type | Description |
|---|---|---|
| sequence | int | Client-assigned sequence number |
| payload | str/bytes | Message content |
| in_reply_to | int | Optional sequence being replied to |
| payload_type | str | "", "data", "end", or "error" |
Read messages from a channel.
entries = client.channel_read(processid, channel_name, after_seq, limit, prvkey)| Parameter | Type | Description |
|---|---|---|
| after_seq | int | Read messages after this sequence (0 for all) |
| limit | int | Max messages to return |
Returns: List of message entries with sequence, payload, type, inreplyto
Subscribe to channel messages via WebSocket.
# Blocking mode - returns all messages
messages = client.subscribe_channel(processid, channel_name, prvkey, timeout=30)
# Callback mode - streaming
def on_message(entries):
for entry in entries:
print(entry['payload'].decode())
return True # False to stop
client.subscribe_channel(processid, channel_name, prvkey, timeout=30, callback=on_message)Add a blueprint definition (requires colony owner key).
definition = {
"kind": "MyResource",
"metadata": {
"name": "my-resource-def",
"colonyname": colonyname
},
"spec": {
"names": {"kind": "MyResource"}
}
}
client.add_blueprint_definition(definition, colony_prvkey)Get a blueprint definition by name.
definition = client.get_blueprint_definition(colonyname, name, prvkey)List all blueprint definitions.
definitions = client.get_blueprint_definitions(colonyname, prvkey)Remove a blueprint definition.
client.remove_blueprint_definition(colonyname, name, colony_prvkey)Add a blueprint instance.
blueprint = {
"kind": "MyResource",
"metadata": {
"name": "my-instance",
"colonyname": colonyname
},
"handler": {
"executortype": "my-reconciler"
},
"spec": {
"replicas": 3,
"image": "nginx:latest"
}
}
client.add_blueprint(blueprint, prvkey)Get a blueprint by name.
blueprint = client.get_blueprint(colonyname, name, prvkey)List blueprints, optionally filtered.
blueprints = client.get_blueprints(colonyname, prvkey, kind=None, location=None)Update a blueprint's spec.
blueprint['spec']['replicas'] = 5
client.update_blueprint(blueprint, prvkey, force_generation=False)Update a blueprint's status (called by reconciler).
status = {
"replicas": 5,
"ready": True,
"lastSeen": "2024-01-01T12:00:00Z"
}
client.update_blueprint_status(colonyname, name, status, prvkey)Trigger reconciliation for a blueprint.
process = client.reconcile_blueprint(colonyname, name, prvkey, force=False)Get change history for a blueprint.
history = client.get_blueprint_history(blueprintid, prvkey, limit=None)Remove a blueprint.
client.remove_blueprint(colonyname, name, prvkey)Add a cron job.
from pycolonies import Workflow
wf = Workflow(colonyname=colonyname)
# ... add function specs to workflow
client.add_cron(cronname, cronexpr, wait, wf, colonyname, prvkey)| Parameter | Type | Description |
|---|---|---|
| cronexpr | str | Cron expression (e.g., "0 * * * *") |
| wait | bool | Wait for previous run to complete |
Get a cron job by ID.
cron = client.get_cron(cronid, prvkey)List cron jobs.
crons = client.get_crons(colonyname, count, prvkey)Manually trigger a cron job.
process = client.run_cron(cronid, prvkey)Delete a cron job.
client.del_cron(cronid, prvkey)Add a generator.
client.add_generator(generator, prvkey)Get a generator by ID.
generator = client.get_generator(generatorid, prvkey)List generators.
generators = client.get_generators(colonyname, prvkey, count=100)Remove a generator.
client.remove_generator(generatorid, prvkey)Add a user to a colony.
user = {
"colonyname": colonyname,
"userid": userid,
"name": "username",
"email": "user@example.com",
"phone": ""
}
client.add_user(user, server_prvkey)List users in a colony.
users = client.get_users(colonyname, server_prvkey)Remove a user.
client.remove_user(colonyname, name, server_prvkey)Register a function for an executor.
client.add_function(colonyname, executorname, funcname, prvkey)List all functions in a colony.
functions = client.get_functions_by_colony(colonyname, prvkey)List functions for a specific executor.
functions = client.get_functions_by_executor(colonyname, executorname, prvkey)Add an attribute to a running process.
attr = client.add_attribute(processid, key, value, prvkey)Get an attribute by ID.
attr = client.get_attribute(attributeid, prvkey)Add a log message to a process.
client.add_log(processid, message, prvkey)Get logs for a process.
logs = client.get_process_log(colonyname, processid, count, since, prvkey)| Parameter | Type | Description |
|---|---|---|
| count | int | Max log entries to return |
| since | int | Timestamp to start from |
Get logs for an executor.
logs = client.get_executor_log(colonyname, executorid, count, since, prvkey)Upload a file to storage.
client.upload_file(colonyname, filepath, label, keeplocal, prvkey)Upload data directly to storage.
client.upload_data(colonyname, data, filename, label, prvkey)Download a file from storage.
client.download_file(colonyname, fileid, localpath, prvkey)Download data from storage.
data = client.download_data(colonyname, fileid, prvkey)Get file metadata.
file = client.get_file(colonyname, fileid, prvkey)List files by label.
files = client.get_files(colonyname, label, prvkey)List file labels.
labels = client.get_file_labels(colonyname, prvkey, name="", exact=False)Delete a file from storage.
client.delete_file(colonyname, fileid, prvkey)Sync files between local and remote storage.
client.sync(colonyname, label, dir, keeplocal, prvkey)Create a snapshot of files.
client.create_snapshot(colonyname, label, name, prvkey)List snapshots.
snapshots = client.get_snapshots(colonyname, prvkey)Get a snapshot by name.
snapshot = client.get_snapshot_by_name(colonyname, name, prvkey)Get a snapshot by ID.
snapshot = client.get_snapshot_by_id(colonyname, snapshotid, prvkey)| State | Value | Description |
|---|---|---|
| WAITING | 0 | Process waiting for executor assignment |
| RUNNING | 1 | Process assigned and executing |
| SUCCESS | 2 | Process completed successfully |
| FAILED | 3 | Process failed |
from pycolonies import ColoniesError, ColoniesConnectionError
try:
process = client.assign(colonyname, 10, prvkey)
except ColoniesConnectionError as e:
print(f"Connection error: {e}")
except ColoniesError as e:
print(f"API error: {e}")