Pipeable

A DSL for workflows built atop native Function Composition which leverages the Railway Pattern. This allows you to write a sequence of steps that cleanly read from top-to-bottom or left-to-right resulting in a single success or a failure. This allows you to avoid relying on exceptions for expensive control flows and/or complex conditional logic in general.

Table of Contents

Features
Requirements
Setup
Usage
- Pipe
- Steps
  - Basic
    - alt
    - amap
    - as
    - bind
    - check
    - fmap
    - insert
    - map
    - merge
    - tee
    - to
    - try
    - use
    - validate
  - Advanced
    - Procs
    - Lambdas
    - Methods
    - Custom
- Superpipes
- Containers
- Composition
Development
- Architecture
- Style Guide
- Debugging
- Troubleshooting
  - Type Errors
  - No Method Errors
Tests
Benchmarks
License
Security
Code of Conduct
Contributions
Developer Certificate of Origin
Versions
Community
Credits

Features

Built atop native Function Composition.
Adheres to the Railway Pattern.
Provides built-in and customizable domain-specific steps.
Provides chainable pipes which can be used to build more complex workflows.
Compatible with the Containable, Infusible, and Dry Monads gems.

Requirements

Ruby.
A strong understanding of Function Composition and Method Parameters And Arguments.

Setup

To install with security, run:

# 💡 Skip this line if you already have the public certificate installed.
gem cert --add <(curl --compressed --location https://alchemists.io/gems.pem)
gem install pipeable --trust-policy HighSecurity

To install without security, run:

gem install pipeable

You can also add the gem directly to your project:

bundle add pipeable

Once the gem is installed, you only need to require it:

require "pipeable"

Usage

You can turn any object into a pipe by requiring and including this gem as follows:

require "csv"
require "pipeable"

class Demo
  include Pipeable

  def initialize client: CSV
    @client = client
  end

  def call data
    pipe data,
         check(/Book.+Price/, :match?),
         :parse,
         map { |item| "#{item[:book]}: #{item[:price]}" }
  end

  private

  attr_reader :client

  def parse result
    result.fmap do |data|
      client.instance(data, headers: true, header_converters: proc { |key| key.downcase.to_sym })
            .to_a
            .map(&:to_h)
    end
  end
end

The above allows Demo#call to be a sequence of steps which may pass or fail due to all steps using Dry Monads for input and output. This is the essence of the Railway Pattern.

To execute the above example, you’d only need to pass CSV content to it:

Demo.new.call <<~CSV
  Book,Author,Price,At
  Mystics,urGoh,10.50,2022-01-01
  Skeksis,skekSil,20.75,2022-02-13
CSV

The computed result is a success with each book listing a price:

Success ["Mystics: 10.50", "Skeksis: 20.75"]

Pipe

Once you’ve included the Pipeable module within your class, the #pipe method is available to you and is how you build a sequence of steps for processing. The method signature is:

pipe(input, *steps)

The first argument is your input which can be a Ruby primitive or a monad. Regardless, the input will be automatically wrapped as a Success — but only if not a Result to begin with — before passing to the first step. From there, all steps are required to answer a monad in order to adhere to the Railway Pattern.

Behind the scenes, the #pipe method is syntactic sugar built atop Function Composition which means if this code were to be rewritten:

pipe csv,
     check(/Book.+Price/, :match?),
     :parse,
     map { |item| "#{item[:book]}: #{item[:price]}" }

…then the above would look like the following (as rewritten in native Ruby):

(
  check(/Book.+Price/, :match?) >>
  method(:parse) >>
  map { |item| "#{item[:book]}: #{item[:price]}" }
).call Success(csv)

Visually, the pipe can be diagramed as follows:

The problem with native function composition is that it reads backwards by passing input at the end of all sequential steps. With the #pipe method, you have the benefit of allowing your eyes to read from top to bottom while not having to type multiple forward composition operators.

Steps

There are several ways to compose steps for your pipe. As long as all steps succeed, you’ll get a successful response. Otherwise, the first step to fail will pass the failure down by skipping all subsequent steps (unless you dynamically turn the failure into a success). Each step can be initialized and called:

#initialize: Arguments vary per step but can be positional, keyword, and/or block arguments. This is how you customize the behavior of each step.
#call: Expects a Dry Monads Result object as input. The output is either the same or new Result object for consumption by the next step in the pipe. Additionally, each step will either unwrap the Result or pass the Result through depending on the step’s implementation (as detailed below).

Basic

The following are the basic (default) steps for building custom pipes for which you can mix and match within your own implementation.

alt

Allows you to operate on a failure and produce either a success or another failure. This is a convenience wrapper to native Dry Monads #or functionality.

Accepts a failure while answering either a success or failure. Example:

pipe %i[a b c], alt { |object| Success object.join("-") }          # Success [:a, :b, :c]
pipe Failure("Danger!"), alt { Success "Resolved" }                # Success "Resolved"
pipe Failure("Danger!"), alt { |object| Failure "Big #{object}" }  # Failure "Big Danger!"

amap

Allows you to unwrap a failure, make a modification, and wrap the modification as a new failure. This is a convenience wrapper to native Dry Monads #alt_map functionality.

Accepts and answers a failure. Example:

pipe Failure("Danger"), amap { |object| "#{object}!" }  # Failure "Danger!"
pipe Success("Pass"), amap { |object| "#{object}!" }    # Success "Pass"

as

Allows you to message an object as a different result. The first argument is the method but additional positional and/or keyword arguments can be passed along if the method accepts them.

Accepts and answers a success. Example:

pipe :a, as(:inspect)                  # Success ":a"
pipe %i[a b c], as(:dig, 1)            # Success :b
pipe Failure("Danger!"), as(:inspect)  # Failure "Danger!"

bind

Allows you to perform operations upon success only. You are then responsible for answering a success or failure accordingly. This is a convenience wrapper to native Dry Monads #bind functionality.

Accepts a success while answering either a success or failure. Example:

pipe %i[a b c], bind { |object| Success object.join("-") }           # Success "a-b-c"
pipe %i[a b c], bind { |object| Failure object }                     # Failure [:a, :b, :c]
pipe Failure("Danger!"), bind { |object| Success object.join("-") }  # Failure "Danger!"

check

Allows you to check if an object matches the proof (with message). The first argument is your proof while the second argument is the message to send to your proof. A check only passes if the messaged object evaluates to true or Success. When successful, the object is passed through as a Success. When false, the object is passed through as a Failure.

Accepts a success while answering a success or failure depending on whether unwrapped object checks against the proof. Example:

pipe :a, check(%i[a b], :include?)                  # Success :a
pipe :a, check(%i[b c], :include?)                  # Failure :a
pipe Failure("Danger!"), check(%i[a b], :include?)  # Failure "Danger!"

fmap

Allows you to unwrap a success, make a modification, and wrap the modification as a new success. This is a convenience wrapper to native Dry Monads #fmap functionality.

Accepts and answers a success. Example:

pipe %i[a b c], fmap { |object| object.join "-" }           # Success "a-b-c"
pipe Failure("Danger!"), fmap { |object| object.join "-" }  # Failure "Danger!"

insert

Allows you to insert an element after an object (default behavior) as a single array. This step wraps native Array#insert functionality. If the object is not an array, it will be cast as one. You can use the :at key to specify where you want insertion to happen. This step is most useful when needing to assemble positional arguments for passing as an array to a subsequent step.

Accepts and answers a success. Example:

pipe :a, insert(:b)                  # Success [:a, :b]
pipe :a, insert(:b, at: 0)           # Success [:b, :a]
pipe %i[a c], insert(:b, at: 1)      # Success [:a, :b, :c]
pipe Failure("Danger!"), insert(:b)  # Failure "Danger!"

map

Allows you to map over an object (enumerable) by wrapping native Enumerable#map functionality.

Accepts and answers a success. Example:

pipe %i[a b c], map(&:inspect)           # Success [":a", ":b", ":c"]
pipe Failure("Danger!"), map(&:inspect)  # Failure "Danger!"

merge

Allows you to merge an object with additional attributes as a single hash. This step wraps native Hash#merge functionality. If the input is not a hash, then the object will be merged with step as the key. The default step key can be renamed to a different key by using the :as key. Like the insert step, this step is most useful when assembling keyword arguments and/or a hash for a subsequent steps.

Accepts and answers a success. Example:

pipe({a: 1}, merge(b: 2))             # Success {a: 1, b: 2}
pipe "test", merge(b: 2)              # Success {step: "test", b: 2}
pipe "test", merge(as: :a, b: 2)      # Success {a: "test", b: 2}
pipe Failure("Danger!"), merge(b: 2)  # Failure "Danger!"

tee

Allows you to run an operation and ignore the response while input is passed through as output. This behavior is similar in nature to the tee program in Bash.

Accepts either a success or failure and passes the result through while allowing you to execute arbitrary behavior. Example:

pipe "test", tee(Kernel, :puts, "Example.")

# Example.
# Success "test"

pipe Failure("Danger!"), tee(Kernel, :puts, "Example.")

# Example.
# Failure "Danger!"

to

Allows you to delegate to an object which doesn’t have a callable interface and may or may not answer a result. If the response is not a monad, it’ll be automatically wrapped as a Success.

Accepts a success while sending the unwrapped object to the given object’s corresponding method. The object is expected to answer either a plain Ruby object which will be automatically wrapped as a success or a Dry Monads Result. Example:

Model = Struct.new :label do
  include Dry::Monads[:result]

  def self.for(**) = Success new(**)
end

pipe({label: "Test"}, to(Model, :for))    # Success #<struct Model label="Test">
pipe Failure("Danger!"), to(Model, :for)  # Failure "Danger!"

try

Allows you to try an operation which may fail while catching any exceptions as a failure for further processing. You can catch a single exception by providing the exception as a single value or multiple exceptions as an array of values.

Accepts and answers a success if there are no exceptions. Otherwise, captures any error as a failure. Example:

pipe "test", try(:to_json, catch: JSON::ParserError)
# Success "\"test\""

pipe "test", try(:to_json, catch: [JSON::ParserError, StandardError])
# Success "\"test\""

pipe "test", try(:invalid, catch: NoMethodError)
# Failure(#<NoMethodError: undefined method `invalid' for an instance of String>)

pipe Failure("Danger!"), try(:to_json, catch: JSON::ParserError)
# Failure "Danger!"

use

Allows you to use another pipe to build a superpipe, use an object that adheres to the Command Pattern, or any function which answers a Dry Monads Result object. In other words, you can use use any object which responds to #call that answers a Dry Monads Result object. This is great for chaining multiple pipes together (i.e. superpipes).

Accepts a success while sending the unwrapped object to the command (or pipe) for further processing. A Dry Monads Result is expected to be answered by the command. Example:

function = -> number { Success number * 3 }

pipe 3, use(function)                   # Success 9
pipe Failure("Danger!"), use(function)  # Failure "Danger!"

validate

Allows you to use an contract for validating an object. This is especially useful when using Dry Schema, Dry Validation, or any contract that responds to #call and answers a Result.

By default, the :as key’s value is nil. Use :to_h, for example, as the value for automatic casting to a Hash. You can also pass in any value to the :as key which is a valid method that the contract’s result will respond to.

Accepts a success and rewraps as a success if the :as keyword is supplied. Otherwise, any failure is immediately passed through. Example:

schema = Dry::Schema.Params { required(:label).filled :string }

pipe({label: "Test"}, validate(schema))
# Success label: "Test"

pipe({label: "Test"}, validate(schema, as: nil))
# Success #<Dry::Schema::Result{:label=>"Test"} errors={} path=[]>

pipe Failure("Danger!"), validate(schema)
# Failure "Danger!"

💡 Ensure you enable the Dry Monads extension for Dry Schema and/or Dry Validation when using this step since this step expects the contract to respond to the #to_monad message.

Advanced

Several options are available should you need to advance beyond the basic steps. Each is described in detail below.

Procs

You can always use a Proc as a custom step. Example:

include Dry::Monads[:result]
include Pipeable

pipe :a,
     insert(:b),
     proc { Success "input_ignored" },
     as(:to_sym)

# Yields: Success :input_ignored

Lambdas

In addition to procs, lambdas can be used too. Example:

include Pipeable

pipe :a,
     insert(:b),
     -> result { result.fmap { |items| items.join "_" } },
     as(:to_sym)

# Yields: Success :a_b

Methods

Methods, in addition to procs and lambdas, are the preferred way to add custom steps due to the concise syntax. Example:

class Demo
  include Pipeable

  def call(input) = pipe input, insert(:b), :join, as(:to_sym)

  private

  def join(result) = result.fmap { |items| items.join "_" }
end

Demo.new.call :a  # Success :a_b

All methods can be referenced by symbol as shown via :join above. Using a symbol is syntactic sugar for Object#method so :join (symbol) is the same as using method(:join). Both work but the former requires less typing.

Custom

If you’d like to define permanent and reusable steps, you can register a custom step which requires you to:

Define a custom step as a class, lambda, or proc.
Register your custom step along side the existing default steps.

Here’s what this would look like:

module CustomSteps
  class Join < Pipeable::Steps::Abstract
    def initialize(delimiter = "_", **)
      super(**)
      @delimiter = delimiter
    end

    def call(result) = result.fmap { |items| items.join delimiter }

    private

    attr_reader :delimiter
  end
end

Pipeable::Steps::Container.register :join, CustomSteps::Join

include Pipeable

pipe :a, insert(:b), join, as(:to_sym)
# Success :a_b

pipe :a, insert(:b), join(""), as(:to_sym)
# Success :ab

A lambda or proc can be used too (albeit in limited capacity). Here’s a version of the above using a lambda:

module CustomSteps
  Join = -> result { result.fmap { |items| items.join "_" } }
end

Pipeable::Steps::Container.register :join, CustomSteps::Join

include Pipeable

puts pipe(:a, insert(:b), join, as(:to_sym))
# Success :a_b

Superpipes

Superpipes, as first hinted at in the use step above, are a combination of pipeable objects chained together as individual steps. This allows you to reuse existing pipeable objects in new and interesting ways. Here’s an contrived, but simple, example of what a superpipe looks like when built from pipeable objects:

class One
  include Pipeable

  def initialize label = "one"
    @label = label
  end

  def call(item) = pipe item, insert(label, at: 0)

  private

  attr_reader :label
end

class Two
  include Pipeable

  def initialize label = "two"
    @label = label
  end

  def call(item) = pipe item, insert(label)

  private

  attr_reader :label
end

class Three
  include Pipeable

  def initialize one: One.new, two: Two.new
    @one = one
    @two = two
  end

  def call(item) = pipe item, use(one), use(two)

  private

  attr_reader :one, :two
end

Notice, One and Two are normal pipeable objects with individual steps while Three injects both One and Two as dependencies and then subsequently pipes them together in the #call method via the use step. This is the power of a superpipe. …and, yes, a superpipe can be an individual step in some other object. Turtles all the way down (or up). 😉

Again, the above is contrived but hopefully illustrates how you can build more complex architectures from smaller pipes.

Containers

Should you not want the basic steps, need custom steps, or a hybrid of default and custom steps, you can define your own container — using the Containable gem — and provide the container as an argument to .[] when including pipeable behavior. Example:

require "containable"

module CustomContainer
  extend Containable

  register :echo, -> result { result }
  register :insert, Pipeable::Steps::Insert
end

include Pipeable[CustomContainer]

pipe :a, echo, insert(:b)

# Yields: Success [:a, :b]

The above is a hybrid example where the CustomContainer registers a custom echo step along with the default insert step to make a new container. This is included when passed in as an argument via .[] (i.e. include Pipeable[CustomContainer]).

Whether you use default, custom, or hybrid steps, you have maximum flexibility when using containers.

Composition

Should you ever need to make a plain old Ruby object functionally composable, then you can include the Pipeable::Composable module which will give you the necessary #>>, #<<, and #call methods where you only need to implement the #call method.

Development

To contribute, run:

git clone https://github.com/bkuhlmann/pipeable
cd pipeable
bin/setup

You can also use the IRB console for direct access to all objects:

bin/console

Architecture

The architecture of this gem is built on top of the following concepts and gems:

Function Composition: Made possible through the use of the #>> and #<< methods on the Method and Proc objects.
Containable: Allows related dependencies to be grouped together for injection as desired.
Dry Monads: Critical to ensuring the entire pipeline of steps adhere to the Railway Pattern and leans heavily on the Result object.
Marameters: Through the use of the .categorize method, dynamic message passing is possible by inspecting the object’s method parameters.

Style Guide

Pipes
- Use a single method (i.e. #call) which is public and adheres to the Command Pattern so multiple pipes can be piped together (i.e. superpipes) if desired.
Steps
- Inherit from the Abstract class to gain monad, composition, and dependency behavior. This allows subclasses to have direct access to the base positional, keyword, and block arguments. These variables are prefixed with base_* in order to not conflict with subclasses which might only want to use non-prefixed variables for convenience.
- All filtered arguments — in other words, unused arguments — need to be passed up to the superclass from the subclass (i.e. super(*positionals, **keywords, &block)). Doing so allows the superclass (i.e. Abstract) to provide access to base_positionals, base_keywords, and base_block for use if desired by the subclass.
- The #call method must define a single positional result parameter since a monad will be passed as an argument. Example: def call(result) = # Implementation.
- Each block within the #call method should use the object parameter to be consistent. More specific parameters like operation or contract should be used to improve readability when context allows. Example: def call(result) = result.bind { |object| # Implementation }.
- Use implicit blocks sparingly. Most of the default steps shy away from using blocks because the code becomes more complex. Use private methods, custom steps, and/or separate pipes if the code becomes too complex because you might have a smaller object which needs extraction.

Debugging

If you need to debug (i.e. Debug) your pipe, use a lambda. Example:

pipe data,
     check(/Book.+Price/, :match?),
     -> result { binding.break; result }, # Breakpoint
     :parse

The above breakpoint will allow you inspect the result of the #check step and/or build a modified result for passing to the subsequent :parse method step.

Troubleshooting

The following might be of aid to as you implement your own pipes.

Type Errors

If you get a TypeError: Step must be functionally composable and answer a monad, it means:

The step must be a Proc, Method, or any object which responds to #>>, #<<, and #call.
The step doesn’t answer a result monad (i.e. Success object or Failure object).

No Method Errors

If you get a NoMethodError: undefined method success? exception, this might mean that you forgot to add a comma after one of your steps. Example:

# Valid
pipe "https://www.wikipedia.org",
     to(client, :get),
     try(:parse, catch: HTTP::Error)

# Invalid
pipe "https://www.wikipedia.org",
     to(client, :get) # Missing comma.
     try(:parse, catch: HTTP::Error)

Tests

To test, run:

bin/rake

Benchmarks

To view/compare performance, run:

bin/benchmark

💡 You can view current benchmarks at the end of the above file if you don’t want to manually run them.

Credits

Built with Gemsmith.
Engineered by Brooke Kuhlmann.

Runtime

Pipeable

Features

Requirements

Setup

Usage

Pipe

Steps

Basic

alt

amap

as

bind

check

fmap

insert

map

merge

tee

to

try

use

validate

Advanced

Procs

Lambdas

Methods

Custom

Superpipes

Containers

Composition

Development

Architecture

Style Guide

Debugging

Troubleshooting

Type Errors

No Method Errors

Tests

Benchmarks

Credits