Connection configuration

Configure named connections, set default variants per IO type, and register custom connection implementations with Moler’s ConnectionFactory.

Moler separates what connection you want (io type) from how it is implemented (variant). This separation lets you swap threading models or IO frameworks without changing any test code.

Connection type and variant

Every connection in Moler is identified by a pair:

• io_type — the transport kind: terminal, tcp, memory, sshshell, terminal_no_fork
• variant — the concurrency implementation: threaded, asyncio, asyncio-in-thread, multi-threaded, single-threaded

For example, a tcp/threaded connection is a TCP socket driven by a background thread; a tcp/asyncio connection is the same TCP socket driven by asyncio.

Built-in connection types

io_type	Variants	Platform	Description
terminal	threaded, multi-threaded, single-threaded, asyncio, asyncio-in-thread	Unix only	Spawns a local shell via a PTY
terminal_no_fork	threaded, multi-threaded, single-threaded	Unix only	Local shell without forking
tcp	threaded, asyncio, asyncio-in-thread	All	TCP socket
memory	threaded	All	In-process FIFO buffer (useful for unit tests)
sshshell	threaded	All	SSH shell via Paramiko

terminal and terminal_no_fork are not available on Windows. Use sshshell instead.

Default variants set automatically at import time:

• terminal → threaded
• sshshell → threaded

No default is set for tcp or memory — you must configure one explicitly.

Setting a default variant in YAML

The IO_TYPES section of your config file sets the default variant per io_type. This is the most common way to configure the concurrency model for a connection type:

IO_TYPES:
  default_variant:
    tcp: threaded

Once configured, code that calls get_connection(io_type='tcp', ...) does not need to specify the variant.

Defining named connections in YAML

The NAMED_CONNECTIONS section assigns human-readable names to fully-specified connections, including all constructor parameters:

NAMED_CONNECTIONS:
  net_1:
    io_type: tcp
    host: localhost
    port: 5671
  net_2:
    io_type: tcp
    host: localhost
    port: 5672

IO_TYPES:
  default_variant:
    tcp: threaded

Named connections can then be retrieved by name without specifying any parameters at the call site:

from moler.connection_factory import get_connection

conn = get_connection(name='net_1')
with conn.open():
    ...

get_connection API

from moler.connection_factory import get_connection

# By name (name must be defined in NAMED_CONNECTIONS or via define_connection)
conn = get_connection(name='net_1')

# By io_type (variant taken from default_variant config)
conn = get_connection(io_type='tcp', host='localhost', port=5671)

# By io_type + explicit variant
conn = get_connection(io_type='tcp', variant='threaded', host='localhost', port=5671)

# Terminal connection (no additional params needed)
terminal = get_connection(io_type='terminal', variant='threaded')

You may provide either name or io_type, but not both. If variant is omitted, the value from default_variant configuration is used.

from moler.connection_factory import get_connection
import time
from moler.cmd.unix.ping import Ping

terminal = get_connection(io_type='terminal', variant='threaded')
with terminal.open():
    ping_cmd = Ping(connection=terminal.moler_connection,
                    destination='www.google.com', options='-w 6')
    ping_cmd.start()
    time.sleep(3)
    ping_stats = ping_cmd.await_done(timeout=4)
    print(f"packet_loss={ping_stats['packet_loss']}")

Configuring connections via Python code

You can configure connections in Python without a YAML file. This is equivalent to what load_config does internally.

from moler.config import connections as conn_cfg

# Set default variant for tcp connections
conn_cfg.set_default_variant(io_type='tcp', variant='threaded')

# Define named connections
conn_cfg.define_connection(name='net_1', io_type='tcp', host='localhost', port=5671)
conn_cfg.define_connection(name='net_2', io_type='tcp', host='localhost', port=5672)

Registering a custom connection type

ConnectionFactory.register_construction installs a callable as the factory for a given type/variant pair. Use this to plug in a custom IO implementation.

from moler.connection_factory import ConnectionFactory

def my_tcp_conn(host='localhost', port=8080, name=None, **kwargs):
    from moler.moler_connection_for_single_thread_runner import MolerConnectionForSingleThreadRunner
    from moler.io.raw.tcp import ThreadedTcp
    mlr_conn = MolerConnectionForSingleThreadRunner(
        encoder=lambda data: data.encode('utf-8'),
        decoder=lambda data: data.decode('utf-8'),
        name=name,
    )
    return ThreadedTcp(moler_connection=mlr_conn, host=host, port=port)

ConnectionFactory.register_construction(
    io_type='tcp',
    variant='my-custom',
    constructor=my_tcp_conn,
)

# Now usable via get_connection
from moler.connection_factory import get_connection
conn = get_connection(io_type='tcp', variant='my-custom', host='10.0.0.1', port=9000)

The constructor must be any callable that returns a fully wired connection object. It receives the keyword arguments passed to get_connection.

To check which variants are registered for a given io_type:

from moler.connection_factory import ConnectionFactory

ConnectionFactory.available_variants(io_type='tcp')
# ['threaded', 'asyncio', 'asyncio-in-thread']

Combined YAML example

A configuration that defines named connections and sets TCP defaults:

NAMED_CONNECTIONS:
  www_svr1:
    io_type: tcp
    host: 10.20.30.41
    port: 80
  www_svr2:
    io_type: tcp
    host: 10.20.30.42
    port: 80
  www_svr3:
    io_type: tcp
    host: 10.20.30.43
    port: 80

IO_TYPES:
  default_variant:
    tcp: threaded

Client code retrieves connections by name:

from moler.config import load_config
from moler.connection_factory import get_connection

load_config(config='/absolute/path/to/connections.yml')

svr1 = get_connection(name='www_svr1')
svr2 = get_connection(name='www_svr2')

Naming connections after their logical role (main_dns_server, backup_ntp_server) rather than their address makes test code more readable and easier to maintain when addresses change.