pipeable

A domain specific language for building functionally composable steps.

https://github.com/bkuhlmann/pipeable

Keywords

Repository

A domain specific language for building functionally composable steps.

README.adoc

:toc: macro
:toclevels: 5
:figure-caption!:

:command_pattern_link: link:https://alchemists.io/articles/command_pattern[Command Pattern]
:containable_link: link:https://alchemists.io/projects/containable[Containable]
:debug_link: link:https://github.com/ruby/debug[Debug]
:dry_monads_link: link:https://dry-rb.org/gems/dry-monads[Dry Monads]
:dry_schema_link: link:https://dry-rb.org/gems/dry-schema[Dry Schema]
:dry_validation_link: link:https://dry-rb.org/gems/dry-validation[Dry Validation]
:function_composition_link: link:https://alchemists.io/articles/ruby_function_composition[Function Composition]
:infusible_link: link:https://alchemists.io/projects/infusible[Infusible]
:method_parameters_and_arguments_link: link:https://alchemists.io/articles/ruby_method_parameters_and_arguments[Method Parameters And Arguments]
:railway_pattern_link: link:https://fsharpforfunandprofit.com/posts/recipe-part2[Railway Pattern]

= Pipeable

A DSL for workflows built atop native {function_composition_link} which leverages the {railway_pattern_link}. 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.

toc::[]

== Features

* Built atop native {function_composition_link}.
* Adheres to the {railway_pattern_link}.
* Provides built-in and customizable domain-specific steps.
* Provides chainable _pipes_ which can be used to build more complex workflows.
* Compatible with the {containable_link}, {infusible_link}, and {dry_monads_link} gems.

== Requirements

. link:https://www.ruby-lang.org[Ruby].
. A strong understanding of {function_composition_link} and {method_parameters_and_arguments_link}.

== Setup

To install _with_ security, run:

[source,bash]
----
# 💡 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:

[source,bash]
----
gem install pipeable
----

You can also add the gem directly to your project:

[source,bash]
----
bundle add pipeable
----

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

[source,ruby]
----
require "pipeable"
----

== Usage

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

[source,ruby]
----
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_link} for input and output. This is the essence of the {railway_pattern_link}.

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

[source,ruby]
----
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:

[source,ruby]
----
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_link}.

Behind the scenes, the `#pipe` method is syntactic sugar built atop {function_composition_link} which means if this code were to be rewritten:

[source,ruby]
----
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):

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

Visually, the pipe can be diagramed as follows:

image::https://alchemists.io/images/projects/pipeable/diagrams/pipe.png[A diagram of pipe steps,width=591,height=734,role=focal_point]

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_link} `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

Short for _alternate_ which is the `or` branch of conditional logic. This allows you to operate on a failure and produce either a success or another failure. This is a convenience wrapper to native {dry_monads_link} `#or` functionality.

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

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

===== amap

Short for _alternate map_ which 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_link} `#alt_map` functionality.

Accepts and answers a failure. Example:

[source,ruby]
----
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 always the method to message but additional positional and/or keyword arguments can be passed along if the method accepts them.

Accepts and answers a success. Example:

[source,ruby]
----
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 responsible for answering a success or failure accordingly. This is a convenience wrapper to native {dry_monads_link} `#bind` functionality.

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

[source,ruby]
----
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:

[source,ruby]
----
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

Short for _function map_ which 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_link} `#fmap` functionality.

Accepts and answers a success. Example:

[source,ruby]
----
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 one or more elements after an object (default behavior) as a single array. This step wraps native link:https://rubyapi.org/o/array#method-i-insert[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 to a subsequent step.

⚠️ If given an array from the previous step, this step will mutate it.

Accepts and answers a success. Example:

[source,ruby]
----
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 :a, insert(:b, :c)              # Success [:a, :b, :c]
pipe :a, insert([:b])                # Success [:a, [:b]]
pipe :a, insert({b: 2})              # Success [:a, {b: 2}]
pipe Failure("Danger!"), insert(:b)  # Failure "Danger!"
----

===== map

Allows you to map over an object (enumerable) by wrapping native link:https://rubyapi.org/o/enumerable#method-i-map[Enumerable#map] functionality.

Accepts and answers a success. Example:

[source,ruby]
----
pipe %i[a b c], map(&:inspect)           # Success [":a", ":b", ":c"]
pipe %i[a b c], map { "#{it}1" }         # Success ["a1", "b1", "c1"]
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 link:https://rubyapi.org/o/hash#method-i-merge[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.

⚠️ If given a hash from the previous step, this step will mutate it.

Accepts and answers a success. Example:

[source,ruby]
----
pipe({a: 1}, merge(b: 2))             # Success {a: 1, b: 2}
pipe({a: 1}, merge(b: 2, c: 3))       # Success {a: 1, b: 2, c: 3}
pipe "demo", merge(b: 2)              # Success {step: "demo", b: 2}
pipe "demo", merge(as: :a, b: 2)      # Success {a: "demo", 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 link:https://www.gnu.org/savannah-checkouts/gnu/gawk/manual/html_node/Tee-Program.html[tee] Bash program.

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

[source,ruby]
----
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_link} `Result`. Example:

[source,ruby]
----
Model = Struct.new :label do
  include Dry::Monads[:result]

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

pipe({label: "Test"}, to(Model, :for))    # Success #
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:

[source,ruby]
----
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(#)

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_link}, or any function which answers a {dry_monads_link} `Result` object. In other words, you can use _use_ any object which responds to `#call` that answers a {dry_monads_link} `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_link} `Result` is expected to be answered by the command. Example:

[source,ruby]
----
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_link}, {dry_validation_link}, 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 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:

[source,ruby]
----
schema = Dry::Schema.Params { required(:label).filled :string }

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

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

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

💡 Ensure you enable the {dry_monads_link} extension for {dry_schema_link} and/or {dry_validation_link} 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:

[source,ruby]
----
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:

[source,ruby]
----
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:

[source,ruby]
----
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 link:https://rubyapi.org/o/object#method-i-method[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:

[source,ruby]
----
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:

[source,ruby]
----
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:

[source,ruby]
----
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_link} gem -- and provide the container as an argument to `.[]` when including pipeable behavior. Example:

[source,ruby]
----
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:

[source,bash]
----
git clone https://github.com/bkuhlmann/pipeable
cd pipeable
bin/setup
----

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

[source,bash]
----
bin/console
----

=== Architecture

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

* {function_composition_link}: Made possible through the use of the `\#>>` and `#<<` methods on the link:https://rubyapi.org/3.1/o/method[Method] and link:https://rubyapi.org/3.1/o/proc[Proc] objects.
* {containable_link}: Allows related dependencies to be grouped together for injection as desired.
* {dry_monads_link}: Critical to ensuring the entire pipeline of steps adhere to the {railway_pattern_link} and leans heavily on the `Result` object.
* link:https://alchemists.io/projects/marameters[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_link} 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_link}) your pipe, use a lambda. Example:

[source,ruby]
----
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` or `Failure`).

==== 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:

[source,ruby]
----
# 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:

[source,bash]
----
bin/rake
----

== Benchmarks

To view/compare performance, run:

[source,bash]
----
bin/benchmark
----

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

== link:https://alchemists.io/policies/license[License]

== link:https://alchemists.io/policies/security[Security]

== link:https://alchemists.io/policies/code_of_conduct[Code of Conduct]

== link:https://alchemists.io/policies/contributions[Contributions]

== link:https://alchemists.io/policies/developer_certificate_of_origin[Developer Certificate of Origin]

== link:https://alchemists.io/projects/pipeable/versions[Versions]

== link:https://alchemists.io/community[Community]

== Credits

* Built with link:https://alchemists.io/projects/gemsmith[Gemsmith].
* Engineered by link:https://alchemists.io/team/brooke_kuhlmann[Brooke Kuhlmann].

Owner

Citation (CITATION.cff)

cff-version: 1.2.0
message: Please use the following metadata when citing this project in your work.
title: Pipeable
abstract: A domain specific language for building functionally composable steps.
version: 1.4.0
license: Hippocratic-2.1
date-released: 2025-07-15
authors:
  - family-names: Kuhlmann
    given-names: Brooke
    affiliation: Alchemists
    orcid: https://orcid.org/0000-0002-5810-6268
keywords:
 - function composition
 - railway pattern
 - ruby
repository-code: https://github.com/bkuhlmann/pipeable
repository-artifact: https://rubygems.org/gems/pipeable
url: https://alchemists.io/projects/pipeable

GitHub Events

Issues and Pull Requests

Last synced: 6 months ago

Packages