A light-service influenced project in Python.

Are you tired of 500 lines long Python code with conditionals, iterators, and function calls? Testing this logic is close to impossible, and with the lack of testing, you don't dare to touch it.

All complex logic can be decomposed into small functions, invoked sequentially. The functions should protect themselves from execution if a previous failure occurs, a more elegant solution exists: Railway-Oriented Programming.

Let's see how that looks with pyservice. There are two functions, one that adds 2 to the initial number, and one that adds 3. The data is carried over between the functions in an extended dictionary we call Context, just like how a conveyor belt would be used in an assembly line:

from pyservice import action, Context, Organizer


@action()
def add_two(ctx: Context) -> Context:
    number = ctx.get("n", 0)

    ctx["result"] = number + 2

    return ctx


@action()
def add_three(ctx: Context) -> Context:
    result = ctx["result"]

    ctx["result"] = result + 3

    return ctx


def test_can_run_functions():
    ctx = Context.make({"n": 4})
    organizer = Organizer([add_two, add_three])
    result_ctx = organizer.run(ctx)

    assert ctx.is_success
    assert result_ctx["result"] == 9

The Context is an extended dictionary, it stores failure and success states in it besides its key-value pairs. This is the "state" that is carried between the actions by the Organizer. All Organizers expose a run function that is responsible for executing the provided actions in order.

This is the happy path, but what happens when there is a failure between the two functions? I add a fail_context function that will fail the context with a message:

@action()
def fail_context(ctx: Context) -> Context:
    ctx.fail("I don't like what I see here")
    return ctx

The context will be in a failure state and only the first action will be executed as processing stops after the second action (4+2=6):

def test_can_run_functions_with_failure():
    ctx = Context.make({"n": 4})
    organizer = Organizer([add_two, fail_context, add_three])
    result_ctx = organizer.run(ctx)

    assert ctx.is_failure
    assert result_ctx["result"] == 6

Look at the actions, no conditional logic was added to them, the function wrapper protects the action from execution once it's in a failure state.

You can find these examples here.

But there is more to it!

You can define contracts for the actions with the expects and promises list of keys like this:

@action(expects=["n"], promises=["result"])
def add_two(ctx: Context) -> Context:
    number = ctx.get("n", 0)

    ctx["result"] = number + 2

    return ctx


@action(expects=["result"])
def add_three(ctx: Context) -> Context:
    result = ctx["result"]

    ctx["result"] = result + 3

    return ctx

The action will verify - before it's invoked - that the expected keys are in the Context hash. If there are any missing, ExpectedKeyNotFoundError will be thrown and all of the missing keys will be listed in the exception message. Similarly, PromisedKeyNotFoundError is raised when the action fails to provide a value with the defined promised keys.

You can find the relevant examples here.

One of your actions might fail while they have logic that permanently changes state in a data store or in an API resource. A trivial example is charging your customer while you can't complete the order. When that happens, you can leverage pyservice's rollback functionality like this:

def add_two_rollback(ctx: Context) -> Context:
    ctx["result"] -= 2
    return ctx


@action(expects=["n"], promises=["result"], rollback=add_two_rollback)
def add_two(ctx: Context) -> Context:
    number = ctx.get("n", 0)

    ctx["result"] = number + 2

    return ctx


@action()
def fail_context(ctx: Context) -> Context:
    ctx.fail("I don't like what I see here")

The action accepts a function reference to roll back its state changes when a Context fails. The rollback field is optional, nothing happens when you don't provide one.

Take a look at this basic example.