Call a function in BEAM, returning the result. Thread-safe.

async def call(
    module: str | Atom | Term,
    function: str | Atom | Term,
    args: Iterable[object],
    *,
    node: str | Atom | Term | None = None,
    result_encoding_opts: EncodingOpts | None = None
) -> Any

module, function and node can be given as strings, atoms, or snex.Term that resolves either to a string, or an atom. They will be converted to atoms on the BEAM side.

The function will be called in a new process on the BEAM side.

Args

• module - The module to call the function from
• function - The function to call
• args - The arguments to pass to the function
• node - The node to call the function on; defaults to the current node
• result_encoding_opts - The options to use for encoding the result. They will override the interpreter's :encoding_opts.

Call a function in BEAM without waiting for a response. Thread-safe.

async def cast(
    module: str | Atom | Term,
    function: str | Atom | Term,
    args: Iterable[object],
    *,
    node: str | Atom | Term | None = None
) -> None

module, function and node can be given as strings, atoms, or snex.Term that resolves either to a string, or an atom. They will be converted to atoms on the BEAM side.

The function will be called in a new process on the BEAM side.

Args

• module - The module to call the function from
• function - The function to call
• args - The arguments to pass to the function
• node - The node to call the function on; defaults to the current node

Run an IO loop for a reader/writer pair of asyncio streams. Thread-safe.

async def io_loop_for_connection(
    sub_reader: asyncio.StreamReader,
    sub_writer: asyncio.StreamWriter
) -> None

Can be used to connect a subprocess to the main Snex loop. See snex.serve for an example.

Closes the writer on the return path and waits until it's closed.

Examples

await asyncio.start_unix_server(snex.io_loop_for_connection, socket_path)

Send data to a BEAM process. Thread-safe.

async def send(to: object, data: object) -> None

Args

• to - The BEAM process to send the data to; can be any identifier that can be used as destination in Kernel.send/2
• data - The data to send; will be encoded and sent to the BEAM process

Examples

# Send to a registered process with an atom name
snex.send("registered_name", "hello to a registered process!")

# Send to a remote process with a tuple name and node
snex.send(("myname", "mynode@localhost"), "hello to a remote process!")

# Use `snex.Term` passed through with `Snex.pyeval/4`
# e.g. `Snex.pyeval(env, "snex.cast(self, 'hello self!')", %{"self" => self()}`)
snex.send(self, "hello self!")

Run Snex server loop as a context manager.

def serve(
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter
) -> AsyncGenerator[None, None]

The server loop is necessary to use snex.call/snex.cast from external Python processes. Usually, you'll want to run it in a subprocess, and connect the other end of the reader/writer pair to the main Snex process.

Examples

import asyncio
from asyncio import start_unix_server
from concurrent.futures import ProcessPoolExecutor

import snex


async def in_subprocess_loop(sock_path: str) -> int:
    reader, writer = await asyncio.open_unix_connection(sock_path)
    async with snex.serve(reader, writer):
        return await snex.call("Elixir.System", "pid", [])


def in_subprocess(sock_path: str) -> int:
    return asyncio.run(in_subprocess_loop(sock_path))


async def main():
    sock_path = "/tmp/snex.sock"
    loop = asyncio.get_running_loop()
    async with await start_unix_server(snex.io_loop_for_connection, sock_path):
        with ProcessPoolExecutor() as pool:
            return await loop.run_in_executor(pool, in_subprocess, sock_path)

Run Snex server loop until canceled.

async def serve_forever(
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    *,
    on_ready: Callable[[], None] | None = None
) -> None

Similar to async with snex.serve(reader, writer).

Set a custom encoding function for objects that are not supported by default.

def set_custom_encoder(
    encoder_fun: Callable[[object], object]
) -> None

Args

• encoder_fun - The function to use to encode the objects