fp-ops 0.2.4

an ergonomic wrapper around the expression library

FP-Ops: Functional Programming Operations for Python

FP-Ops is a functional programming library for Python that lets you convert you functions into composable operations.

Features

• Composition as a First-class Citizen: Build complex pipelines using simple operators like >>, &, and |
• Context Awareness: Pass context through operation chains with automatic validation
• Async-First: Designed for asynchronous operations from the ground up
• Type Safety: Comprehensive type hints for better IDE support and code safety
• Functional Patterns: Implements common functional programming patterns like map, filter, and reduce
• Lazy Execution: Only execute operations when the result is needed
• Composition is associative: (a >> b) >> c == a >> (b >> c)

Installation

pip install fp-ops

Getting Started

Here's a simple example to get you started:

from fp_ops.operator import operation
import asyncio

# Define some operations
@operation
async def get_user(user_id: int) -> dict:
    # Simulate API call
    return {"id": user_id, "name": "John Doe", "age": 30}

@operation
async def format_user(user: dict) -> str:
    return f"User {user['name']} is {user['age']} years old"

# Compose operations
get_and_format = get_user >> format_user

get_and_format(1)

Key Concepts

Operations

The core concept in FP-Ops is the Operation class. An operation wraps an async function and provides methods for composition using operators:

• >> (pipeline): Passes the result of one operation to the next
• & (parallel): Executes operations in parallel and returns all results
• | (alternative): Tries the first operation and falls back to the second if it fails

Placeholders

You can use the placeholder _ to specify where the result of a previous operation should be inserted:

from fp_ops.placeholder import _

# Define operations
@operation
async def double(x: int) -> int:
    return x * 2

@operation
async def add(x: int, y: int) -> int:
    return x + y

# These are equivalent:
pipeline1 = double >> (lambda x: add(x, 10))
pipeline2 = double >> add(_, 10)

Context Awareness

Operations can be context-aware, allowing you to pass contextual information through the pipeline:

from fp_ops.operator import operation
from fp_ops.context import BaseContext
from pydantic import BaseModel

class UserContext(BaseContext):
    auth_token: str
    user_id: int

@operation(context=True, context_type=UserContext)
async def get_user_data(context: UserContext) -> dict:
    return {"id": context.user_id, "name": "Jane Doe"}

# Initialize context
context = UserContext(auth_token="abc123", user_id=42)

# Execute with context
result = await get_user_data(context=context)

Advanced Usage

Error Handling

FP-Ops uses the Result type for robust error handling:

@operation
async def divide(a: int, b: int) -> int:
    if b == 0:
        raise ValueError("Division by zero")
    return a / b

# Handle errors with default values
safe_divide = divide.default_value(0)

# Or with custom error handling
safe_divide = divide.catch(lambda e: 0 if isinstance(e, ValueError) else -1)

Composition Functions

Besides operators, FP-Ops provides various composition functions:

from fp_ops.composition import sequence, pipe, parallel, fallback, map, transform

# Run operations in sequence and collect all results
results = await sequence(op1, op2, op3)

# Complex pipelines with conditional logic
pipeline = pipe(
    op1,
    lambda x: op2 if x > 10 else op3,
    op4
)

# Run operations in parallel
combined = await parallel(op1, op2, op3)

# Try operations until one succeeds
result = await fallback(op1, op2, op3)

# Apply an operation to each item in an iterable
# (e.g., transforming [1, 2, 3] to [2, 3, 4] if item_op increments by 1)
mapped_results = await map(item_op, max_concurrency=5)([item1, item2, item3]) 

# Transform the output of a single operation
transformed_result = await transform(op1, lambda x: x * 2)

Higher-Order Flow Operations

FP-Ops provides utilities for creating higher-order operations:

from fp_ops.flow import branch, attempt, retry, wait, loop_until

# Conditional branching
conditional = branch(
    lambda x: x > 0,
    positive_op,
    negative_op
)

# Retry an operation
resilient_op = retry(flaky_operation, max_retries=3, delay=0.5)

# Loop until a condition is met
counter = loop_until(
    lambda x: x >= 10,
    lambda x: x + 1,
    max_iterations=20
)

API Reference

Core Classes

• Operation: The main class representing a composable asynchronous operation
• BaseContext: Base class for all operation contexts
• Placeholder: Used to represent where a previous result should be inserted

Decorators

• @operation: Convert a function to an Operation
• @operation(context=True, context_type=MyContext): Create a context-aware operation

Operators

• op1 >> op2: Pipeline composition
• op1 & op2: Parallel execution
• op1 | op2: Alternative execution

Methods

• operation.transform(func): Apply a transformation to the output
• operation.filter(predicate): Filter the result using a predicate
• operation.bind(binder): Bind to another operation
• operation.catch(handler): Add error handling
• operation.default_value(default): Provide a default value for errors
• operation.retry(attempts, delay): Retry the operation
• operation.tap(side_effect): Apply a side effect without changing the value

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is licensed under the MIT License - see the LICENSE file for details.