API Reference

Complete reference for all public functions, classes, and constants in the crous Python module.

Core Functions

crous.dumps(obj, default=None)

Serialize a Python object to FLUX binary bytes.

ParameterTypeDescription
objAnyThe Python object to serialize
defaultCallable[[Any], Any] | NoneFallback handler for unsupported types

Returns: bytes — FLUX binary data

Raises: CrousEncodeError — if serialization fails

crous.loads(data, object_hook=None)

Deserialize FLUX/CROUS binary bytes to a Python object.

ParameterTypeDescription
databytesBinary data to deserialize
object_hookCallable[[dict], Any] | NoneTransform decoded dictionaries

Returns: Any — Deserialized Python object

Raises: CrousDecodeError — if deserialization fails

crous.dump(obj, file, default=None)

Serialize and write to a file path or file object.

ParameterTypeDescription
objAnyThe Python object to serialize
filestr | IO[bytes]File path or writable binary file object
defaultCallable | NoneFallback handler for unsupported types

Returns: None

Raises: CrousEncodeError, IOError

crous.load(file, object_hook=None)

Read and deserialize from a file path or file object.

ParameterTypeDescription
filestr | IO[bytes]File path or readable binary file object
object_hookCallable[[dict], Any] | NoneTransform decoded dictionaries

Returns: Any — Deserialized Python object

Raises: CrousDecodeError, FileNotFoundError

Stream Functions

crous.dump_to_stream(obj, stream, default=None)

Serialize and write to a writable stream.

ParameterTypeDescription
objAnyObject to serialize
streamIO[bytes]Any object with a write() method
defaultCallable | NoneFallback handler

crous.load_from_stream(stream, object_hook=None)

Read and deserialize from a readable stream.

ParameterTypeDescription
streamIO[bytes]Any object with a read() method
object_hookCallable | NoneTransform decoded dictionaries

CROUT Functions

crous.to_crout(obj)

Serialize a Python object to CROUT text format.

ParameterTypeDescription
objAnyObject to serialize

Returns: str — CROUT text representation

crous.from_crout(text)

Deserialize CROUT text to a Python object.

ParameterTypeDescription
textstrCROUT text data

Returns: Any — Deserialized Python object

crous.crout_to_flux(text)

Convert CROUT text directly to FLUX binary (no Python intermediary).

ParameterTypeDescription
textstrCROUT text data

Returns: bytes — FLUX binary data

crous.flux_to_crout(data)

Convert FLUX binary directly to CROUT text (no Python intermediary).

ParameterTypeDescription
databytesFLUX binary data

Returns: str — CROUT text representation

Custom Serializer Registry

crous.register_serializer(type, serializer, tag=None)

Register a custom serializer for a Python type.

ParameterTypeDescription
typetypeThe Python class to register
serializerCallable[[Any], Any]Function that converts instance to serializable value
tagint | NoneOptional explicit tag number (auto-assigned from 100 if omitted)

crous.unregister_serializer(type)

Remove a previously registered serializer.

ParameterTypeDescription
typetypeThe Python class to unregister

crous.register_decoder(tag, decoder)

Register a custom decoder for a tagged value.

ParameterTypeDescription
tagintThe tag number to decode
decoderCallable[[Any], Any]Function that reconstructs the custom type

crous.unregister_decoder(tag)

Remove a previously registered decoder.

ParameterTypeDescription
tagintThe tag number to unregister

Classes

crous.CrousEncoder

Reusable encoder with persistent options.

class CrousEncoder:
    def __init__(self, default=None, allow_custom=True):
        """
        Args:
            default: Fallback handler for unsupported types
            allow_custom: Whether to use registered custom serializers
        """

    def encode(self, obj) -> bytes:
        """Serialize obj to FLUX binary bytes."""

crous.CrousDecoder

Reusable decoder with persistent options.

class CrousDecoder:
    def __init__(self, object_hook=None):
        """
        Args:
            object_hook: Transform function for decoded dictionaries
        """

    def decode(self, data: bytes) -> Any:
        """Deserialize FLUX binary bytes to Python object."""

Exceptions

crous.CrousError

Base exception for all Crous errors. Inherits from Exception.

crous.CrousEncodeError

Raised during serialization. Inherits from CrousError.

crous.CrousDecodeError

Raised during deserialization. Inherits from CrousError.

Module Attributes

AttributeTypeValueDescription
crous.__version__str"2.0.0"Library version string
crous.__version_tuple__tuple(2, 0, 0)Version as tuple

crous.version Module

The crous.version submodule provides detailed version and compatibility utilities.

Constants

ConstantValueDescription
VERSION_MAJOR2Major version
VERSION_MINOR0Minor version
VERSION_PATCH0Patch version
VERSION_STRING"2.0.0"Version string
VERSION_HEX0x020000Version as hex
WIRE_VERSION_CURRENT2Current wire format version
WIRE_VERSION_MIN_READ1Minimum readable wire version
WIRE_VERSION_MAX_READ2Maximum readable wire version

Feature Flags (IntFlag)

FlagBitDescription
Feature.NONE0No features
Feature.TAGGED1Tagged values
Feature.TUPLE2Native tuple support
Feature.SET4Set type
Feature.FROZENSET8Frozenset type
Feature.COMPRESSION16Compression
Feature.STREAMING32Streaming mode
Feature.SCHEMA64Schema validation
Feature.ENCRYPTION128Encryption
Feature.DATETIME256Datetime support
Feature.DECIMAL512Decimal support
Feature.UUID1024UUID support
Feature.PATH2048Path support
Feature.COMMENTS4096Comments
Feature.METADATA8192Metadata
Feature.CHECKSUM16384Checksums

SemanticVersion

from crous.version import SemanticVersion

# Create from string
v = SemanticVersion.parse("2.0.0")

# Get current library version
v = SemanticVersion.current()

# Comparison operators
v1 = SemanticVersion.parse("1.0.0")
v2 = SemanticVersion.parse("2.0.0")
assert v1 < v2

# Constraint checking (supports >=, >, <=, <, ==, !=, ^, ~)
assert v2.satisfies(">=1.0.0")
assert v2.satisfies("^2.0.0")
assert v2.satisfies("~2.0")

# Bump versions
v3 = v2.bump_minor()   # 2.1.0
v4 = v2.bump_major()   # 3.0.0

Header

from crous.version import Header

header = Header.parse(binary_data)
header.magic         # b'FLUX'
header.wire_version  # 1
header.flags         # 0
header.is_valid      # True if magic == b'FLUX'

check_compatibility(data)

from crous.version import check_compatibility

result = check_compatibility(binary_data)
result.status         # Compatibility enum value
result.is_compatible  # True/False
result.is_warning     # True if warnings
result.is_error       # True if incompatible
result.message        # Human-readable description

Serializable Types

# Type alias for all natively serializable types
CrousSerializable = Union[
    None,
    bool,
    int,
    float,
    str,
    bytes,
    List['CrousSerializable'],
    Tuple['CrousSerializable', ...],
    Dict[str, 'CrousSerializable'],
    Set['CrousSerializable'],
    FrozenSet['CrousSerializable'],
]