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

Context Operations

Name: tessl/pypi-returns
Author: tessl

Context-dependent computations using the Reader monad pattern for dependency injection and environment passing. These containers enable clean dependency management without explicit parameter threading through function calls.

Capabilities

RequiresContext (Reader)

The basic Reader monad that represents a computation requiring some context/dependencies to produce a value.

class RequiresContext[T, Deps]:
    """Reader monad for context-dependent computations"""
    def __init__(self, func: Callable[[Deps], T]): ...
    def __call__(self, deps: Deps) -> T: ...
    def bind(self, func: Callable[[T], RequiresContext[U, Deps]]) -> RequiresContext[U, Deps]: ...
    def map(self, func: Callable[[T], U]) -> RequiresContext[U, Deps]: ...
    def apply(self, wrapped_func: RequiresContext[Callable[[T], U], Deps]) -> RequiresContext[U, Deps]: ...
    
    @classmethod
    def ask(cls) -> RequiresContext[Deps, Deps]:
        """Get the current context"""
        
    @classmethod  
    def from_value(cls, value: T) -> RequiresContext[T, Deps]:
        """Lift a value into the context"""

# Type alias
Reader[T, Deps] = RequiresContext[T, Deps]

Usage examples:

from returns.context import RequiresContext, Reader

# Define dependencies
class Config:
    def __init__(self, db_url: str, api_key: str):
        self.db_url = db_url
        self.api_key = api_key

# Create context-dependent functions
def get_db_connection(config: Config) -> str:
    return f"Connected to {config.db_url}"

def get_api_client(config: Config) -> str:
    return f"API client with key {config.api_key}"

# Compose operations
def setup_services() -> Reader[tuple[str, str], Config]:
    return (
        RequiresContext(get_db_connection)
        .bind(lambda db: RequiresContext(get_api_client).map(lambda api: (db, api)))
    )

# Run with dependencies
config = Config("postgresql://localhost", "secret")
db_conn, api_client = setup_services()(config)

RequiresContextResult (ReaderResult)

Combines Reader monad with Result for context-dependent computations that can fail.

class RequiresContextResult[T, E, Deps]:
    """Reader + Result combination"""
    def __init__(self, func: Callable[[Deps], Result[T, E]]): ...
    def __call__(self, deps: Deps) -> Result[T, E]: ...
    def bind(self, func: Callable[[T], RequiresContextResult[U, E, Deps]]) -> RequiresContextResult[U, E, Deps]: ...
    def map(self, func: Callable[[T], U]) -> RequiresContextResult[U, E, Deps]: ...
    def apply(self, wrapped_func: RequiresContextResult[Callable[[T], U], E, Deps]) -> RequiresContextResult[U, E, Deps]: ...
    def alt(self, func: Callable[[E], RequiresContextResult[T, F, Deps]]) -> RequiresContextResult[T, F, Deps]: ...
    def lash(self, func: Callable[[E], RequiresContextResult[T, F, Deps]]) -> RequiresContextResult[T, F, Deps]: ...
    
    @classmethod
    def ask(cls) -> RequiresContextResult[Deps, E, Deps]:
        """Get the current context"""
        
    @classmethod
    def from_success(cls, value: T) -> RequiresContextResult[T, E, Deps]:
        """Lift successful value into context"""
        
    @classmethod  
    def from_failure(cls, error: E) -> RequiresContextResult[T, E, Deps]:
        """Lift failure into context"""

# Type aliases
ReaderResult[T, E, Deps] = RequiresContextResult[T, E, Deps]
ReaderResultE[T, Deps] = RequiresContextResult[T, Exception, Deps]

Usage examples:

from returns.context import RequiresContextResult, ReaderResult
from returns.result import Success, Failure

# Context-dependent operations that can fail
def validate_config(config: Config) -> Result[Config, str]:
    if not config.db_url:
        return Failure("Database URL required")
    if not config.api_key:
        return Failure("API key required")
    return Success(config)

def connect_database() -> ReaderResult[str, str, Config]:
    return RequiresContextResult(
        lambda config: validate_config(config).map(
            lambda cfg: f"Connected to {cfg.db_url}"
        )
    )

# Chain operations
def setup_with_validation() -> ReaderResult[str, str, Config]:
    return connect_database().bind(
        lambda conn: RequiresContextResult.from_success(f"Setup complete: {conn}")
    )

# Run with context
result = setup_with_validation()(config)  # Success("Setup complete: Connected to...")

RequiresContextIOResult (ReaderIOResult)

Combines Reader, IO, and Result for context-dependent side effects that can fail.

class RequiresContextIOResult[T, E, Deps]:
    """Reader + IO + Result combination"""
    def __init__(self, func: Callable[[Deps], IOResult[T, E]]): ...
    def __call__(self, deps: Deps) -> IOResult[T, E]: ...
    def bind(self, func: Callable[[T], RequiresContextIOResult[U, E, Deps]]) -> RequiresContextIOResult[U, E, Deps]: ...
    def map(self, func: Callable[[T], U]) -> RequiresContextIOResult[U, E, Deps]: ...
    def apply(self, wrapped_func: RequiresContextIOResult[Callable[[T], U], E, Deps]) -> RequiresContextIOResult[U, E, Deps]: ...
    def alt(self, func: Callable[[E], RequiresContextIOResult[T, F, Deps]]) -> RequiresContextIOResult[T, F, Deps]: ...
    
    @classmethod
    def ask(cls) -> RequiresContextIOResult[Deps, E, Deps]:
        """Get the current context"""

# Type aliases  
ReaderIOResult[T, E, Deps] = RequiresContextIOResult[T, E, Deps]
ReaderIOResultE[T, Deps] = RequiresContextIOResult[T, Exception, Deps]

RequiresContextFutureResult (ReaderFutureResult)

Combines Reader with Future and Result for async context-dependent operations.

class RequiresContextFutureResult[T, E, Deps]:
    """Reader + Future + Result combination"""
    def __init__(self, func: Callable[[Deps], FutureResult[T, E]]): ...
    def __call__(self, deps: Deps) -> FutureResult[T, E]: ...
    async def bind(self, func: Callable[[T], RequiresContextFutureResult[U, E, Deps]]) -> RequiresContextFutureResult[U, E, Deps]: ...
    async def map(self, func: Callable[[T], U]) -> RequiresContextFutureResult[U, E, Deps]: ...
    async def apply(self, wrapped_func: RequiresContextFutureResult[Callable[[T], U], E, Deps]) -> RequiresContextFutureResult[U, E, Deps]: ...
    async def alt(self, func: Callable[[E], RequiresContextFutureResult[T, F, Deps]]) -> RequiresContextFutureResult[T, F, Deps]: ...
    
    @classmethod
    def ask(cls) -> RequiresContextFutureResult[Deps, E, Deps]:
        """Get the current context"""

# Type aliases
ReaderFutureResult[T, E, Deps] = RequiresContextFutureResult[T, E, Deps]  
ReaderFutureResultE[T, Deps] = RequiresContextFutureResult[T, Exception, Deps]

Context Management Utilities

# Type alias for no dependencies
NoDeps = Any

# Pointfree operations for context
def bind_context(func: Callable[[T], RequiresContext[U, Deps]]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[U, Deps]]: ...

def modify_env(func: Callable[[Deps], NewDeps]) -> Callable[[RequiresContext[T, Deps]], RequiresContext[T, NewDeps]]: ...

Usage Patterns

Dependency Injection

from returns.context import RequiresContext

class Database:
    def get_user(self, user_id: int) -> dict:
        return {"id": user_id, "name": "John"}

class Logger:
    def log(self, message: str) -> None:
        print(f"LOG: {message}")

class Dependencies:
    def __init__(self, db: Database, logger: Logger):
        self.db = db
        self.logger = logger

def get_user_name(user_id: int) -> RequiresContext[str, Dependencies]:
    def impl(deps: Dependencies) -> str:
        deps.logger.log(f"Fetching user {user_id}")
        user = deps.db.get_user(user_id)
        return user["name"]
    return RequiresContext(impl)

# Usage
deps = Dependencies(Database(), Logger())
name = get_user_name(123)(deps)  # "John"

Configuration Management

from returns.context import RequiresContextResult
from returns.result import Success, Failure

class AppConfig:
    def __init__(self, environment: str, debug: bool):
        self.environment = environment
        self.debug = debug

def get_feature_flag(flag_name: str) -> RequiresContextResult[bool, str, AppConfig]:
    def impl(config: AppConfig) -> Result[bool, str]:
        if config.environment == "production":
            return Success(False)  # Disable in production
        elif config.environment == "development":
            return Success(True)   # Enable in development
        else:
            return Failure(f"Unknown environment: {config.environment}")
    return RequiresContextResult(impl)

# Chain operations
def should_use_new_feature() -> RequiresContextResult[bool, str, AppConfig]:
    return get_feature_flag("new_ui").bind(
        lambda enabled: RequiresContextResult.from_success(
            enabled and True  # Additional logic
        )
    )

Error Handling with Context

from returns.context import RequiresContextResult
from returns.pointfree import bind_context

def safe_operation(x: int) -> RequiresContextResult[int, str, AppConfig]:
    def impl(config: AppConfig) -> Result[int, str]:
        if config.debug and x < 0:
            return Failure("Debug mode: negative values not allowed")
        return Success(x * 2)
    return RequiresContextResult(impl)

# Compose with pointfree style
pipeline = bind_context(safe_operation)
result = pipeline(RequiresContextResult.from_success(5))

Context operations enable clean separation of business logic from infrastructure concerns, making code more testable and maintainable by explicitly managing dependencies through the type system.

Install with Tessl CLI