Erlang Binary Term Format for Python with encoding/decoding of Erlang/Elixir data structures
npx @tessl/cli install tessl/pypi-erlang_py@2.0.0A Python library that provides complete encoding and decoding functionality for the Erlang Binary Term Format (ETF). This library enables Python applications to communicate with Erlang/Elixir systems via binary protocols, message passing, and data serialization, with enhanced Elixir compatibility features including symmetrical encoding/decoding and proper None/nil conversion.
pip install erlang_pyimport erlangimport erlang
# Encode Python data to Erlang binary format
data = {'key': 'value', 'number': 42, 'list': [1, 2, 3]}
binary_data = erlang.term_to_binary(data)
# Decode Erlang binary back to Python
decoded_data = erlang.binary_to_term(binary_data)
print(decoded_data) # {'key': 'value', 'number': 42, 'list': [1, 2, 3]}
# Working with Erlang-specific types
atom = erlang.OtpErlangAtom('hello')
binary_atom = erlang.term_to_binary(atom)
decoded_atom = erlang.binary_to_term(binary_atom)
# Compressed encoding for large data
large_data = [i for i in range(10000)]
compressed_binary = erlang.term_to_binary(large_data, compressed=True)
decompressed_data = erlang.binary_to_term(compressed_binary)The Erlang Binary Term Format (ETF) is a standardized binary protocol used by Erlang and Elixir for serializing data structures. This library provides a complete Python implementation that enables seamless interoperability between Python applications and Erlang/Elixir systems.
Key Design Concepts:
This architecture makes the library suitable for distributed systems, message queues, and any scenario requiring Python-Erlang data exchange.
Core functionality for encoding Python data to Erlang Binary Term Format and decoding Erlang terms back to Python objects.
def binary_to_term(data):
"""
Decode Erlang terms within binary data into Python types.
Parameters:
- data: bytes - Binary data containing Erlang terms
Returns:
Decoded Python object (any type)
Raises:
- ParseException: If data is invalid or cannot be parsed
"""
def term_to_binary(term, compressed=False):
"""
Encode Python types into Erlang terms in binary data.
Parameters:
- term: Any - Python object to encode
- compressed: bool or int - Compression level (False=no compression, True=level 6, 0-9=specific level)
Returns:
Binary data (bytes) containing encoded Erlang terms
Raises:
- InputException: If compression level is invalid (must be in [0..9])
- OutputException: If encoding fails (e.g., uint32 overflow)
"""Specialized classes representing Erlang-specific data types that don't have direct Python equivalents.
class OtpErlangAtom:
"""Represents an Erlang atom."""
def __init__(self, value):
"""
Create an Erlang atom.
Parameters:
- value: int | str | bytes - Integer for atom cache reference,
string or bytes for atom name
"""
def binary(self):
"""Return encoded representation of the atom."""
class OtpErlangBinary:
"""Represents an Erlang binary with optional bit-level precision."""
def __init__(self, value, bits=8):
"""
Create an Erlang binary.
Parameters:
- value: bytes - Binary value
- bits: int - Number of bits in last byte (default: 8)
"""
def binary(self):
"""Return encoded representation of the binary."""
class OtpErlangFunction:
"""Represents an Erlang function."""
def __init__(self, tag, value):
"""
Create an Erlang function.
Parameters:
- tag: Function tag identifier
- value: Function value/data
"""
def binary(self):
"""Return encoded representation of the function."""
class OtpErlangList:
"""Represents an Erlang list with optional improper list support."""
def __init__(self, value, improper=False):
"""
Create an Erlang list.
Parameters:
- value: list - List contents
- improper: bool - Whether list has no empty list tail (default: False)
"""
def binary(self):
"""Return encoded representation of the list."""
class OtpErlangPid:
"""Represents an Erlang process identifier (PID)."""
def __init__(self, node, id_value, serial, creation):
"""
Create an Erlang PID.
Parameters:
- node: OtpErlangAtom - Node atom where the process resides
- id_value: bytes - Process ID value
- serial: bytes - Serial number
- creation: bytes - Creation identifier
"""
def binary(self):
"""Return encoded representation of the PID."""
class OtpErlangPort:
"""Represents an Erlang port."""
def __init__(self, node, id_value, creation):
"""
Create an Erlang port.
Parameters:
- node: OtpErlangAtom - Node atom where the port resides
- id_value: bytes - Port ID value
- creation: bytes - Creation identifier
"""
def binary(self):
"""Return encoded representation of the port."""
class OtpErlangReference:
"""Represents an Erlang reference."""
def __init__(self, node, id_value, creation):
"""
Create an Erlang reference.
Parameters:
- node: OtpErlangAtom - Node atom where the reference was created
- id_value: bytes - Reference ID value
- creation: bytes - Creation identifier
"""
def binary(self):
"""Return encoded representation of the reference."""Exception classes for different error conditions during encoding/decoding operations.
class InputException(ValueError):
"""
InputError describes problems with function input parameters.
Extends ValueError.
"""
def __init__(self, s):
"""
Parameters:
- s: Error message string
"""
class OutputException(TypeError):
"""
OutputError describes problems with creating function output data.
Extends TypeError.
"""
def __init__(self, s):
"""
Parameters:
- s: Error message string
"""
class ParseException(SyntaxError):
"""
ParseError provides specific parsing failure information.
Extends SyntaxError.
"""
def __init__(self, s):
"""
Parameters:
- s: Error message string
"""Advanced functionality that extends beyond the core binary term format conversion.
def consult(string_in):
"""
Provide file:consult/1 functionality with Python types.
Parse textual Erlang data representation into Python objects,
avoiding external dependencies for simple data parsing scenarios.
Parameters:
- string_in: str - String containing textual Erlang data
Returns:
Python object representation of the parsed Erlang data
Note: This function is not in __all__ and should be considered
internal/advanced API. Use with caution in production code.
Raises:
- ParseException: If the string cannot be parsed as valid Erlang data
"""# All OTP classes support these common methods:
def __repr__(self) -> str:
"""Return string representation of the object."""
def __hash__(self) -> int:
"""Return hash value for use in sets and dictionaries."""
def __eq__(self, other) -> bool:
"""Test equality with another object."""
# All exception classes support:
def __str__(self) -> str:
"""Return string representation of the error message."""
# Internal types available for advanced usage:
class frozendict(dict):
"""
Immutable dictionary that cannot be modified after creation.
Used internally for representing Erlang maps.
Available for import but not part of official public API.
"""
def __init__(self, *args, **kw):
"""
Create immutable dictionary, recursively converting nested dicts.
Parameters:
- *args: Positional arguments passed to dict constructor
- **kw: Keyword arguments passed to dict constructor
"""
def __hash__(self) -> int:
"""Return hash value based on dictionary contents."""
def __setitem__(self, key, value):
"""Raise TypeError - frozendict is immutable."""
def __delitem__(self, key):
"""Raise TypeError - frozendict is immutable."""
def clear(self):
"""Raise TypeError - frozendict is immutable."""
def pop(self, key, *args):
"""Raise TypeError - frozendict is immutable."""
def popitem(self):
"""Raise TypeError - frozendict is immutable."""
def setdefault(self, key, default=None):
"""Raise TypeError - frozendict is immutable."""
def update(self, *args, **kw):
"""Raise TypeError - frozendict is immutable."""import erlang
# Create atoms from strings
atom1 = erlang.OtpErlangAtom("hello")
atom2 = erlang.OtpErlangAtom("world")
# Encode and decode
encoded = erlang.term_to_binary([atom1, atom2])
decoded = erlang.binary_to_term(encoded)
print(decoded) # [OtpErlangAtom('hello'), OtpErlangAtom('world')]import erlang
# Mix of Python types and Erlang-specific types
complex_data = {
'atoms': [erlang.OtpErlangAtom('ok'), erlang.OtpErlangAtom('error')],
'binary': erlang.OtpErlangBinary(b'hello world'),
'regular_list': [1, 2, 3, 'string'],
'erlang_list': erlang.OtpErlangList([1, 2, 3]),
'nested': {'inner': {'value': 42}}
}
# Round-trip encoding/decoding
encoded = erlang.term_to_binary(complex_data)
decoded = erlang.binary_to_term(encoded)
# Verify round-trip integrity
assert decoded == complex_dataimport erlang
try:
# Invalid binary data
erlang.binary_to_term(b'invalid')
except erlang.ParseException as e:
print(f"Parse error: {e}")
try:
# Invalid compression level
erlang.term_to_binary("test", compressed=15)
except erlang.InputException as e:
print(f"Input error: {e}")import erlang
large_data = list(range(10000))
# No compression
uncompressed = erlang.term_to_binary(large_data)
# Default compression (level 6)
compressed = erlang.term_to_binary(large_data, compressed=True)
# Specific compression level
highly_compressed = erlang.term_to_binary(large_data, compressed=9)
print(f"Uncompressed: {len(uncompressed)} bytes")
print(f"Compressed: {len(compressed)} bytes")
print(f"Highly compressed: {len(highly_compressed)} bytes")
# All decode to the same data
assert erlang.binary_to_term(uncompressed) == large_data
assert erlang.binary_to_term(compressed) == large_data
assert erlang.binary_to_term(highly_compressed) == large_dataimport erlang
# Parse textual Erlang data (advanced usage)
erlang_text = "{ok, [1, 2, 3]}."
try:
parsed_data = erlang.consult(erlang_text)
print(parsed_data) # {'ok': [1, 2, 3]}
except erlang.ParseException as e:
print(f"Parse error: {e}")
# Note: consult is not in __all__ and should be used carefully
# For most use cases, prefer binary_to_term/term_to_binary