@spec args(list()) :: keyword()

Convenience helper for passing extra positional args.

@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

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

Get an attribute from a ref, raising on error.

@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

@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 (e.g., :timeout, :pool_name, :affinity)

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

@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 a helper function.

@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"

@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")

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

Get a module-level attribute, raising on error.

@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.

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

Call a method on a ref, raising on error.

@spec opts(keyword()) :: keyword()

Builds SnakeBridge options with explicit sections for kwargs, runtime, args, and idempotency.

@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

@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
• Cleanup logs are opt-in via config :snakebridge, session_cleanup_log_level: :debug

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

Releases a Python object reference, freeing memory in the Python process.

Call this to explicitly release a ref when you're done with it, rather than waiting for session cleanup or process termination.

Parameters

• ref - A SnakeBridge.Ref to release
• opts - Runtime options (optional)

Examples

{:ok, ref} = SnakeBridge.call("pathlib", "Path", ["/tmp"])
# ... use ref ...
:ok = SnakeBridge.release_ref(ref)

Notes

• After release, the ref is invalid and should not be used
• Releasing an already-released ref is a no-op
• For bulk cleanup, use release_session/1 instead

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

Releases all Python object references associated with a session.

Use this for bulk cleanup of all refs in a session, rather than releasing them individually.

Parameters

• session_id - The session ID to release
• opts - Runtime options (optional)

Examples

session_id = SnakeBridge.current_session()
# ... create many refs ...
:ok = SnakeBridge.release_session(session_id)

Notes

• After release, all refs from that session are invalid
• The session can still be reused for new calls
• For auto-sessions, prefer release_auto_session/0

@spec rt(keyword()) :: keyword()

Convenience helper for building __runtime__ options.

@spec run_as_script(
  (-> any()),
  keyword()
) :: any() | {:error, term()}

Runs a function as a script with Snakepit lifecycle management.

Defaults:

• exit_mode: :auto (only when no exit options/env vars are set)
• stop_mode: :if_started

exit_mode can also be controlled via SNAKEPIT_SCRIPT_EXIT when no exit options are provided.

Runs a script with sensible defaults for exit/stop behavior.

This is a thin wrapper around run_as_script/2.

Runs a script with explicit options.

@spec set_attr(SnakeBridge.Ref.t(), atom() | String.t(), term(), keyword()) ::
  {:ok, term()} | {: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")

@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

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

Check if a value is an unserializable marker.

When Python returns data containing objects that cannot be serialized to JSON, Snakepit replaces them with marker maps. This function detects those markers.

Examples

# Regular values
SnakeBridge.unserializable?(%{"key" => "value"})
# => false

# Marker from unserializable Python object
SnakeBridge.unserializable?(%{
  "__ffi_unserializable__" => true,
  "__type__" => "some.Module.Class",
  "__repr__" => "Class(...)"
})
# => true

Usage Pattern

case SnakeBridge.call("module", "function", []) do
  {:ok, result} ->
    if SnakeBridge.unserializable?(result) do
      {:ok, info} = SnakeBridge.unserializable_info(result)
      Logger.warning("Got unserializable: #{info.type}")
    else
      process(result)
    end
  {:error, _} = err -> err
end

See Snakepit.Serialization for details on the serialization layer.

@spec unserializable_info(term()) ::
  {:ok, %{type: String.t() | nil, repr: String.t() | nil}} | :error

Extract information from an unserializable marker.

Returns {:ok, info} with :type and :repr fields if the value is an unserializable marker, or :error otherwise.

Examples

marker = %{
  "__ffi_unserializable__" => true,
  "__type__" => "requests.models.Response",
  "__repr__" => "<Response [200]>"
}

{:ok, info} = SnakeBridge.unserializable_info(marker)
info.type
# => "requests.models.Response"
info.repr
# => "<Response [200]>"

SnakeBridge.unserializable_info(%{"normal" => "map"})
# => :error

Security Note

The repr field may contain sensitive information from the Python object's string representation. Avoid logging or persisting without review.

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

Returns the SnakeBridge version.

Context manager macro for Python with statements.

Executes a block with process-scoped runtime defaults.