While this is certainly useful, it's pretty basic. Something that I find is a much closer fit to UNIX pipes is iterators. UNIX pipes work similarly in that all of the commands in the pipeline are executed in "parallel" and the OS passes data incrementally between each process.

I primarily work on a Python codebase, and I've found that using iterators for complex, fault-tolerant data pipelines allows decoupled design without many of the performance & additional complexity drawbacks often encountered with cleanly abstracted, decoupled code. For example, when executing a multi-get for objects by primary key, the pipeline looks roughly like this:

1. Fetch from heap cache

2. Fetch from remote cache

3. Fetch from backing store

4. Backfill the heap cache

5. Backfill the remote cache

6. Apply basic filters (e.g. deleted == False, etc)

At each step there are usually two or three layers of abstraction underneath. Much of the space requirements, and some of the overhead time at each step can be collapsed to O(1) instead of O(N).

For example, a cache multiget abstraction on-top of memcache might look something like this:

    def deserialize_user(serialized_user):
        return json.loads(serialized_user)

    def build_prefixed_memcache_key(prefix, key):
        return "%s:%s" % (prefix, key)

    def get_users_from_remote_cache(user_keys):
        cached_users = get_from_memcache(user_keys, "user")
        deserialized_users = {deserialize_user(value) for key, value in cached_users.iteritems()}
        return deserialized_users

    def get_from_memcache(keys, prefix):
        rekeyed = {build_prefixed_memcache_key(prefix, key): key for key in keys}
        from_memcache = []
        for chunk in chunks(rekeyed.keys(), 20):
            results = memcache_mget(chunk)
            from_memcache.extend(results)
        unkeyed = {rekeyed[key]: value for key, value in from_memcache}
        return unkeyed

Notice how at each step there is a large amount of "buffering" that causes allocation, copying, and quite a bit of additional work. Each layer of abstraction adds a pretty large cost to the step. Using an iterator implementation, we can clean up this code and make it more performant:

    def get_users_from_remote_cache(user_keys):
        for key, user in get_from_memcache(user_keys, "user"):
            yield deserialize_user(user)

    def get_from_memcache(keys, prefix):
        for chunk in ichunks(keys, 20):
            rekeyed = {build_prefixed_memcache_key(prefix, key): key for key in chunk}
            for key, value in memcache_mget(rekeyed.keys()):
                yield (rekeyed[key], value)

It's clear how much cleaner this code is. Notice how this snippet avoids the large amount of "buffering" of data between steps and short-circuits quite a bit of code when possible (for instance, if all of the fetches miss). In real code that's heavily abstracted & layered, avoiding all of this work translates into significant performance & cost advantages.

Iterators also allow building portions of the pipeline with built-in functions that avoid the interpeter.

    def deserialize_users(users):
        return imap(pickle.loads, users)

    def get_users_from_remote_cache(user_keys):
        cached_users = get_from_memcache(user_keys, "user")
        only_values = imap(itemgetter(1), cached_users)
        return deserialize_users(only_values)

This block of code executes in only one pass through the interpreter (imap, itemgetter, and pickle.loads are all implemented as native functions). This is incredibly powerful because it means that iterator-based abstractions can be built by combining these native building blocks without the overhead of recursion within the interpreter at each step.

Pushing all the recursion into native code:

    def gprefix(prefix, delimiter):
        prefix_with_delim = prefix + delimiter
        prefixfn = partial(add, prefix_with_delim)
        unprefixfn = itemgetter(slice(len(prefix_with_delim), None))
        return prefixfn, unprefixfn

    def memcache_mget_chunked(keys):
        chunks = ichunks(keys, 20)
        result_blocks = imap(memcache_mget, chunks)
        flattened_results = chain.from_iterable(result_blocks)
        return flattened_results

    def get_values_from_memcache(keys, prefix):
        prefixfn = partial(add, "%s:" % prefix)
        prefixed_keys = imap(prefixfn, keys)
        key_pairs_from_cache = memcache_mget_chunked(prefixed_keys)
        values_from_cache = imap(itemgetter(1), key_pairs_from_cache)
        return values_from_cache

* I apologize for any code errors, this code wasn't tested in it's entirety.