Easy and efficient Python interop for Elixir.

defmodule SnexTest.NumpyInterpreter do
  use Snex.Interpreter,
    pyproject_toml: """
    [project]
    name = "my-numpy-project"
    version = "0.0.0"
    requires-python = "==3.10.*"
    dependencies = ["numpy>=2"]
    """
end

{:ok, interpreter} = SnexTest.NumpyInterpreter.start_link(init_script: "import numpy as np")

{:ok, 6.0} =
  Snex.pyeval(interpreter, """
    width, height = await Elixir.Enum.map([3, 3], mapper)
    matrix = np.fromfunction(lambda i, j: (-1) ** (i + j), (width, height), dtype=int)

    return np.linalg.norm(matrix)
    """, %{"mapper" => &(&1 * 2)})

Custom Interpreter

You can define your Python project settings using use Snex.Interpreter in your module.

Set a required Python version and any dependencies - both the Python binary and the dependencies will be fetched & synced at compile time with uv, and put into the _build/$MIX_ENV/snex directory.

defmodule SnexTest.NumpyInterpreter do
  use Snex.Interpreter,
    pyproject_toml: """
    [project]
    name = "my-numpy-project"
    version = "0.0.0"
    requires-python = "==3.10.*"
    dependencies = ["numpy>=2"]
    """
end

Modules using Snex.Interpreter have to be start_linked to be used. Each Snex.Interpreter (BEAM) process manages a separate Python (OS) process.

{:ok, interpreter} = SnexTest.NumpyInterpreter.start_link()
{:ok, "hello world!"} = Snex.pyeval(interpreter, "return 'hello world!'")

Snex.pyeval

The main way of interacting with the interpreter process is Snex.pyeval/4 (and other arities). This is the function that runs Python code, returns data from the interpreter, and more.

{:ok, interpreter} = SnexTest.NumpyInterpreter.start_link()

{:ok, 6.0} =
  Snex.pyeval(interpreter, """
    import numpy as np
    matrix = np.fromfunction(lambda i, j: (-1) ** (i + j), (6, 6), dtype=int)
    return np.linalg.norm(matrix)
    """)

Environments

The Snex.Env struct, also called "environment", is an Elixir-side reference to a Python-side variable context in which your Python code will run. New environments can be allocated with Snex.make_env/3 (and other arities).

Environments are mutable, and will be modified by your Python code. In Python parlance, they are the global & local symbol table your Python code is executed with.

Reusing a single environment, you can use variables defined in the previous Snex.pyeval/4 calls:

{:ok, inp} = Snex.Interpreter.start_link()
{:ok, env} = Snex.make_env(inp)

# `pyeval` returns the return value of the code block. If there's
# no `return` statement, the success return value will always be `nil`
{:ok, nil} = Snex.pyeval(env, "x = 10")

# additional variables can be provided for `pyeval` to put in the environment
# before running the code
{:ok, nil} = Snex.pyeval(env, "y = x * z", %{"z" => 2})

{:ok, {10, 20, 2}} = Snex.pyeval(env, "return (x, y, z)")

Using Snex.make_env/2 and Snex.make_env/3, you can also create a new environment:

• copying variables from an old environment

  Snex.make_env(interpreter, from: old_env)
  
  # You can also omit the `interpreter` when using `:from`
  Snex.make_env(from: old_env)

• copying variables from multiple environments (later override previous)

  Snex.make_env(interpreter, from: [
    oldest_env,
    {older_env, only: ["pool"]},
    {old_env, except: ["pool"]}
  ])

• setting some initial variables (taking precedence over variables from :from)

  Snex.make_env(interpreter, %{"hello" => 42.0}, from: {old_env, only: ["world"]})

Initialization script

Snex.Interpreter can be given an :init_script option. The init script runs on interpreter startup, and prepares a "base" environment state that will be cloned to every new environment made with Snex.make_env/3.

{:ok, inp} = SnexTest.NumpyInterpreter.start_link(
  init_script: """
  import numpy as np
  my_var = 42
  """)
{:ok, env} = Snex.make_env(inp)

# The brand new `env` contains `np` and `my_var`
{:ok, 42} = Snex.pyeval(env, "return int(np.array([my_var])[0])")

If your init script takes significant time, you can pass sync_start: false to start_link/1. This will return early from the interpreter startup, and run the Python interpreter - and the init script - asynchronously. The downside is that an issue with Python or the initialization code will cause the process to crash asynchronously instead of returning an error directly from start_link/1.

Passing Snex.Env between Erlang nodes

Snex.Env garbage collection can only track usage within the node it was created on. If you send a %Snex.Env{} value to another node b@localhost, and drop any references to the value on the original node a@localhost, the garbage collector may clean up the environment even though it's used on b.

In general, it's best to use Snex.Env instances created on the same node as their interpreter - this simplifies reasoning about cleanup, especially with node disconnections and shutdowns. Agents are a great way to share Snex.Env between nodes without giving up garbage collection:

# a@localhost
{:ok, interpreter} = Snex.Interpreter.start_link()

{:ok, env_agent} = Agent.start_link(fn ->
  {:ok, env} = Snex.make_env(interpreter, %{"hello" => "hello from a@localhost!"})
  env
end)

:erpc.call(:"b@localhost", fn ->
  remote_env = Agent.get(env_agent, & &1)
  {:ok, "hello from a@localhost!"} = Snex.pyeval(remote_env, "return hello")
end)

Alternatively, you can opt into manual management of Snex.Env lifetime by calling Snex.Env.disable_gc/1 on the original node, and later destroying the env by calling Snex.destroy_env/1 on any node.

Python interface documentation

See Python Interface Documentation on HexDocs

Serialization

Elixir data is serialized using a limited subset of Python's Pickle format (version 5), and deserialized on the Python side using pickle.loads(). Python data is serialized with a subset of Erlang's External Term Format, and deserialized on the Elixir side using :erlang.binary_to_term/1.

Encoding/decoding table

The serialization happens as outlined in the table (alias Snex.Serde, as: S)

Customizing serialization

You can control struct encoding by implementing the Snex.Serde.Encoder protocol. Snex.Serde.Encoder.encode/1 will be called for any struct not explicitly handled in the table above, iff it implements the Snex.Serde.Encoder protocol. The result of the encode/1 function will then be encoded again according to the table, with the same Snex.Serde.Encoder treatment if that result contains a struct. If encode/1 returns the same struct type (e.g. Snex.Serde.Encoder.encode(%X{}) -> %X{}), the result will be encoded like a generic struct (i.e. as a dict with a __struct__ key).

Additionally, encoding defaults can be set for Snex.Interpreter and its derivatives through the encoding_opts (Snex.Serde.encoding_opts/0) option to Snex.Interpreter.start_link/1. The same option can be given to Snex.make_env/3 and Snex.pyeval/4 to selectively influence encoding of passed additional_vars.

On the Python side, you can call snex.set_custom_encoder(encoder_fun) to add encoders for your objects. encoder_fun will only be called for objects that Snex doesn't know how to serialize. The result will then be encoded further according to the table above.

Example: roundtrip serialization between an Elixir struct and a Python object

defimpl Snex.Serde.Encoder, for: Date do
  def encode(%Date{} = d),
    do: Snex.Serde.object("datetime", "date", [d.year, d.month, d.day])
end

{:ok, inp} = Snex.Interpreter.start_link(init_script: """
  import datetime

  def custom_encoder(obj):
    if isinstance(obj, datetime.date):
      return {
        snex.Atom("__struct__"): snex.Atom("Elixir.Date"),
        snex.Atom("day"): obj.day,
        snex.Atom("month"): obj.month,
        snex.Atom("year"): obj.year,
        snex.Atom("calendar"): snex.Atom("Elixir.Calendar.ISO"),
      }

    raise TypeError(f"Cannot serialize object of {type(obj)}")

  snex.set_custom_encoder(custom_encoder)
  """)

{:ok, {{"datetime", "date"}, %Date{year: 2026, month: 12, day: 27}}} =
  Snex.pyeval(inp, """
    typ = type(date)
    next_date = date + datetime.timedelta(days=364)

    return ((typ.__module__, typ.__name__), next_date)
    """,
    %{"date" => ~D[2025-12-28]})

Releases

Snex puts its managed files under _build/$MIX_ENV/snex. This works out of the box with iex -S mix and other local ways of running your code, but requires an additional step to copy files around to prepare your releases.

Fortunately, accommodating releases is easy: just add &Snex.Release.after_assemble/1 to :steps of your Mix release config. The only requirement is that it's placed after :assemble (and before :tar, if you use it.)

# mix.exs
def project do
  [
    releases: [
      demo: [
        steps: [:assemble, &Snex.Release.after_assemble/1]
      ]
    ]
  ]
end

Cookbook

Run async code

Code run by Snex lives in an asyncio loop. You can include async functions in your snippets and await them on the top level:

{:ok, inp} = Snex.Interpreter.start_link()

{:ok, "hello"} =
  Snex.pyeval(inp, """
    import asyncio
    async def do_thing():
        await asyncio.sleep(0.01)
        return "hello"

    return await do_thing()
    """)

Run blocking code

A good way to run any blocking code is to prepare and use your own thread or process pools:

{:ok, inp} = Snex.Interpreter.start_link()
{:ok, pool_env} = Snex.make_env(inp)

{:ok, nil} =
  Snex.pyeval(pool_env, """
    import asyncio
    from concurrent.futures import ThreadPoolExecutor

    pool = ThreadPoolExecutor(max_workers=cnt)
    loop = asyncio.get_running_loop()
    """, %{"cnt" => 5})

# You can keep the pool environment around and copy it into new ones
{:ok, env} = Snex.make_env(inp, from: {pool_env, only: ["pool", "loop"]})

{:ok, "world!"} =
  Snex.pyeval(env, """
    def blocking_io():
        return "world!"

    return await loop.run_in_executor(pool, blocking_io)
    """)

Use your in-repo project

You can reference your existing project path in use Snex.Interpreter.

The existing pyproject.toml and uv.lock will be used to seed the Python environment.

defmodule SnexTest.MyProject do
  use Snex.Interpreter,
    project_path: "test/my_python_proj"

  # Overrides `start_link/1` from `use Snex.Interpreter`
  def start_link(opts) do
    # Provide the project's path at runtime - you'll likely want to use
    # :code.priv_dir(:your_otp_app) and construct a path relative to that.
    my_project_path = Path.absname("test/my_python_proj")

    opts
    |> Keyword.put(:environment, %{"PYTHONPATH" => my_project_path})
    |> super()
  end
end

# $ cat test/my_python_proj/foo.py
# def bar():
#     return "hi from bar"

{:ok, inp} = SnexTest.MyProject.start_link()
{:ok, env} = Snex.make_env(inp)

{:ok, "hi from bar"} =
  Snex.pyeval(env, """
    import foo
    return foo.bar()
    """)

Send messages from Python code

Snex allows sending asynchronous BEAM messages from within your running Python code.

Every env imports the snex module, that contains a send() method which can be passed a BEAM pid. The message contents are encoded/decoded as described in Serialization.

This works especially well with async processing, where you can send updates while the event loop processes your long-running tasks.

{:ok, inp} = Snex.Interpreter.start_link()
{:ok, env} = Snex.make_env(inp)

Snex.pyeval(env, """
  await snex.send(self, b'hello from snex!')
  # insert long computation here
  """,
  %{"self" => self()}
)

"hello from snex!" = receive do val -> val end

# You can use any term supported by `Kernel.send/2` as destination
Process.register(self(), :myname)
Snex.pyeval(env, """
  await snex.send((snex.Atom('myname'), node), b'hello from snex (again!)')
  """,
  %{"node" => node()}
)

"hello from snex (again!)" = receive do val -> val end

In your external Python code (see Use your in-repo project), you can import snex (ensure snex/py_src is in your Python path) so your code and IDE are aware of Snex types, such as snex.Atom, and available functions, such as send().

Cast and call Elixir code from Python

Similar to snex.send above, you can also use snex.cast(m, f, a) and snex.call(m, f, a) to call Elixir functions from Python. In fact, snex.send is just a convenient interface on top of snex.cast!

snex.cast and snex.call differ only in how they handle results. snex.call must be awaited on, and will return the result of whatever was called, while snex.cast is fire-and-forget. Both functions will run apply(m, f, a) in a new process (m and f will be converted to atoms if given as str). They also accept an optional node argument to apply the function on a remote node.

Additionally, snex.call accepts result_encoding_opts to control how the result is encoded to Python. See Python Interface Documentation / snex.EncodingOpts for more details.

{:ok, inp} = Snex.Interpreter.start_link()
{:ok, agent} = Agent.start_link(fn -> 42 end)

{:ok, 42} =
  Snex.pyeval(
    inp,
    "return await snex.call('Elixir.Agent', 'get', [agent, identity])",
    %{"agent" => agent, "identity" => & &1}
  )

snex.Elixir

Snex includes syntax sugar for snex.call and snex.cast in the form of the Elixir object. Elixir allows calling Elixir functions using an Elixir-like syntax:

{:ok, inp} = Snex.Interpreter.start_link()

{:ok, %{"a" => 3, "b" => 2, "d" => 1}} =
  Snex.pyeval(
    inp,
    "return await Elixir.Enum.frequencies(['a', 'b', 'a', 'a', 'd', 'b'])"
  )

See Python Interface Documentation / snex.BeamModuleProxy for more details.

snex.Elixir is auto-imported into every Snex environment as Elixir

Accurate code locations in Python traces

When you use Snex.pyeval/4 with code given as String.t(), as in all examples above, Snex doesn't know where the code literal actually lives. When you inspect a stacktrace, or receive an exception result, your code will have a virtual <Snex.Code> location:

{:ok, inp} = Snex.Interpreter.start_link()

{:error, %Snex.Error{} = reason} = Snex.pyeval(inp, "raise RuntimeError('nolocation')")

~s'  File "<Snex.Code>", line 1, in <module>\n' = Enum.at(reason.traceback, -2)

To help with that, Snex defines sigils ~p and ~P (also aliased as ~PY) that annotate your code with location information, building a Snex.Code struct out of your string literal.

import Snex.Sigils

{:ok, inp} = Snex.Interpreter.start_link()

{:error, %Snex.Error{} = reason} = Snex.pyeval(inp, ~p"raise RuntimeError('nolocation')")

assert ~s'  File "#{__ENV__.file}", line 565, in <module>\n' == Enum.at(reason.traceback, -2)

All functions accepting string code also accept Snex.Code; that includes Snex.pyeval and Snex.Interpreter.start_link/1's :init_script opt.

As with other sigils, lowercase ~p interpolates and parses escapes in the literal, while uppercase ~P (and ~PY) passes the literal as-is:

iex> IO.puts(~p"12\t#{34}")
12  34
iex> IO.puts(~P"12\t#{34}")
12\t#{34}
iex> IO.puts(~PY"12\t#{34}")
12\t#{34}

Using Elixir Logger from Python

The snex Python module provides a snex.LoggingHandler class for logging to Elixir's Logger from Python. snex.LoggingHandler subclasses logging.Handler, and can be used with traditional Python's logging subsystem.

Among other things, snex.LoggingHandler ensures that the Elixir-side log has proper location information and timestamp. It also adds several extra metadata parameters, all prefixed with python_.

Besides the standard level argument, snex.LoggingHandler also accepts default_metadata and extra_metadata_keys keyword arguments. default_metadata is merged into the metadata passed to Elixir's Logger. extra_metadata_keys specifies which keys from the Python logging functions' extra dictionary should be included.

import logging

logger = logging.getLogger(__name__)
logger.addHandler(
  LoggingHandler(
    default_metadata={"application": "my_app"},
    extra_metadata_keys={"foo"}
  ),
)

logger.info("hello from Python", extra={"foo": "bar"})

# depending on your Elixir Logger's metadata settings, you may see a line similar to
# [info] hello from Python application=my_app foo=bar python_module=mymodule

Using snex.send & snex.call from external Python processes

See snex.serve documentation for a working example.