SnakeBridge (SnakeBridge v0.7.4)

View Source

Universal FFI bridge to Python.

SnakeBridge provides two ways to call Python:

  1. Generated wrappers (compile-time): Type-safe, documented Elixir modules generated from Python library introspection.

  2. Dynamic calls (runtime): Direct calls to any Python module without code generation, using string module paths.

Universal FFI API

The universal FFI requires no code generation:

# Call any Python function
{:ok, result} = SnakeBridge.call("math", "sqrt", [16])

# Get module attributes
{:ok, pi} = SnakeBridge.get("math", "pi")

# Work with Python objects
{:ok, path} = SnakeBridge.call("pathlib", "Path", ["/tmp"])
{:ok, exists?} = SnakeBridge.method(path, "exists", [])

Sessions

SnakeBridge automatically manages Python object sessions. Each Elixir process gets an isolated session, and refs are automatically cleaned up when the process terminates.

For explicit session control, use SnakeBridge.SessionContext.with_session/1.

Type Mapping

ElixirPython
nilNone
true/falseTrue/False
integersint
floatsfloat
stringsstr
SnakeBridge.bytes(data)bytes
listslist
mapsdict
tuplestuple
MapSetset
atomstagged atom (decoded to string by default)
DateTimedatetime
SnakeBridge.RefPython object reference

Summary

Functions

Get an attribute from a Python object reference.

Get an attribute from a ref, raising on error.

Create a Bytes wrapper for explicit binary data.

Call a Python function, raising on error.

Get the current session ID.

Get a module-level attribute from Python.

Get a module-level attribute, raising on error.

Call a method on a Python object reference.

Call a method on a ref, raising on error.

Check if a value is a Python object reference.

Release and clear the auto-session for the current process.

Set an attribute on a Python object reference.

Stream results from a Python generator or iterator.

Returns the SnakeBridge version.

Context manager macro for Python with statements.

Functions

attr(ref, attr, opts \\ [])

@spec attr(SnakeBridge.Ref.t(), atom() | String.t(), keyword()) ::
  {:ok, term()} | {:error, term()}

Get an attribute from a Python object reference.

Parameters

  • ref - A SnakeBridge.Ref from a previous call
  • attr - Attribute name as atom or string
  • opts - Runtime options

Examples

{:ok, path} = SnakeBridge.call("pathlib", "Path", ["/tmp/file.txt"])
{:ok, name} = SnakeBridge.attr(path, "name")
# => {:ok, "file.txt"}

{:ok, parent} = SnakeBridge.attr(path, "parent")
# => {:ok, %SnakeBridge.Ref{...}}  # parent is also a Path

attr!(ref, attr, opts \\ [])

@spec attr!(SnakeBridge.Ref.t(), atom() | String.t(), keyword()) :: term()

Get an attribute from a ref, raising on error.

bytes(data)

@spec bytes(binary()) :: SnakeBridge.Bytes.t()

Create a Bytes wrapper for explicit binary data.

By default, SnakeBridge encodes UTF-8 valid strings as Python str. Use this function to explicitly send data as Python bytes.

Examples

# Crypto
{:ok, hash_ref} = SnakeBridge.call("hashlib", "md5", [SnakeBridge.bytes("abc")])
{:ok, hex} = SnakeBridge.method(hash_ref, "hexdigest", [])

# Binary protocols
{:ok, packed} = SnakeBridge.call("struct", "pack", [">I", 12345])

# Base64
{:ok, encoded} = SnakeBridge.call("base64", "b64encode", [SnakeBridge.bytes("hello")])

When to Use

Python distinguishes str (text) from bytes (binary). Use bytes/1 for:

  • Cryptographic operations (hashlib, hmac, cryptography)
  • Binary packing (struct)
  • Base64 encoding
  • Network protocols
  • File I/O in binary mode

call(module, function, args \\ [], opts \\ [])

@spec call(module() | String.t(), atom() | String.t(), list(), keyword()) ::
  {:ok, term()} | {:error, term()}

Call a Python function.

Accepts either a generated SnakeBridge module or a Python module path string.

Parameters

  • module - A generated module atom (e.g., Numpy) or a module path string (e.g., "numpy")
  • function - Function name as atom or string
  • args - List of positional arguments (default: [])
  • opts - Keyword arguments passed to Python, plus:
    • :idempotent - Mark call as cacheable (default: false)
    • :__runtime__ - Pass-through options to Snakepit

Examples

# Call stdlib function
{:ok, 4.0} = SnakeBridge.call("math", "sqrt", [16])

# With keyword arguments
{:ok, 3.14} = SnakeBridge.call("builtins", "round", [3.14159], ndigits: 2)

# Submodule
{:ok, path} = SnakeBridge.call("os.path", "join", ["/tmp", "file.txt"])

# Create objects
{:ok, ref} = SnakeBridge.call("pathlib", "Path", ["."])

Return Values

  • {:ok, value} - Decoded Elixir value for JSON-serializable results
  • {:ok, %SnakeBridge.Ref{}} - Reference for non-serializable Python objects
  • {:error, reason} - Error from Python

Notes

  • String module paths trigger dynamic dispatch (no codegen required)
  • Sessions are automatic; refs are isolated per Elixir process
  • Non-JSON-serializable returns are wrapped in refs for safe access

call!(module, function, args \\ [], opts \\ [])

@spec call!(module() | String.t(), atom() | String.t(), list(), keyword()) :: term()

Call a Python function, raising on error.

Same as call/4 but raises on error instead of returning {:error, reason}.

Examples

result = SnakeBridge.call!("math", "sqrt", [16])
# => 4.0

# Raises on error
SnakeBridge.call!("nonexistent_module", "fn", [])
# ** (Snakepit.Error) ...

call_helper(helper, args \\ [], opts \\ [])

Call a helper function.

current_session()

@spec current_session() :: String.t()

Get the current session ID.

Returns the session ID for the current Elixir process. Sessions are automatically created on first Python call.

Examples

session_id = SnakeBridge.current_session()
# => "auto_<0.123.0>_1703944800000"

# With explicit session
SnakeBridge.SessionContext.with_session(session_id: "my_session", fn ->
  SnakeBridge.current_session()
end)
# => "my_session"

get(module, attr, opts \\ [])

@spec get(module() | String.t(), atom() | String.t(), keyword()) ::
  {:ok, term()} | {:error, term()}

Get a module-level attribute from Python.

Retrieves constants, classes, or any attribute from a Python module.

Parameters

  • module - A generated module atom or a module path string
  • attr - Attribute name as atom or string
  • opts - Runtime options

Examples

# Module constant
{:ok, pi} = SnakeBridge.get("math", "pi")
# => {:ok, 3.141592653589793}

# Module-level class (returns ref)
{:ok, path_class} = SnakeBridge.get("pathlib", "Path")

# Nested attribute
{:ok, sep} = SnakeBridge.get("os", "sep")

get!(module, attr, opts \\ [])

@spec get!(module() | String.t(), atom() | String.t(), keyword()) :: term()

Get a module-level attribute, raising on error.

method(ref, method, args \\ [], opts \\ [])

@spec method(SnakeBridge.Ref.t(), atom() | String.t(), list(), keyword()) ::
  {:ok, term()} | {:error, term()}

Call a method on a Python object reference.

Parameters

  • ref - A SnakeBridge.Ref from a previous call
  • method - Method name as atom or string
  • args - Positional arguments (default: [])
  • opts - Keyword arguments

Examples

{:ok, path} = SnakeBridge.call("pathlib", "Path", ["."])
{:ok, exists?} = SnakeBridge.method(path, "exists", [])
{:ok, resolved} = SnakeBridge.method(path, "resolve", [])

# With arguments
{:ok, child} = SnakeBridge.method(path, "joinpath", ["subdir", "file.txt"])

Notes

This is equivalent to SnakeBridge.Dynamic.call/4 but with a clearer name for the universal FFI context.

method!(ref, method, args \\ [], opts \\ [])

@spec method!(SnakeBridge.Ref.t(), atom() | String.t(), list(), keyword()) :: term()

Call a method on a ref, raising on error.

ref?(value)

@spec ref?(term()) :: boolean()

Check if a value is a Python object reference.

Examples

{:ok, path} = SnakeBridge.call("pathlib", "Path", ["."])
SnakeBridge.ref?(path)
# => true

SnakeBridge.ref?("string")
# => false

release_auto_session()

@spec release_auto_session() :: :ok

Release and clear the auto-session for the current process.

Call this to eagerly release Python object refs when you're done with Python calls, rather than waiting for process termination.

Examples

{:ok, ref} = SnakeBridge.call("numpy", "array", [[1,2,3]])
# ... use ref ...
SnakeBridge.release_auto_session()  # Clean up now

Notes

  • This releases all refs in the current process's auto-session
  • A new session is created automatically on the next Python call
  • Use SessionContext.with_session/1 for more fine-grained control

set_attr(ref, attr, value, opts \\ [])

@spec set_attr(SnakeBridge.Ref.t(), atom() | String.t(), term(), keyword()) ::
  :ok | {:error, term()}

Set an attribute on a Python object reference.

Parameters

  • ref - A SnakeBridge.Ref from a previous call
  • attr - Attribute name as atom or string
  • value - New value for the attribute
  • opts - Runtime options

Examples

{:ok, obj} = SnakeBridge.call("some_module", "SomeClass", [])
:ok = SnakeBridge.set_attr(obj, "property", "new_value")

stream(module, function, args, opts, callback)

@spec stream(module() | String.t(), atom() | String.t(), list(), keyword(), (term() ->
                                                                         term())) ::
  :ok | {:ok, :done} | {:error, term()}

Stream results from a Python generator or iterator.

Calls a Python function that returns an iterable and invokes the callback for each element.

Parameters

  • module - Module atom or path string
  • function - Function name
  • args - Positional arguments
  • opts - Keyword arguments for the Python function
  • callback - Function called with each streamed element

Examples

# Process file in chunks
SnakeBridge.stream("pandas", "read_csv", ["large.csv"], [chunksize: 1000], fn chunk ->
  IO.puts("Processing chunk")
end)

# Iterate range
SnakeBridge.stream("builtins", "range", [10], [], fn i ->
  IO.puts("Got: #{i}")
end)

Return Value

  • {:ok, :done} - Iteration completed successfully (for string module paths)
  • :ok - Iteration completed successfully (for atom modules)
  • {:error, reason} - Error during iteration

version()

@spec version() :: String.t()

Returns the SnakeBridge version.

with_python(ref, list)

(macro)

Context manager macro for Python with statements.