tessl/pypi-vine

Python promises library providing specialized promise-based asynchronous operations and lazy evaluation

—

Pending

Overview

Eval results

Files

Promise Operations

Name: tessl/pypi-vine
Author: tessl

The core promise implementation providing asynchronous operations and lazy evaluation. Promises can represent both future values and deferred computations, allowing for complex chaining and composition patterns.

Capabilities

Promise Creation

Create promise objects with optional functions, arguments, and error handling.

class promise:
    def __init__(self, fun=None, args=None, kwargs=None, callback=None, on_error=None, weak=False, ignore_result=False):
        """
        Create a new promise.

        Parameters:
        - fun: callable, function to execute when promise is fulfilled
        - args: tuple, positional arguments for the function
        - kwargs: dict, keyword arguments for the function  
        - callback: promise or callable, callback to chain immediately
        - on_error: promise or callable, error handler for exceptions
        - weak: bool, use weak references for the function (prevents memory leaks)
        - ignore_result: bool, ignore function return value (performance optimization)
        """

Usage Examples:

from vine import promise

# Empty promise to be fulfilled later
p = promise()

# Promise with function and arguments
p = promise(print, args=("Hello, World!",))

# Promise that ignores return value (optimization)
p = promise(some_function, ignore_result=True)

# Promise with error handling
p = promise(risky_function, on_error=error_handler)

Promise Fulfillment

Execute the promise with arguments, triggering callback chain.

def __call__(self, *args, **kwargs):
    """
    Fulfill the promise by executing its function with provided arguments.
    
    Parameters:
    - *args: additional positional arguments 
    - **kwargs: additional keyword arguments
    
    Returns:
    Return value of the promise function, or None if cancelled/failed
    """

Usage Examples:

# Fulfill empty promise with value
p = promise()
p.then(lambda x: print(f"Got: {x}"))
p(42)  # Prints: Got: 42

# Fulfill promise with function
p = promise(lambda x, y: x + y, args=(10,))
result = p(20)  # Returns 30

Promise Chaining

Chain promises together using the .then() method for sequential operations.

def then(self, callback, on_error=None):
    """
    Chain a callback to execute when promise is fulfilled.
    
    Parameters:
    - callback: callable or promise, function to execute with promise result
    - on_error: callable or promise, error handler for this callback
    
    Returns:
    The callback promise for further chaining
    """

Usage Examples:

# Simple chaining
p1 = promise(lambda x: x * 2)
p2 = p1.then(lambda x: x + 10)
p3 = p2.then(lambda x: print(f"Final: {x}"))

p1(5)  # Prints: Final: 20

# Chain with error handling
p = promise(risky_operation)
p.then(
    success_handler,
    on_error=lambda exc: print(f"Error: {exc}")
)

Error Handling

Handle exceptions and propagate errors through promise chains.

def throw(self, exc=None, tb=None, propagate=True):
    """
    Throw an exception through the promise chain.
    
    Parameters:
    - exc: Exception, exception to throw (uses current exception if None)
    - tb: traceback, traceback object for exception
    - propagate: bool, whether to re-raise if no error handler
    """

def throw1(self, exc=None):
    """
    Throw exception to this promise only (no propagation).
    
    Parameters:
    - exc: Exception, exception to throw (uses current exception if None)
    """

Usage Examples:

# Handle errors in promise chain
p = promise()
p.then(
    lambda x: x / 0,  # Will cause error
    on_error=lambda exc: print(f"Caught: {exc}")
)
p(10)

# Manually throw exception
p = promise()
p.then(lambda x: print("Success"))
p.throw(ValueError("Something went wrong"))

Promise Cancellation

Cancel promises and stop execution of callback chains.

def cancel(self):
    """
    Cancel the promise and all pending callbacks.
    
    Cancellation propagates through the entire callback chain,
    stopping execution and cleaning up resources.
    """

Usage Examples:

# Cancel individual promise
p = promise()
p.then(lambda x: print("This won't run"))
p.cancel()
p(42)  # Nothing happens

# Cancellation propagates through chains
p1 = promise()
p2 = p1.then(lambda x: x * 2)
p3 = p2.then(lambda x: print(x))

p1.cancel()  # Cancels entire chain

Promise Properties

Access promise state and values.

@property
def ready(self) -> bool:
    """True when promise has been fulfilled (successfully or with error)."""

@property  
def failed(self) -> bool:
    """True when promise failed with an exception."""

@property
def cancelled(self) -> bool:
    """True when promise has been cancelled."""

@property
def value(self) -> tuple:
    """
    Promise fulfillment value as (args, kwargs) tuple.
    Only available when ready=True and failed=False.
    """

@property
def reason(self) -> Exception:
    """
    Exception that caused promise failure.
    Only available when failed=True.
    """

@property
def listeners(self) -> list:
    """List of pending callback promises."""

Usage Examples:

p = promise(lambda x: x * 2)

print(p.ready)     # False
print(p.failed)    # False
print(p.cancelled) # False

p(10)

print(p.ready)     # True
print(p.value)     # ((20,), {})
print(p.failed)    # False

Advanced Usage

Weak References

Use weak references to prevent memory leaks in long-lived promise chains:

# Weak reference prevents circular references
method_promise = promise(obj.method, weak=True)

Performance Optimization

For promises that don't need return values:

# Ignore return value for better performance
logging_promise = promise(logger.info, ignore_result=True)

Complex Chaining

Promises can be chained in complex patterns:

# Conditional chaining
def process_data(data):
    p = promise(validate_data, args=(data,))
    
    # Chain different operations based on validation
    p.then(lambda valid_data: 
        transform_data(valid_data) if valid_data else None
    ).then(lambda result:
        save_result(result) if result else None
    )
    
    return p

Install with Tessl CLI