Flux Python Basics

Importing the flux Python package

Note

The flux package which is used to interact with Flux cannot be installed into a virtual environment with pip or conda.

Flux's Python bindings are available with any installation of Flux. When running in a Flux instance, Flux will export the PYTHONPATH environment variable so that Python processes can import the flux package by the usual import mechanism (import flux).

If you want to import the package from outside of a Flux instance, running /path/to/flux env | grep PYTHONPATH in your shell will show you what PYTHONPATH would be set to if you were running in a Flux instance built/installed at /path/to/flux.

Note however that if you import the flux package from outside a Flux instance, you will need to specify a Flux instance to communicate with, or most of the package's operations will fail.

Additional Examples

In addition to the examples given in these pages, there is a set of detailed examples maintained on the workflow examples page.

The Flux class

Almost all of the functionality of the flux package revolves around flux.Flux objects, often called "Flux handles", which represent a connection to a Flux instance. It is possible to simultaneously have multiple connections open to the same Flux instance, or to multiple Flux instances.

Note

Flux handles are not thread-safe and should not be shared between threads.

In addition to the methods defined on flux.Flux objects, the flux package also provides a number of functions which accept flux.Flux objects as arguments, such as the functions described here.

class flux.core.handle.Flux(url=_flux._core.ffi.NULL, flags=0, handle=None)

The general Flux handle class, create one of these to connect to the nearest enclosing flux instance

Example

>>> flux.Flux()
<flux.core.Flux object at 0x...>
add_watcher(watcher)

Add a reference to a watcher so it avoids garbage collection

close()

The underlying flux handle is automatically closed when a Flux instance is deconstructed. Prevent users from manually closing, the handle, leading to a double free.

del_watcher(watcher)

Remove ref to watcher so it is eligible for garbage collection

event_create(topic, payload=None)

Create a new event message.

Parameters

  • topic -- A string, the event's topic

  • payload -- If a string, the payload is used unmodified, if it is another type json.dumps() is used to stringify it

event_send(topic, payload=None)

Create and send a new event in one step

event_subscribe(topic)

Subscribe to events

Parameters

topic (str, bytes, or unicode) -- The event's topic to subscribe to

Raises

  • EnvironmentError -- if the topic is None or NULL

  • TypeError -- if the topic is not a str, bytes, or unicode

log(level, fstring)

Log to the flux logging facility

Parameters

  • level -- A syslog log-level, check the syslog module for possible values

  • fstring -- A string to log, C-style formatting is not supported

classmethod raise_if_exception()

Re-raise any class global exception if set

If a global exception is currently set for the Flux handle class, re-raise it and reset the exception state to None.

The exception is raised from None to preserve the original stack trace.

reactor_run(reactor=None, flags=0)

Run reactor associated with this Flux handle or reactor argument if it is provided. Sets a signal watcher for SIGINT to return from the reactor on Ctrl-C, and raise KeyboardInterrupt.

classmethod reactor_running()

Return True if this thread is running the Flux reactor

recv(type_mask=<bound method ? of <flux.core.inner.Core object>>, match_tag=<bound method ? of <flux.core.inner.Core object>>, topic_glob=None, flags=0)

Receive a message, returns a flux.Message containing the result or None

respond(message, payload=None)

Respond to a flux rpc

Parameters

  • message (Message) -- The message to respond to

  • payload (None, str, bytes, unicode, or json-serializable) -- The (optional) payload to include in the response

rpc(topic, payload=None, nodeid=<bound method ? of <flux.core.inner.Core object>>, flags=0)

Create a new RPC object

send(message, flags=0)

Send a pre-constructed flux message

classmethod set_exception(exception)

Set a global, per-thread exception for Flux

This class method allows Python callbacks called from the Flux reactor to set a global exception which can be re-thrown after the return to Python (when reactor_run() returns). This is implemented as a class attribute since the Flux handle object which is available in a Python callback from C will be a different instantiation than the Flux handle object which started the reactor (with the same underlying flux_t however)

Parameters

exception (Exception) -- A reference to the exception thrown.

Returns

The previously set exception, or None

Return type

Exception

The Flux reactor

Content coming soon.