renderflow

Official Python client for the RenderFlow API.

Installation

pip install pulze-renderflow

Quick Start

from renderflow import RenderFlow

rf = RenderFlow(api_key="your_api_key")

jobs = rf.jobs.list()
print(f"Found {len(jobs)} jobs")

Submitting a Job

from renderflow import RenderFlow, IPublicJobCreate
from renderflow.models import ISoftwareValue, IEngineValue, SoftwareId, EngineId

rf = RenderFlow(api_key="your_api_key")

job = rf.jobs.create(IPublicJobCreate(
    name="Exterior Shot",
    file="C:/projects/scene.max",
    type="3dsmax.render",
    host=ISoftwareValue(id=SoftwareId._3dsmax, name="3ds Max", version="2026"),
    engine=IEngineValue(id=EngineId.vray, name="V-Ray", version="7.20.04"),
    frame="1-100",
    resolution="1920x1080",
    priority=75,
    status="pending"
))

rf.jobs.start(job.id)

Managing Jobs

# Get a specific job
job = rf.jobs.get("job_id")

# Update job settings
rf.jobs.update("job_id", IPublicJobUpdate(
    priority=100,
    limit=5,
    max_batch_size=10
))

# Control jobs
rf.jobs.stop("job_id")
rf.jobs.reset("job_id")
rf.jobs.archive("job_id")
rf.jobs.delete("job_id")

Tasks

# List tasks for a job (paginated)
tasks = rf.tasks.list("job_id", page=1, limit=50)

# Get a specific task
task = rf.tasks.get("task_id")

# Get task logs
logs = rf.tasks.logs("task_id", offset=0, limit=500)

# Get task thumbnail
thumbnail = rf.tasks.thumbnail("task_id")

# Stream task logs live (SSE)
stream = rf.tasks.stream_logs("task_id", lambda line: print(line))
# ... later
stream.close()

Nodes

# List all nodes
nodes = rf.nodes.list()

# Get node details
node = rf.nodes.get("node_id")

# Update node status
rf.nodes.update_status("node_id", "suspended")

# Assign to a pool
rf.nodes.update_pool("node_id", "pool_id")

# Get node utilization
util = rf.nodes.utilization("node_id")

# Get benchmark rankings
benchmarks = rf.nodes.benchmarks("vray")

Pools

pools = rf.pools.list()
pool = rf.pools.get("pool_id")

Users

users = rf.users.list()
user = rf.users.get("user_id")

Errors

errors = rf.errors.list()
job_errors = rf.errors.by_job("job_id")
node_errors = rf.errors.by_node("node_id")

Icons

# Icon bytes for a job type or a software (e.g. write to a file)
job_icon = rf.icons.job("vray")
software_icon = rf.icons.software("software_id")

Real-time Events

Subscribe to live updates using Server-Sent Events:

# Listen to all job changes
listener = rf.jobs.on(lambda event: print(
    event.type,      # "insert" | "update" | "delete" | "replace"
    event.document,  # full job document
    event.updated,   # changed fields (on update)
    event.time       # event timestamp
))

# Listen to task events for a specific job
task_listener = rf.tasks.on("job_id", lambda event: print(
    f"Task {event.document['_id']}: {event.document['status']}"
))

# Listen to node events
node_listener = rf.nodes.on(lambda event: print(
    f"Node {event.document['name']}: {event.document['status']}"
))

# Error handling
listener = rf.jobs.on(
    lambda event: print(event),
    lambda error: print(f"Connection error: {error}")
)

# Stop listening
listener.close()

# Also works as a context manager
with rf.jobs.on(on_event) as listener:
    import time
    time.sleep(60)

Service Info

info = rf.info.get()
print(f"RenderFlow {info.version} ({info.node_type})")