Chains a series of generator functions using unix command line pipe like syntax

gpipe.py

# Python is great for sysadmin.
# Generators are great for plumbing.
# Nesting generator function calls is messy.
# (i for i in GPipe(inital)|func1|func2|(func3, args..))

class GPipe:
	"""Connects a chain of generators
	
	"""

	def __init__(self, g):
		self.initial_payload = g
		self.chain = []

	def connect(self, func, *args):
		self.chain.append((func, args))
		return self

	def __or__(self, other):
		if callable(other):
			other = (other,)
		return self.connect(*other)

	def apply(self):
		g = self.initial_payload
		for (f, args) in self.chain:
				g = f(g, *args)
		return g
	
	def __iter__(self):
		return self.apply()

def plus(g, n):
	for i in g:
		yield i + n

def mul(g, n):
	for i in g:
		yield i * n

def odd(g):
	return (i for i in g if i % 2)

def first(g, n = 3):
	c = 0
	for i in g:
		if c >= n:
			return
		c += 1
		yield i

def count(g):
	c = 0
	for i in g:
		c += 1
	yield c
	
for i in GPipe(range(15))|(plus, 1)|odd|(first, 9):
	print i

for i in GPipe(range(15))|(plus, 1)|odd|count:
	print i

pipe.rb

require 'pp'

class Enumlet
  include Enumerable

  def initialize(enum, &blk)
    @src = enum
    @blk = blk
  end

  def each
    yielder = lambda {|e| yield e}
    @blk.call(@src, yielder)
  end
end

class Pipe
  def initialize(initial_payload)
    @en = initial_payload
  end

  def connect(&blk)
    @en = Enumlet.new(@en, &blk)
    self
  end

  def |(lambdaa)
    connect(&lambdaa)
  end

  def each(*args, &blk)
    @en.each(*args, &blk)
  end
end

number = lambda {|a, y| a.each_with_index {|v, i| y.call "#{i+1} #{v}"}}
upcase = lambda {|a, y| a.each {|v| y.call v.upcase}}
# remove adjacent duplicating lines
uniq   = lambda {|a, y| last = nil; a.each {|e| y.call last = e unless last == e}}

lines = %w{one more thing to to mention}.join("\n")

for i in Pipe.new(lines.each_line).
  connect(&upcase).
  connect(&uniq).
  connect(&number).
  # passing a block, prepend a # to each line 
  connect {|a, y| a.each {|e| y.call '#' + e}} do
  puts i
end

for i in Pipe.new(lines.each_line) | upcase | uniq | number |
  # passing a block, prepend a # to each line 
  lambda {|a, y| a.each {|e| y.call '#' + e}} do
  puts i
end