Snapshots

Classes for iterative execution and resuming paused code

Monty provides snapshot classes that allow you to pause and resume code execution. This is useful for handling external function calls iteratively, implementing async operations, or serializing execution state.

FunctionSnapshot

Represents a paused execution waiting for an external function call return value. Contains information about the pending external function call and allows resuming execution with the return value.

Properties

script_name · str

The name of the script being executed

is_os_function · bool

Whether this snapshot is for an OS function call (e.g., Path.stat)

is_method_call · bool

Whether this snapshot is for a dataclass method call (first arg is self)

function_name · str | OsFunction

The name of the function being called (external function or OS function like 'Path.stat'). Will be an OsFunction if is_os_function is True.

args · tuple[Any, ...]

The positional arguments passed to the external function

kwargs · dict[str, Any]

The keyword arguments passed to the external function

call_id · int

The unique identifier for this external function call

Methods

resume()

def resume(
    self,
    *,
    return_value: Any = ...,
    exception: BaseException = ...,
    future: EllipsisType = ...,
) -> FunctionSnapshot | NameLookupSnapshot | FutureSnapshot | MontyComplete

Resume execution with a return value, exception, or future. Only one of the parameters should be provided. resume() may only be called once on each FunctionSnapshot instance. The GIL is released allowing parallel execution.

return_value · Any

The value to return from the external function call

exception · BaseException

An exception to raise in the Monty interpreter

future · EllipsisType

A future to await in the Monty interpreter. Pass ... (ellipsis) to indicate a pending future.

• FunctionSnapshot if another external function call is pending
• NameLookupSnapshot if another name lookup is pending
• FutureSnapshot if futures need to be resolved
• MontyComplete if execution finished

Raises:

• TypeError: If both arguments are provided or if invalid arguments are passed
• RuntimeError: If execution has already completed or if resume was already called
• MontyRuntimeError: If the code raises an exception during execution

dump()

def dump(self) -> bytes

Serialize the FunctionSnapshot instance to a binary format. This allows suspending execution and resuming later, potentially in a different process.

Bytes containing the serialized FunctionSnapshot instance

Note: The print_callback is not serialized and must be re-provided via set_print_callback() after loading if print output is needed.

Raises:

• ValueError: If serialization fails
• RuntimeError: If the progress has already been resumed

load() (static method)

@staticmethod
def load(
    data: bytes,
    *,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    dataclass_registry: list[type] | None = None,
) -> FunctionSnapshot

Deserialize a FunctionSnapshot instance from binary format.

data · bytes · required

The serialized FunctionSnapshot data from dump()

print_callback · Callable[[Literal['stdout'], str], None] | None

Optional callback for print output

dataclass_registry · list[type] | None

Optional list of dataclass types to register for proper isinstance() support on output

A new FunctionSnapshot instance

Raises:

• ValueError: If deserialization fails

Example

import pydantic_monty

code = """
data = fetch(url)
len(data)
"""

m = pydantic_monty.Monty(code, inputs=['url'])

# Start execution - pauses when fetch() is called
progress = m.start(inputs={'url': 'https://example.com'})

if isinstance(progress, pydantic_monty.FunctionSnapshot):
    print(f"Function: {progress.function_name}")
    print(f"Args: {progress.args}")
    print(f"Kwargs: {progress.kwargs}")
    print(f"Call ID: {progress.call_id}")
    
    # Resume with a return value
    progress = progress.resume(return_value='hello world')

if isinstance(progress, pydantic_monty.MontyComplete):
    print(f"Output: {progress.output}")

MontyComplete

The result of a completed code execution.

Properties

output · Any

The final output value from the executed code

Example

import pydantic_monty

m = pydantic_monty.Monty('2 + 2')
result = m.start()

if isinstance(result, pydantic_monty.MontyComplete):
    print(result.output)
    # Output: 4

NameLookupSnapshot

Represents a paused execution waiting for a name lookup to be resolved. This is used in advanced scenarios where variable lookups need external resolution.

Properties

script_name · str

The name of the script being executed

variable_name · str

The name of the variable being looked up

Methods

resume()

def resume(
    self,
    *,
    value: Any | None = None,
) -> FunctionSnapshot | NameLookupSnapshot | FutureSnapshot | MontyComplete

Resume execution with the value from a name lookup, if any. If no value is passed, a NameError is raised. resume() may only be called once on each NameLookupSnapshot instance. The GIL is released allowing parallel execution.

value · Any | None

The value from the name lookup, if any

• FunctionSnapshot if an external function call is pending
• NameLookupSnapshot if more futures need to be resolved
• FutureSnapshot if another name lookup is pending
• MontyComplete if execution finished

Raises:

• TypeError: If result dict has invalid keys
• RuntimeError: If execution has already completed
• MontyRuntimeError: If the code raises an exception during execution

dump() and load()

Similar to FunctionSnapshot, NameLookupSnapshot also supports serialization:

def dump(self) -> bytes

@staticmethod
def load(
    data: bytes,
    *,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    dataclass_registry: list[type] | None = None,
) -> NameLookupSnapshot

FutureSnapshot

Represents a paused execution waiting for multiple futures to be resolved. This is used when handling async operations or parallel external function calls.

Properties

script_name · str

The name of the script being executed

pending_call_ids · list[int]

The call IDs of the pending futures. Raises an error if the snapshot has already been resumed.

Methods

resume()

def resume(
    self,
    results: dict[int, ExternalResult],
) -> FunctionSnapshot | NameLookupSnapshot | FutureSnapshot | MontyComplete

Resume execution with results for one or more futures. resume() may only be called once on each FutureSnapshot instance. The GIL is released allowing parallel execution.

results · dict[int, ExternalResult] · required

Dict mapping call_id to result dict. Each result dict must have either 'return_value' or 'exception' key (not both).

• FunctionSnapshot if an external function call is pending
• NameLookupSnapshot if more futures need to be resolved
• FutureSnapshot if more futures need to be resolved
• MontyComplete if execution finished

Raises:

• TypeError: If result dict has invalid keys
• RuntimeError: If execution has already completed
• MontyRuntimeError: If the code raises an exception during execution

dump() and load()

Similar to FunctionSnapshot, FutureSnapshot also supports serialization:

def dump(self) -> bytes

@staticmethod
def load(
    data: bytes,
    *,
    print_callback: Callable[[Literal['stdout'], str], None] | None = None,
    dataclass_registry: list[type] | None = None,
) -> FutureSnapshot

ExternalResult Type

The ExternalResult type is used when resuming FutureSnapshot instances. It’s a union type:

class ExternalReturnValue(TypedDict):
    return_value: Any

class ExternalException(TypedDict):
    exception: Exception

class ExternalFuture(TypedDict):
    future: EllipsisType

ExternalResult = ExternalReturnValue | ExternalException | ExternalFuture

Example with FutureSnapshot

import pydantic_monty

code = """
result1 = fetch('url1')
result2 = fetch('url2')
result1 + result2
"""

m = pydantic_monty.Monty(code)
progress = m.start()

# First function call
if isinstance(progress, pydantic_monty.FunctionSnapshot):
    call_id_1 = progress.call_id
    progress = progress.resume(future=...)

# Second function call
if isinstance(progress, pydantic_monty.FunctionSnapshot):
    call_id_2 = progress.call_id
    progress = progress.resume(future=...)

# Now we have a FutureSnapshot waiting for both results
if isinstance(progress, pydantic_monty.FutureSnapshot):
    print(f"Pending calls: {progress.pending_call_ids}")
    
    # Provide results for both futures
    results = {
        call_id_1: {'return_value': 'data1'},
        call_id_2: {'return_value': 'data2'}
    }
    progress = progress.resume(results)

if isinstance(progress, pydantic_monty.MontyComplete):
    print(progress.output)
    # Output: data1data2

Serialization Example

import pydantic_monty

m = pydantic_monty.Monty('fetch(url)', inputs=['url'])
progress = m.start(inputs={'url': 'https://example.com'})

# Serialize the execution state
state = progress.dump()

# Later, restore and resume (e.g., in a different process)
progress2 = pydantic_monty.FunctionSnapshot.load(state)
result = progress2.resume(return_value='response data')

if isinstance(result, pydantic_monty.MontyComplete):
    print(result.output)
    # Output: response data

Related

• Monty class - Main class that creates snapshots via start()
• Errors - Error types that can be raised during execution