Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/nokia/moler/llms.txt

Use this file to discover all available pages before exploring further.

One of Moler’s core design goals is to be IO-agnostic. The same Command and Event objects work identically whether the underlying transport is a local terminal, a TCP socket, an SSH shell, or an in-memory buffer used for testing. This is achieved by splitting the connection into two layers:
  • External IO (io_connection) — handles actual bytes: opening/closing the channel, reading, writing. Examples: ThreadedTerminal, ThreadedTcp, ThreadedSshShell.
  • Moler connection (moler_connection) — an AbstractMolerConnection that dispatches decoded data to all registered ConnectionObservers (commands and events) and exposes sendline() / send().
Commands and events only interact with the moler connection. The external IO layer is transparent to them.

​
Getting a connection

The get_connection() function is the standard way to obtain a connection object: 
from moler.connection_factory import get_connection

# Terminal connection (spawns a local shell)
terminal = get_connection(io_type='terminal', variant='threaded')

# TCP connection
tcp_conn = get_connection(io_type='tcp', variant='threaded',
                          host='localhost', port=2222)

# SSH shell connection
ssh_conn = get_connection(io_type='sshshell', variant='threaded',
                          host='192.168.1.1', login='admin', password='secret')

# In-memory FIFO buffer (useful for unit tests)
mem_conn = get_connection(io_type='memory', variant='threaded')
Provide either io_type or name (a named connection from config), but not both.

​
Using a connection as a context manager

Connections implement the context manager protocol. Use with connection.open(): to ensure the connection is always closed, even if an exception occurs: 
import time
from moler.cmd.unix.ping import Ping
from moler.connection_factory import get_connection

host = 'www.google.com'
terminal = get_connection(io_type='terminal', variant='threaded')

with terminal.open():
    ping_cmd = Ping(connection=terminal.moler_connection,
                    destination=host, options="-w 6")
    ping_cmd.start()
    time.sleep(3)
    ping_stats = ping_cmd.await_done(timeout=4)
    print("packet_loss={}, time_avg={} [{}]".format(
        ping_stats['packet_loss'],
        ping_stats['time_avg'],
        ping_stats['time_unit']))
Pass terminal.moler_connection (not terminal itself) to command and event constructors. Commands and events operate on the moler-connection layer, not the IO layer directly.

​
Connection types

io_typeDescriptionPlatform
terminalSpawns a local PTY shell processUnix only
terminal_no_forkTerminal without forkingUnix only
tcpRaw TCP socketAll
sshshellSSH shell session via ParamikoAll
memoryIn-memory FIFO buffer (for testing)All

​
Variants (execution model)

Each io_type can have multiple variants that differ in their concurrency model:
VariantDescription
threaded / multi-threadedUses threads for I/O. Default for terminal and sshshell.
single-threadedUses a single-thread runner (MolerConnectionForSingleThreadRunner).
asyncioUses asyncio event loop (available for terminal and tcp).
asyncio-in-threadRuns the asyncio loop in a dedicated thread.
The default variant for terminal and sshshell is threaded. You can override defaults in config: 
CONNECTIONS:
  default_variant:
    terminal: threaded
    tcp: asyncio

​
Building a connection manually

For full control, build the IO and moler layers yourself: 
from moler.threaded_moler_connection import ThreadedMolerConnection
from moler.io.raw.terminal import ThreadedTerminal

mlr_conn = ThreadedMolerConnection()
terminal_connection = ThreadedTerminal(moler_connection=mlr_conn)
This is equivalent to get_connection(io_type='terminal', variant='multi-threaded') but without going through the factory.

​
Named connections

Connections can be named in the config file and referenced by name: 
CONNECTIONS:
  www_svr1:
    io_type: tcp
    host: 10.20.30.41
    port: 80

conn = get_connection(name='www_svr1')

​
Registering custom connections

ConnectionFactory is a plugin system. Register any callable that produces a connection object for a given (io_type, variant) pair: 
from moler.connection_factory import ConnectionFactory

def my_custom_conn(host, port, name=None):
    # build and return a fully operable connection
    ...

ConnectionFactory.register_construction(
    io_type="my_protocol",
    variant="threaded",
    constructor=my_custom_conn,
)

# Now usable via the standard API
conn = get_connection(io_type='my_protocol', variant='threaded',
                      host='10.0.0.1', port=9999)
Use ConnectionFactory.available_variants(io_type) to inspect which variants are currently registered for a given IO type.

​
Connection responsibilities

The moler-connection (AbstractMolerConnection) is responsible for:
  • Sending — send(data) and sendline(data) write encoded bytes through the external IO layer.
  • Receiving and dispatching — incoming bytes are decoded and delivered to every subscribed ConnectionObserver via their data_received() method.
  • Encoding/decoding — configurable encoder/decoder pair (defaults to UTF-8 for most types; terminal uses VT100-cleaned Unicode).
  • Logging — raw I/O is optionally written to device log files.

Commands

How commands use the moler connection to send and receive

Devices

How devices manage their connection lifecycle

Build docs developers (and LLMs) love

Get started for freeTalk to us