orinoco

Functional composable pipelines allowing clean separation of the business logic and its implementation.

Features

• powerful chaining capabilities
• separation of business logic from the implementation
• functional approach
• async support
• PEP 561 compliant (checked with strict mypy)
• complete set of logic operations
• built-in execution measurement via observer pattern
• typed data container with powerful lookup capabilities
• easy to extend by user defined actions

Consider pipeline like this:

ParseUserName() >> GetUserDataFromDb() >> GetEmailTemplate() >> SendEmail()

Even without knowing how the implementation looks like or even what this library does, it's quite obvious what will happen when the pipeline is executed. Main idea of this library is to operate with simple composable blocks ("actions") to build more complex pipelines.

orinoco provides many action bases to simply define new actions. An example implementation of the first action from the above example could look like this:

class ParseUserName(TypedAction):
    def __call__(self, payload: str) -> str
        return json.loads(payload)["user_name"]

The action above is using annotations to get the signature of the input and output data. However, we can be more explicit:

class ParseUserName(TypedAction):
    CONFIG = ActionConfig(
          INPUT=({"payload": Signature(key="payload")}), OUTPUT=Signature(key="user_name", type_=str)
        )
    def __call__(self, payload: str) -> str
        return json.loads(payload)["user_name"]

We don't have to limit ourselves to simple "straightforward" pipelines as the one above. The execution flow can be controlled or modified via several predefined actions. These actions allow to perform conditional execution, branching, loops, context managers, error handling etc.

Switch()
    .case(
        If(ClaimInValidState("PAID")),
        Then(
            ~ClaimHasPaymentAuthorizationAssigned(
                fail_message="You cannot reset a claim {claim} that has payment authorizations assigned!"
            ),
            StateToAuthorize(),
            ResetEligibleAmount(),
        ),
    )
    .case(
        If(
            ClaimInValidState("DECLINED", "CANCELLED"),
            ClaimHasPreAuth(),
            ~ClaimHasValidNotification(),
            ~ClaimIsRetroactive(),
        ),
        Then(UserCanResetClaim(), StateToPendingAuthorization()),
    )
>> GetClaimChangedNotificationMessage()
>> NotifyUser()

See the docs for more info.

Installation

Use pypi to install the package:

pip install orinoco

Motivation

Python is a very powerful programming language allowing developers to quickly transform their ideas into the code. As you can imagine, this could be a double-edged sword. On one hand, it renders Python easy to use, on the other hand, larger projects can get messy if the team is not well-disciplined. Moreover, if the problem domain is complex enough, even seasoned developers can struggle with producing maintainable and easily readable code.

orinoco aims to help developers to express complex business rules in a more readable, understandable and maintainable fashion. Usual approach of implementing routines as a sequence of commands (e.g. querying a database, communicating with an external API) is replaced with pipelines composed from individual actions.

Example scenario

Let's imagine an application authorizing payments, for instance the ones send by a payment terminal in a shop. The authorisation logic is, understandably, based on various business rules.

Suppose the card holder is also an insurance policy holder. Their card could be then used to cover their insurance claims. Ideally, we would like to authorise payments based not only on the details of the current transaction, but also based on their insurance policy. An example implementation could look like this:

class Api:
    def __init__(self, parser, fraud_service, db_service, policies_matcher):
        self.parser = parser
        self.fraud_service = fraud_service
        self.db_service = db_service
        self.policies_matcher = policies_matcher

    def payment_auth_endpoint(self.request):
        payment_data = self.parser.parser(request)
        
        self.db_service.store_payment(payment_data)
        
        if self.fraud_service.is_it_fraud(payment_data):
            return Response(json={"authorization_decision": False, "reason": "Fraud payment"})
        
        policy = self.policies_matcher.get_policy_for_payment(card_data)
        
        if not policy:
            return Response(json={"authorization_decision": False, "reason": "Not matching policy"})
            
        funding_account = self.db_service.get_funding_account(policy["funding_account_id"])
        
        if funding_account.amount < payment_data["amount"]:
            return Response(json={"authorization_decision": False, "reason": "Not enough money"})
            
            
        self.db_service.update_policy(policy, payment_data)
        self.db_service.update_funding_account(funding_account, payment_data)
        self.db_service.link_policy_to_payment(policy, payment_data)
        
        return Response(json={"authorization_decision": True})

In this example we abstracted all we could into services and simple methods, we leveraged design patterns (such as explicit dependency injection), but it still feels there is a lot going on in this method. In order to understand the ins and outs of the method, it's rather necessary to go through the code line by line.

Let's look at an alternative version implemented using orinoco:

class Api:
    AUTH_ACTION = (
        ParseData()
        >> StorePayment()
        >> IsFraud().if_then(GetDeniedFraudResponse().finish())
        >> GetUserPolicy()
        >> GetFundingAccount()
        >> (~EnoughMoney()).if_then(GetNoMoneyResponse().finish())
        >> UpdatePolicy()
        >> UpdateFundingAccount()
        >> UpdatePolicy()
        >> GetAuthorizedResponse()
    )
    def payment_auth_endpoint(self, request):
        return self.AUTH_ACTION.run_with_data(request=request).get_by_type(Response)

We moved from the actual implementation of the process as a series of commands into an actual description of the business process. This makes it readable even for people without any programming knowledge. We can go even further and separate the pipeline into another file which will serve as a source of truth for our business processes.

Building blocks

Actions

Actions are main building blocks carrying the business logic and can be chained together. There are many predefined actions that can be used to build more complex pipelines such as actions for boolean logic and loops.

Actions can be created directly by inheriting from specialized bases (see subsections below). If there is no suitable base for your use case, you can inherit from orinoco.action.Action too, but it's generally discouraged. In the latter can you would proceed by overriding run(action_data: ActionData) -> ActionData method.

Pipelines can be then executed by providing ActionData container directly to run method or by run_with_data(**kwargs: Any) method which will basically create the ActionData and pass it to the run method. Find more info about ActionData below.

TypedAction

This is an enhanced Action that uses ActionConfig as the way to configure the input and the output.

Business logic is defined in the call method with "normal" parameters as the input, this means that no raw ActionData is required. The propagation of the data from the ActionData to the method is done automatically based on the ActionConfig which can be either passed directly to the initializer (see config) or as a class variable (see CONFIG) or it could be implicitly derived from annotations of the __call__ method.

The result of the method is propagated to the ActionData with a matching signature. Note that implicit config will use only annotated return type for the signature, unless it's annotated by typing.Annotated, where the arguments are: type, key, *tags. For more control, please define the ActionConfig manually.

The implicit approach:

class SumValuesAndRound(TypedAction):
    def __call__(self, x: float, y: float) -> str:
        return int(x + y)

class SumValuesAndRoundAnnotated(TypedAction):
    def __call__(self, x: float, y: float) -> Annotated[str, "my_sum", "optional_tag1", "optional_tag2"]:
        return int(x + y)
        
assert 3 == SumValuesAndRound().run_with_data(x=1.2, y=1.8).get_by_type(int)
assert 3 == SumValuesAndRoundAnnotated().run_with_data(x=1.2, y=1.8).get("my_sum")
assert 3 == SumValuesAndRoundAnnotated().run_with_data(x=1.2, y=1.8).get_by_tag("optional_tag1")
assert 3 == SumValuesAndRoundAnnotated().run_with_data(x=1.2, y=1.8).get_by_tag("optional_tag1", "optional_tag1")

Explicit approach:

class SumValuesAndRound(TypedAction):
    CONFIG = ActionConfig(
          INPUT=({"x": Signature(key="x"), "y": Signature(key="y")}), OUTPUT=Signature(key="sum_result", type_=int)
        )
    def __call__(self, x: float, y: float) -> int:
        return int(x + y)
result: ActionData = SumValuesAndRound().run_with_data(x=1.2, y=1.8)
assert 3 == result.get_by_type(int) == result.get("sum_result")

Notice there are more possibilities how to retrieve the values from the ActionData since it's explicitly annotated. See the next section for more information about ActionData container. In this "mode" the type annotations are optional.

assert 5 == SumValuesAndRound()(x=1.9, y=3.1)

Default values

class SumValuesAndRound(TypedAction):
    def __call__(self, x: float, y: float = 1.0) -> str:
        return int(x + y)

assert 2.2 == SumValuesAndRound().run_with_data(x=1.2)

Retry

TypedAction and `TypedCondition