nexusflow-sdk 0.3.0

Python SDK for NexusFlow model editing and simulation workflows

nexusflow-sdk

nexusflow-sdk is a Python SDK for NexusFlow model editing, simulation execution, streaming result retrieval, and runtime command control.

Features

• Token-based authentication against the NexusFlow backend
• Create models for supported workspace types and open existing models
• Add, remove, connect, and configure components in the model graph
• Save model snapshots and start simulation jobs
• Poll or stream logs, plots, tables, inspect data, and container messages
• Send runtime commands to components during online or transient execution

Requirements

• Python 3.10 or later
• A reachable NexusFlow service endpoint
• A NexusFlow account that can obtain access tokens

Install

Install from a built wheel or from PyPI after publication:

pip install nexusflow-sdk

For local development in this repository:

pip install -e .

Quick Start

from nexusflow_sdk import NexusFlowClient

client = NexusFlowClient(
    base_url="http://localhost:8000",
    username="admin",
    password="your-password",
)

model = client.create_model(
    name="sdk-demo-model",
    workspace_type="lps_online",
    description="Created by nexusflow-sdk",
)

source = model.add_component("Source", position=(0, 0))
pipe = model.add_component("Pipe", position=(240, 0))
load = model.add_component("Load", position=(480, 0))

model.connect(source, "0", pipe, "0")
model.connect(pipe, "1", load, "0")

model.save()
job = model.run(timeout=120, image="nexusflow")

for message in job.stream_results(timeout=2):
    print(message.get("type"), message.get("key"))

Public API

The package currently exports these main entry points:

• NexusFlowClient
• Model
• Component
• Connection
• Job
• MetaService
• ResultStream
• LogMessage
• PlotMessage
• TableMessage
• InspectMessage
• ContainerMessage
• NexusFlowError
• AuthError
• APIError
• NotFoundError
• ValidationError

Examples

• examples/quickstart.py: create, save, run, and fetch results
• examples/transient_streaming.py: stream transient online results and send commands during execution
• examples/run_existing_model.py: open an existing model by ID and run it

Release Validation

Build and metadata check:

python -m build
python -m twine check dist/*

Minimal install smoke test with the built wheel:

python -m venv .smoke-venv
.smoke-venv\Scripts\python -m pip install dist\nexusflow_sdk-0.1.2-py3-none-any.whl
.smoke-venv\Scripts\python -c "from nexusflow_sdk import NexusFlowClient, Model, Job, ResultStream; print('smoke ok')"

Release Workflow

If you want to publish a new version yourself, the repository now includes a reusable PowerShell script:

.\scripts\release.ps1 -Repository pypi -Token "pypi-..."

Recommended process:

1. Update version in pyproject.toml.
2. Run the release script from the nexusflow-sdk directory.
3. Wait for PyPI index propagation if the published install smoke test fails on the first try.

The script performs these steps automatically:

• clean dist/
• python -m build
• python -m twine check dist/*
• python -m unittest tests.test_public_api
• install the built wheel into a temporary virtual environment for a local smoke test
• upload to PyPI or TestPyPI
• create a fresh temporary virtual environment and install the published package for a post-publish smoke test

Examples:

# Full publish to PyPI
.\scripts\release.ps1 -Repository pypi -Token "pypi-..."

# Full publish to TestPyPI
.\scripts\release.ps1 -Repository testpypi -Token "pypi-..."

# Local validation only, without upload
.\scripts\release.ps1 -SkipUpload -SkipSmokeTest

You can also avoid passing tokens on the command line by using environment variables:

$env:PYPI_TOKEN = "pypi-..."
.\scripts\release.ps1 -Repository pypi

License

This package is currently distributed as proprietary software.