What is flusso?

flusso is a library for Python that aims to safely handle exceptions and missing values, similar to how Rust handles them with its Option and Result types.

In short, Flusso empowers you to craft Python code that is:

• Free from None values
• Devoid of exceptions

In python, None represents intentionally missing values and exceptions are used for handling errors.

Python skips using missing values and exceptions can lead to issues and bugs like:

• NoneType errors
• runtime errors
• unexpected behaviour
• unhandled exceptions
• sensitive data leakages through exceptions
• race conditions
• and so on.

Instead, Python provides two special generic Option and Result to deal with the above cases.

flusso implements the Option & Result types for python.

Why should you use flusso?

There are already several excellent libraries that implement functional patterns in python. Why flusso?

These libraries are usually general-purpose toolkits aiming to implement all the functional programming patterns and abstractions. flusso has a more focused goal. We wanted a library specifically to dominate safely handle exceptions and missing values (None). The same way as it’s implemented in Rust.

Other distinguishing features of flusso:

• Zero dependencies: flusso has no external dependencies.
• Practical: • Rather than bore you with all the Monad / Category theory talk, we focus on the practical applications of Monads in a way you can use today. Just as you don’t need to understand group theory to do basic arithmetic, you don’t need to understand monad theory to use flusso.
• Leverages Python's pattern matching for concise and expressive code
• Provides an intuitive way to handle optional values and error handling
• Eliminates the need for writing code with None or exceptions
• Compatible with the latest Python features and best practices
• Fully typed with annotations, following PEP 484 guidelines

Convinced?

Great! Let’s get started.

Installation

> pip install flusso

If you find this package useful, please click the star button ✨!

Table of contents

• Option[T]
  • Introduction
  • Basic usage
  • Option decorator
  • Benefits
• Result<T,E>
  • Introduction
  • Basic usage
  • Result decorator
  • Benefits
• AsyncResult<T,E>
  • Introduction
  • Basic usage
  • AsyncResult decorator
  • Benefits
• Utils
  • Flatten
  • Pattern matching
  • Do Notation

Option<T>

Introduction

“Null has led to innumerable errors, vulnerabilities, and system crashes, which have probably caused a billion dollars of pain and damage in the last forty years.” - Tony Hoare, the inventor of null

None values can be difficult to detect and handle correctly. When a None value is encountered, it may not be immediately clear why it is there or how to handle it. This can lead to bugs that are hard to diagnose and fix.

Another problem with None values is that they can cause runtime errors if they are not properly handled. For example, if a program attempts to access a property of an object that is None, it will often raise a NullPointerException or similar error. These errors can be difficult to anticipate and debug, especially if they occur deep in the codebase or if there are many layers of abstraction involved.

To avoid these problems, we use Option as an alternative way of representing the absence of a value or the lack of an object reference.

A brief background on Option Monad

A monad is a design pattern that allows for the creation of sequenced computations, or "actions," that can be combined in a predictable way.

The option monad is a specific type of monad that represents computations that may or may not return a value.

Option monad types allow for the explicit representation of the possibility of a missing value, and they provide methods for handling these cases in a predictable and composable way.

The option monad is usually implemented as an algebraic data type with two cases: Some and Nothing. The Some case represents a computation that has a value, and it is parameterized by the type of the value. The Nothing case represents a computation that has a missing value.

Option monad helps us safely handle missing values in a predictable and composable way without being afraid of the null pointer exception, runtime errors, and unexpected behaviour in our code.

⬆️ Back to top

Basic usage

Example I

Let’s start with a common example you see in many codebases today.

  class User:
      def __init__(self, id: int, fullname: str, username: str):
          self.id = id
          self.fullname = fullname
          self.username = username

      users = [
          User(1, "Leonardo Da Vinci", "leo"),
          User(2, "Galileo Galilei", "gaga")
      ]

  def get_user(id: int) -> Union[User, None]:
      return next((user for user in users if user.id == id), None)

  def get_user_name(id: int) -> Union[str, None]:
      user = get_user(id)
      if user is None:
          return None
      return user.username

  username = get_user_name(1)

  if username is not None:
      print(username)
  else:
      print("User not found")

This code focuses on telling the computer how to perform a task, step by step. It involves specifying the sequence of actions that the computer should take and the specific operations it should perform at each step.

The code also uses None to define missing values. Even with a simple example like this, it’s not immediately clear where the None is coming from when we check if the username is None. In large codebases, this can be a nightmare to diagnose and fix.

However, since this code style is more familiar and follows a more traditional control flow, it can be easier to understand for most programmers.

Let's rewrite this with a declarative style using flusso

  from flusso.option import Option, Some, Nothing

  class User:
      def __init__(self, id: int, fullname: str, username: str):
          self.id = id
          self.fullname = fullname
          self.username = username

  users = [
      User(1, "Leonardo Da Vinci", "leo"),
      User(2, "Galileo Galilei", "gaga")
  ]

  def get_user(id: int) -> Option[User]:
      user = next((user for user in users if user.id == id), Nothing)
      return Some(user)

  def get_username(id: int) -> Option[str]:
      return get_user(id).fmap(lambda user: user.username)


  match get_username(1):
      # Matches any `Some` instance and binds its value to the `username` variable
      case Some(username):
          print('User found: {0}'.format(username))

      # Matches `Nothing` instance
      case Nothing:
          print('User not found!')

  # Alternatively
  # if username.is_some():
  #     print(username.unwrap())
  # else:
  #     print("User not found")

This code style focuses on describing the input (the user's ID) and the desired output (the username). The match function handles the case where the user is not found by providing a default value (in this case, a message saying "User not found").

With flusso, we have successfully handled missing values in a predictable and composable way.

Example II

Let’s look at another example of using option to handle optional values.

if the value of an object can be empty or optional like the middle_nameof User in the following example, we can set its data type as an Optiontype.

 from flusso.option import Option, Some, Nothing

  def get_full_name(first_name: str, middle_name: Option[str], last_name: str) -> str:
      match(middle_name):
          case Some(mname):
              print(f"{first_name} {mname} {last_name}")

          # Matches `Nothing` instance
          case Nothing:
              print(f"{first_name} {last_name}")

  get_full_name("Galileo", None, "Galilei"); # Galileo Galilei
  get_full_name("Leonardo", Some("Da"), "Vinci"); # Leonardo Da Vinci

Let’s look at another example by chaining calculations

  def sine(x: float) -> Option[float]:
          return math.sin(x)

  def cube(x: float) -> Option[float]:
      return x * x * x

  def inc(x: float) -> Option[float]:
      return x + 1

  def double(x: float) -> Option[float]:
      return x ** 2

  def divide(x: float, y: float) -> Option[float]:
      return x / y if y > 0 else Nothing

  def sineCubedIncDoubleDivideBy10(x: float):
      return (
          Some(x)
          .fmap(sine)
          .fmap(cube)
          .fmap(inc)
          .fmap(double)
          .fmap(lambda x: divide(x, 10))
      )

  match(sineCubedIncDoubleDivideBy10(30)):
      case Some(result):
          print(f"`Result is {result}")

      # Matches `Nothing` instance
      case Nothing:
          print("Please check your inputs")

⬆️ Back to top

Benefits

There are several reasons why you might want to use flusso.option in your code:

1. To avoid NoneType errors: As mentioned earlier, the Option is a way of representing optional values in a type-safe way. This can help you avoid NoneType errors by allowing you to explicitly handle the absence of a value in your code.
2. To make your code more readable: Using the Option can make your code more readable, because it clearly indicates when a value may be absent. This can make it easier for other developers to understand your code and can reduce the need for comments explaining how None values are handled.
3. To improve code reliability: By explicitly handling the absence of a value, you can make your code more reliable and less prone to runtime errors.
4. To improve code maintainability: Using the Option can make your code more maintainable, because it encourages a clear and explicit handling of optional values. This can make it easier to modify and extend your code in the future.
5. To make you write code that is more declarative and less imperative. This can make your code easier to understand and test.

If you find this package useful, please click the star button ✨!

⬆️ Back to top

Option decorator

When working with functions that return Optional values, it's common to encounter numerous if x is not None: checks in your code. Flusso comes to the rescue with the @option decorator, which simplifies this process by converting functions that return Optional values to return Option instances instead.

Here's how to use the @option decorator in Flusso:

  from typing import Optional
  from flusso.option import Option, Some, option

  @option
  def find_even_number(numbers: list[int]) -> Optional[int]:
      for number in numbers:
          if number % 2 == 0:
              return number
      return None

  result: Option[int] = find_even_number([1, 3, 5, 7, 2, 4])
  assert result == Some(2)

Result[T, E]

Introduction

Exceptions are a mechanism for handling errors and exceptional circumstances in many programming languages. When an exception is thrown, the normal flow of control in the program is interrupted, and the program tries to find an exception handler to handle the exception. If no appropriate exception handler is found, the program may crash or produce unexpected results.

There are several problems with using exceptions for error handling:

1. Exceptions can be difficult to anticipate: Exceptions can be thrown anywhere in the code, making it difficult to anticipate where they might occur and how to handle them. This can make it hard to write robust, reliable code.
2. Exceptions can be hard to debug: When an exception is thrown, the normal flow of control in the program is interrupted, making it difficult to trace the cause of the exception and fix the error.
3. Exceptions can make code harder to read: When exceptions are used extensively, the code can become cluttered with try-except blocks, making it harder to understand what is happening.
4. Exceptions can have performance overhead: Throwing and catching exceptions can have a significant performance overhead, especially if they are used extensively.

Result is a way to handle errors and exceptions in a more predictable and structured way. Instead of using exceptions, the result type uses a variant-based approach, with separate Ok and Err variants representing successful and unsuccessful computations, respectively. This allows for more predictable error handling and makes it easier to anticipate and handle errors in the code.

This provides a more predictable and structured approach to error handling, which can improve the reliability, readability, performance, and composability of code.

⬆️ Back to top

Basic usage

Let’s start with an example of how you might use exceptions in Python.

  def divide(numerator: float, denominator: float) -> float:
      if denominator == 0:
          raise ZeroDivisionError("Division by zero")
      return numerator / denominator

  def add_one(x: float) -> float:
      return x + 1

  def compute(numerator: float, denominator: float) -> float:
      try:
          result = divide(numerator, denominator)
          result = add_one(result)
          return result
      except ZeroDivisionError:
          return 0

  print(compute(10, 2)) # 6.0
  print(compute(10, 0)) # 0

In this example, the divide function throws an exception if the denominator is zero, and the compute function uses a try-except block to handle the exception and return zero if it occurs.

Let rewrite this with a declarative style using flusso

  def divide(numerator: float, denominator: float) -> Result[float,str]:
      if denominator == 0:
          return Err("Division by zero")
      return Ok(numerator / denominator)

  def add_one(x: float) -> Result[float,str]:
      return Ok(x + 1)

  def compute(numerator: float, denominator: float) -> Result[float,str]:
      return divide(numerator, denominator).and_then(add_one)

  match(compute(10, 2)):
      case Ok(result):
          print(result)
      case Err(error):
          print(f"Error #{error}")

  match(compute(10, 0)):
      case Ok(result):
          print(result)
      case Err(error):
          print(f"Error #{error}")

In this example, the divide function returns a result type representing the result of a division operation. If the denominator is zero, it returns an Err variant with an error message. If the denominator is non-zero, it returns an Ok variant holding the result of the division.

The add_one function takes a number and returns a result type representing the result of adding one to that number. In this case, it always returns an Ok variant.

and_then is used to chain the divide and add_one functions together, passing the result of the divide function as input to the add_one function. If the divide function returns an Err variant, and_then short-circuits the chain and returns the Err variant immediately.

You can also use or_else to handle any errors that might occur in the computation. If the result type is an Err variant, the provided fallback function is called with the error as input and its result is returned.

  def compute(numerator: float, denominator: float) -> float:
      return(
          divide(numerator, denominator)
          .and_then(add_one)
          .or_else(lambda error: Ok(0))
          .unwrap()
      )

  print(compute(10, 2)) # 6.0
  print(compute(10, 0)) # 0.0

⬆️ Back to top

Result decorator

When working with functions that might raise exceptions, it's typical to see numerous try-except blocks scattered throughout your code. Flusso swoops in to save the day with the @result decorator, which streamlines this process by converting functions that raise exceptions into functions that return Result instances instead.

Here's how to use the @result decorator in Flusso:

  from flusso import Result, Ok, Err, result

  @result
  def divide_numbers(numerator: float, denominator: float) -> Union[float, str]:
      if denominator == 0:
          return "Division by zero is not allowed"
      return numerator / denominator

  result: Result[float, str] = divide_numbers(10, 2)
  assert result == Ok(5.0)

Benefits

There are several reasons why you might choose to use the flusso.result in your code:

1. Improved error handling: Result provides a structured way to handle errors and exceptions, allowing for more predictable and easy-to-reason-about code.
2. Improved code readability: By using Result, it is clear to anyone reading the code that a computation may or may not be successful, and what to do in each case. This can make the code easier to understand and maintain.
3. Improved code reliability: By using the Result, it is easier to ensure that errors and exceptions are properly handled and do not result in unexpected behavior or crashes.
4. Improved code composability: Result allows for the chaining of operations. This can make it easier to build up complex computations from simpler ones.

⬆️ Back to top

AsyncResult[T, E]

Introduction

AsyncResult is a utility class for working with asynchronous operations that may result in a success or an error. It is built on top of the Result class, which represents a synchronous operation's result. The AsyncResult class is useful for chaining, transforming, and handling results from asynchronous operations.

⬆️ Back to top

Basic usage

Let’s start with an example of how you might use exceptions in Python.

  from flusso.async_result import async_result, Ok, Err

  @async_result
  async def async_fetch_data(url: str) -> Dict[str, Any]:
      async with aiohttp.ClientSession() as session:
          async with session.get(url) as response:
              if response.status != 200:
                  raise ValueError("Failed to fetch data")
              return await response.json()

  async def fetch():
      url = "https://jsonplaceholder.typicode.com/todos/1"
      async_result = await async_fetch_data(url)

      match async_result._result:
          case Ok(value):
              print("Fetched data:", value)
          case Err(error):
              print("Error fetching data:", error)

      # Alternatively
      # if async_result.is_ok():
      #     print("Fetched data:", await async_result.unwrap())
      # else:
      #     print("Error fetching data:", await async_result.unwrap_err())

  asyncio.run(fetch())

AsyncResult provides several methods for working with and transforming the result:

• fmap(fn): Transform the successful value using an asynchronous or synchronous function.
• fmap_err(fn): Transform the error value using an asynchronous or synchronous function.
• and_then(fn): Chain an asynchronous operation that returns a new AsyncResult if the current result is a success.
• or_else(fn): Chain an asynchronous operation that returns a new AsyncResult if the current result is an error.

fmap

Transform the successful value using an asynchronous or synchronous function. If the AsyncResult is an error, the function won't be called, and the original error will be propagated.

    async def async_multiply(value, factor):
        return value * factor

    async_result = AsyncResult(Ok(5))
    mapped_result = await async_result.fmap(async_multiply, 2)  # Ok(10)

fmap_err

Transform the error value using an asynchronous or synchronous function. If the AsyncResult is a success, the function won't be called, and the original success value will be propagated.

    async def async_error_message(code):
        return f"Error {code}"

    async_result = AsyncResult(Err(404))
    mapped_error_result = await async_result.fmap_err(async_error_message)  # Err("Error 404")

and_then

Chain an asynchronous operation that returns a new AsyncResult if the current result is a success. If the AsyncResult is an error, the function won't be called, and the original error will be propagated.

    async def async_double(value):
        return AsyncResult(Ok(value * 2))

    async_result = AsyncResult(Ok(5))
    chained_result = await async_result.and_then(async_double)  # Ok(10)

or_else

Chain an asynchronous operation that returns a new AsyncResult if the current result is an error. If the AsyncResult is a success, the function won't be called, and the original success value will be propagated.

    async def async_handle_error(error):
        return AsyncResult(Ok(f"Recovered from {error}"))

    async_result = AsyncResult(Err("an error"))
    handled_result = await async_result.or_else(async_handle_error)  # Ok("Recovered from an error")

⬆️ Back to top

AsyncResult decorator

When working with asynchronous functions that might raise exceptions, it's common to see many try-except blocks combined with async-await syntax, which can complicate your code. Flusso comes to the rescue with the @async_result decorator, which simplifies this process by converting asynchronous functions that raise exceptions into functions that return AsyncResult instances instead.

Here's how to use the @async_result_decorator in Flusso:

  @async_result
  # Apply the @async_result decorator to your asynchronous functions that might raise exceptions:
  async def async_fetch_data(url: str) -> Result[str, Exception]:
      async with aiohttp.ClientSession() as session:
          async with session.get(url) as response:
              if response.status != 200:
                  raise ValueError("Failed to fetch data")
              return await response.text()

  # When calling the decorated function, it will return an AsyncResult object instead of raising an exception:
  async def main():
      url = "https://example.com/data"
      async_result = await async_fetch_data(url)

      if async_result.is_ok():
          print("Fetched data:", await async_result.unwrap())
      else:
          print("Error fetching data:", await async_result.unwrap_err())

Benefits

There are several reasons why you might choose to use the flusso.async_result in your code:

1. Cleaner code: By using AsyncResult, you can minimize the need for nested try-except blocks and async-await syntax, leading to more readable and maintainable code.

2. Composable error handling: The AsyncResult class allows you to chain error handling and transformation functions, making it easy to compose complex error handling logic in a declarative manner.

3. Separation of concerns: AsyncResult helps you separate the success and error cases, ensuring that your functions are focused on their primary responsibilities and not cluttered with error handling logic.

4. Type safety: AsyncResult is a generic type that allows you to specify the success and error types, providing better type-checking and making it easier to catch potential issues during development.

5. Flexible error transformation: The AsyncResult class provides methods like fmap, fmap_err, and_then, and or_else, which allow you to transform, chain, and handle errors in a flexible way.

6. Easier testing: Since functions that return AsyncResult objects no longer raise exceptions directly, testing various scenarios and edge cases becomes simpler and more intuitive.

7. Consistent error handling: By using AsyncResult throughout your code, you can establish a consistent approach to error handling, making your codebase more robust and easier to understand.

8. Integration with Result: AsyncResult is designed to work seamlessly with Flusso's Result class, allowing you to handle both synchronous and asynchronous operations with a consistent API.

Utils

Flatten

To remove many levels of nesting:

# With Option
print(flatten(Some(Some(Nothing))))  # Nothing
print(flatten(Some("some1")))        # Some("some1")
print(flatten(Nothing))              # Nothing

# With Result
print(flatten(Ok(Ok(Ok(Ok(Ok(Ok(Ok(10)))))))) # Ok(10)
print(flatten(Ok(Ok(Err("error1")))))         # Err("error1")
print(flatten(Ok("ok1")))                     # Ok("ok1")
print(flatten(Err("error1")))                 # Err("error1")

Pattern matching

When using pattern matching with Flusso, you can match on Some, Nothing, Ok, and Err cases and extract the inner values accordingly. This approach allows you to focus on the logic of your application, making your code more maintainable and easier to understand.

  def process_data(data: dict) -> Option[int]:
      return Some(data["value"]) if "value" in data else Nothing

  def calculate_percentage(value: int, total: int) -> Result[float, str]:
      return Err("Total cannot be zero") if total == 0 else Ok((value / total) * 100)

  # Example data
  data = {"value": 10, "total": 50}

  # Using pattern matching with Flusso's Option and Result
  opt_value = process_data(data)
  total = data["total"]

  match opt_value:
      case Some(value):
          result = calculate_percentage(value, total)
          match result:
              case Ok(percentage):
                  print(f"The percentage is {percentage}%")
              case Err(err_msg):
                  print(f"Error: {err_msg}")
      case Nothing:
          print("Value not found in the data")

Do Notation

Flusso provides a simple way to handle chained computations using the Do notation. The do notation offers a more intuitive and readable Imperative-style syntax for working with monadic types like Result, Option, and AsyncResult, allowing you to write sequential-like code while retaining the powerful error handling and encapsulation features of monads. With Flusso's implementation of Do notation, you can easily manage multiple steps in a computation while maintaining clean, readable code. Here's how to use the Do notation for Option, Result, and AsyncResult types in Flusso.

Option example:

    def add_numbers(a: int, b: int) -> Option[int]:
        return Some(a + b)

    def multiply_numbers(a: int, b: int) -> Option[int]:
        return Some(a * b)

    x = 2
    y = 3

    with (
        Option.do(add_numbers(x, y)) as a,
        Option.do(multiply_numbers(a, x)) as b,
        Option.do(add_numbers(b, y)) as c,
    ):
        result = Some(c)

    assert result == Some(((x + y) * x) + y)

Result example:

    def add_numbers_result(a: int, b: int) -> Result[int, str]:
        return Ok(a + b)

    def multiply_numbers_result(a: int, b: int) -> Result[int, str]:
        return Ok(a * b)

    x = 2
    y = 3

    with (
        Result.do(add_numbers_result(x, y)) as a,
        Result.do(multiply_numbers_result(a, x)) as b,
        Result.do(add_numbers_result(b, y)) as c,
    ):
        result = Ok(c)

    assert result == Ok(((x + y) * x) + y)

Similarly, the Do notation can also be used with Result instances. It provides a clean, functional way to handle chained computations and potential errors.

AsyncResult example:

    @async_result
    async def async_fetch_data(url: str) -> Dict[str, Any]:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status != 200:
                    raise ValueError("Failed to fetch data")
                return await response.json()

    async def fetch_do():
        url = "https://jsonplaceholder.typicode.com/todos/1"

        async with (
            AsyncResult.do(fetch_data=async_fetch_data(url)) as fetch_result
        ):

            match fetch_result._result:
                case Ok(data):
                    print("Fetched data:", data)
                case Err(error):
                    print("Error fetching data:", error)

    asyncio.run(fetch_do())

By using the Do notation in Flusso, you can write more expressive and maintainable code when working with Option, Result, and AsyncResult instances.

Coming soon

• Asynchronous support: Integrate seamless handling of asynchronous operations with Result instances, making it even more convenient to work with coroutines.
• Comprehensive documentation and examples: Expand the library's documentation and provide more practical examples to help users get the most out of Flusso.
• Enhancing the Do notation to allow more fine-grained error handling or recovery, such as customizing the behavior for specific error cases or providing default values when certain errors occur.
• Improving the error messages produced by the Do notation to provide more context and clarity when something goes wrong.
• Custom data types: Provide an easy-to-use interface for creating custom data types that adhere to Flusso's functional programming principles and type safety requirements.

⬆️ Back to top

If you find this package useful, please click the star button ✨!