tessl/pypi-returns

Functional programming library providing type-safe containers for error handling, side effects, and composable operations with monadic patterns.

—

Pending

Overview

Eval results

Files

Unsafe Operations

Name: tessl/pypi-returns
Author: tessl

Escape hatch utilities that allow breaking out of the functional programming world when necessary. These operations bypass the safety guarantees of containers and should be used judiciously.

Capabilities

IO Container Execution

Extract values from IO containers by executing their side effects.

def unsafe_perform_io(container: IO[T]) -> T:
    """Execute IO container and extract value (unsafe!)"""

Usage examples:

from returns.io import IO, impure
from returns.unsafe import unsafe_perform_io

# Create IO container
@impure
def read_config_file() -> dict:
    with open("config.json") as f:
        return {"database_url": "localhost", "debug": True}

io_config = read_config_file()  # IO[dict]

# Extract value unsafely  
config = unsafe_perform_io(io_config)  # dict: {"database_url": "localhost", "debug": True}

# Use extracted config
print(f"Connecting to: {config['database_url']}")

When to Use Unsafe Operations

Unsafe operations should only be used in specific scenarios:

1. Application Boundaries

At the edges of your application where you need to interface with the outside world:

from returns.io import IO, impure
from returns.unsafe import unsafe_perform_io
from returns.result import safe

@impure
def setup_database() -> str:
    # Side effect: establish database connection
    return "Database connected"

@safe
def process_user_data(data: dict) -> dict:
    # Pure business logic
    return {"processed": True, "user_id": data["id"]}

# Application entry point
def main():
    # Unsafe: execute side effects at application boundary
    db_status = unsafe_perform_io(setup_database())
    print(db_status)
    
    # Safe: pure business logic
    result = process_user_data({"id": 123})
    print(result)

if __name__ == "__main__":
    main()  # Only place where unsafe operations happen

2. Testing and Development

Extract values for testing purposes:

from returns.io import IO, impure
from returns.unsafe import unsafe_perform_io

@impure  
def fetch_user_data(user_id: int) -> dict:
    # In real app: HTTP request
    return {"id": user_id, "name": "Test User"}

def test_user_processing():
    # Extract value for testing
    io_user = fetch_user_data(123)
    user_data = unsafe_perform_io(io_user)
    
    assert user_data["id"] == 123
    assert "name" in user_data

3. Legacy Code Integration

When integrating with existing non-functional codebases:

from returns.io import IO, impure  
from returns.unsafe import unsafe_perform_io

# Legacy function expects regular values
def legacy_email_sender(config: dict, message: str) -> bool:
    # Existing imperative code
    print(f"Sending '{message}' with config: {config}")
    return True

@impure
def load_email_config() -> dict:
    return {"smtp_host": "smtp.example.com", "port": 587}

# Bridge between functional and imperative code
def send_notification(message: str) -> bool:
    config_io = load_email_config()
    config = unsafe_perform_io(config_io)  # Extract for legacy function
    return legacy_email_sender(config, message)

4. Performance Critical Sections

When container overhead is prohibitive (use sparingly):

from returns.io import IO, impure
from returns.unsafe import unsafe_perform_io

@impure
def read_large_dataset() -> list[int]:
    # Expensive I/O operation
    return list(range(1000000))

def process_dataset_fast():
    # Sometimes you need raw performance
    dataset_io = read_large_dataset()
    dataset = unsafe_perform_io(dataset_io)
    
    # Time-critical processing on raw data
    return sum(x * x for x in dataset)

Safety Guidelines

When using unsafe operations, follow these guidelines:

1. Minimize Usage

Keep unsafe operations at the absolute minimum:

# BAD: Unsafe operations scattered throughout
def bad_example():
    config = unsafe_perform_io(load_config())
    users = unsafe_perform_io(fetch_users(config))
    for user in users:
        result = unsafe_perform_io(process_user(user))
        unsafe_perform_io(save_result(result))

# GOOD: Single unsafe operation at boundary  
def good_example():
    # Pure functional pipeline
    pipeline = (
        load_config()
        .bind(fetch_users)
        .bind(lambda users: sequence([process_user(u) for u in users]))
        .bind(save_all_results)
    )
    
    # Single unsafe execution
    unsafe_perform_io(pipeline)

2. Document Usage

Always document why unsafe operations are necessary:

def critical_system_shutdown():
    """
    Performs emergency system shutdown.
    
    Uses unsafe_perform_io because:
    - Critical path where container overhead is prohibitive
    - Must execute immediately without functional composition
    - Called only in emergency scenarios
    """
    shutdown_io = initiate_shutdown()
    unsafe_perform_io(shutdown_io)  # Required for immediate execution

3. Isolate Side Effects

Keep unsafe operations isolated from pure code:

# Pure functional core
def calculate_metrics(data: list[dict]) -> dict:
    return {"total": len(data), "average": sum(d["value"] for d in data) / len(data)}

# Unsafe shell around pure core
def metrics_service():
    data_io = fetch_data_from_api()
    data = unsafe_perform_io(data_io)  # Isolated unsafe operation
    
    metrics = calculate_metrics(data)  # Pure calculation
    
    save_metrics_io = save_metrics(metrics)
    unsafe_perform_io(save_metrics_io)  # Isolated unsafe operation