As the complexity of code increases, there are generally nested calls that want to exit at points. Generally, the common methods to pass that signal up to parent functions are,
- null returns
- monads
- raise and catch errors (with try-catches)
each with its own advantage and disadvantage. To more easily compare them, let's take an inner and outer function,
def inner_function(number: int) -> int:
"""Function that doubles the value or signals that the flow should change."""
if number == 4: # Stop condition for sake of example
... # We want to signal up here
return number * 2
def outer_function() -> int:
"""Larger function that calls the inner function multiple times."""
first_response = inner_function(2)
second_response = inner_function(4) # Last line that should execute
third_response = inner_function(8)
return first_response + second_response + third_response
NOTE: the example calls the one function multiple times sequentially to emulate (potentially) different functions. In a real-world case you might have several different methods being called with additional sub-calls. Again, calling the same one multiple times is just easier for our purposes.
If the return value of the inner function would never normally be null, we could just pass back None
.
This would look like,
def inner_function(number: int) -> int | None:
"""Function that doubles the value or signals that the flow should change."""
if number == 4: # Stop condition for sake of example
return None
return number * 2
def outer_function() -> int:
"""Larger function that calls the inner function multiple times."""
if (first_response := inner_function(2)) is None:
return 42
if (second_response := inner_function(4)) is None:
return 42
if (third_response := inner_function(8)) is None:
return 42
return first_response + second_response + third_response
In this example, we only want to jump one call level up, so it's easy. However, having to do this for multiple levels gets cumbersome. Additionally, we have no ability for the inner function to choose the return value.
To add the ability for the inner function to specify the return value, we can change the example slightly to use monads. Which allow us to always specify a return value and whether it was the result of an error or similar.
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class Monad:
payload: int
is_error: bool = False
def inner_function(number: int) -> Monad:
"""Function that doubles the value or signals that the flow should change."""
if number == 4: # Stop condition for sake of example
return Monad(payload=42, is_error=True)
return Monad(number * 2)
def outer_function() -> int:
"""Larger function that calls the inner function multiple times."""
if (first_response := inner_function(2)).is_error:
return first_response.payload
if (second_response := inner_function(4)).is_error:
return second_response.payload
if (third_response := inner_function(8)).is_error:
return third_response.payload
return first_response.payload + second_response.payload + third_response.payload
If we don't want to check the return each time for whether it contains a signal, we could raise an exception. This cleans up our code, but changes the appearance of what we're doing (i.e. could mislead someone not expecting it).
def inner_function(number: int) -> int:
"""Function that doubles the value or signals that the flow should change."""
if number == 4: # Stop condition for sake of example
... # Signal up
return number * 2
def outer_function() -> int:
"""Larger function that calls the inner function multiple times."""
try:
first_response = inner_function(2)
second_response = inner_function(4)
third_response = inner_function(8)
except ValueError: # Error indicates we want an early return
return 42
return first_response + second_response + third_response
Finally, we can use divert and (effectively) take a hybrid approach,
from divert.flow import custom_flow_edge, divert
def inner_function(number: int) -> int:
"""Function that doubles the value or signals that the execution flow should change."""
if number == 4: # Stop condition for sake of example
divert() # If we want to specify the return value here we can instead use `payload_to_edge(VALUE)`
return number * 2
@custom_flow_edge(default_return=42)
def outer_function() -> int:
"""Larger function that calls the inner function multiple times."""
first_response = inner_function(2)
second_response = inner_function(4)
third_response = inner_function(8)
return first_response + second_response + third_response