DeviceFactory

Class-based factory for creating and managing Moler device instances, with support for named configuration, state machine transitions, and lifecycle management.

Overview

DeviceFactory is the central registry for all Moler device objects. It creates devices from configuration or explicit class references, caches them by name, and provides lifecycle operations such as removal and bulk teardown.

moler.device.device.DeviceFactory

All methods are class methods — you never instantiate DeviceFactory directly.

Devices are singletons within a DeviceFactory. Requesting the same name twice returns the same object.

Factory methods

get_device

DeviceFactory.get_device(
    name=None,
    device_class=None,
    connection_desc=None,
    connection_hops=None,
    initial_state=None,
    establish_connection=True,
    lazy_cmds_events=False,
    io_connection=None,
    additional_params=None,
) -> AbstractDevice

Return a device instance. If the device has been created before under name, the cached instance is returned. Otherwise a new device is created and cached.

Provide either name or device_class, not both.

name · str · default: "None"

Name of a device pre-defined in Moler’s configuration. Moler looks up the class and connection parameters from that configuration entry.

device_class · str · default: "None"

Fully-qualified Python class name, e.g. 'moler.device.unixlocal.UnixLocal'. Used when creating a device without a named configuration entry.

connection_desc · dict · default: "None"

Dict describing the I/O connection, with at minimum an 'io_type' key (e.g., {'io_type': 'terminal', 'variant': 'threaded'}). Ignored if the device is already created or io_connection is provided.

connection_hops · dict · default: "None"

State machine hop configuration used to define multi-step state transitions (e.g., reaching a nested SSH session).

initial_state · str · default: "None"

Name of the state the device should enter after creation, e.g. 'UNIX_REMOTE'. If None the device’s own default initial state is used.

establish_connection · boolean · default: "True"

If True, open the connection and transition to initial_state immediately. Set False to create the device object without connecting.

lazy_cmds_events · boolean · default: "False"

If True, load command and event classes on first use rather than at device initialization. Useful to reduce startup time when many devices are created.

io_connection · AbstractMolerConnection · default: "None"

Supply an already-open connection object directly. Ignored if the device is already cached.

additional_params · dict · default: "None"

Extra keyword arguments forwarded to the device constructor for device-specific options.

Returns the device instance.

Raises:

• WrongUsage — if neither name nor device_class is given, or if both are given.
• KeyError — if name is not found in the configuration.

create_all_devices

DeviceFactory.create_all_devices(ignore_exception=False) -> None

Instantiate every device defined in Moler’s configuration. Useful in test setup.

ignore_exception · boolean · default: "False"

If False, the first failure raises and stops creation of subsequent devices. If True, failures are logged as warnings and creation continues.

remove_device

DeviceFactory.remove_device(
    name=None,
    device=None,
) -> None

Close and remove a single device. All commands and events attached to it are finished.

name · str · default: "None"

Name of the device to remove.

device · AbstractDevice · default: "None"

Device object to remove. Use this when you hold a reference but not the name.

Provide either name or device. Raises WrongUsage if neither is given.

remove_all_devices

DeviceFactory.remove_all_devices(clear_device_history=False) -> None

Close and remove all devices. Also clears the devices configuration.

clear_device_history · boolean · default: "False"

If True, completely resets the factory’s internal name registry so the same names can be reused. Use with caution — existing log files may be overwritten if devices with the same names are recreated.

was_any_device_deleted

DeviceFactory.was_any_device_deleted() -> bool

Returns True if any device has been removed since the factory was last cleared. Useful in test frameworks to detect unexpected device teardown.

get_cloned_device

DeviceFactory.get_cloned_device(
    source_device,
    new_name,
    initial_state=None,
    establish_connection=True,
    lazy_cmds_events=False,
    io_connection=None,
    additional_params=None,
) -> AbstractDevice

Create a new device with the same class and connection parameters as an existing one.

source_device · str | AbstractDevice · required

Reference to the device to clone — either a device object or a name string.

new_name · str · required

Name for the cloned device.

initial_state · str · default: "None"

Initial state for the clone. Defaults to the source device’s current state.

get_devices_by_type

DeviceFactory.get_devices_by_type(device_type) -> list

Return all cached device instances that are instances of device_type.

device_type · type | None · required

A device class to filter by. Pass None to return all devices.

Device instance methods

The following methods are defined on the device objects returned by get_device().

get_cmd

device.get_cmd(
    cmd_name,
    cmd_params=None,
    check_state=True,
    for_state=None,
) -> Command

Return a Command instance bound to this device’s connection.

cmd_name · str · required

Class name (without package path) of the command, e.g. 'ls', 'ping'.

cmd_params · dict · default: "None"

Parameters forwarded to the command constructor.

check_state · boolean · default: "True"

If True, the device state is verified before executing the command.

for_state · str · default: "None"

Return a command appropriate for this state rather than the current state.

get_event

device.get_event(
    event_name,
    event_params=None,
    check_state=True,
    for_state=None,
) -> Event

Return an Event instance bound to this device’s connection.

event_name · str · required

Class name of the event (without package path).

event_params · dict · default: "None"

Parameters forwarded to the event constructor.

check_state · boolean · default: "True"

Verify device state before creating the event.

for_state · str · default: "None"

Return an event for this state rather than the current state.

goto_state

device.goto_state(
    state,
    timeout=-1,
    rerun=0,
    send_enter_after_changed_state=False,
    log_stacktrace_on_fail=True,
    keep_state=False,
    timeout_multiply=1.0,
    sleep_after_changed_state=0.5,
) -> None

Transition the device to the named state via its state machine.

state · str · required

Target state name, e.g. 'UNIX_LOCAL', 'UNIX_REMOTE'.

timeout · float · default: "-1"

Seconds before the transition is aborted. -1 uses the default timeout.

rerun · int · default: "0"

Number of retries if the state transition fails.

keep_state · boolean · default: "False"

If True and the device leaves the target state unexpectedly, it attempts to return to it automatically.

enable_logging / disable_logging

device.enable_logging() -> None
device.disable_logging() -> None

Enable or disable logging of incoming data on the device’s I/O connection.

set_logging_suffix

device.set_logging_suffix(suffix) -> None

Append a suffix to the device’s logger names. Useful when multiple devices share the same base name and you need to distinguish their log output.

suffix · str | None · required

Suffix string to append. Pass None to remove any existing suffix.

Usage examples

Create from named config

from moler.device.device import DeviceFactory

# 'MyServer' must be defined in moler.yml or programmatic config
device = DeviceFactory.get_device(name='MyServer')
print(device.current_state)

Use device.run(cmd_name, cmd_params={...}) as a one-liner shortcut — it creates the command, calls it synchronously, and returns the result. Use device.start(cmd_name, cmd_params={...}) to start it in the background and get the command object back to await_done() later.