README
¶
Chainmorph - Go meets ETL through method chaining
Go library that allows you to create a step-wise ETL pipeline using Go 1.27+
This is all possible thanks to the ability to create generic methods that define their own generic constraints!
The name 'chain' comes from the fact that I've built the basics of this library because I focused very hard on making the public API to use method-chaining as much as possible because it looks cool, and I like doing cool-stuff. The second part, 'morph', is because every method morphs the previous step in some way via closures.
Why this library even exists
Recently I tried to finish a project that I had already resigned on doing called RivenBot which is essentially a Charlamagne wanna-be Discord bot that gives Destiny 2 players insights on their raid statistics. The initial work for this project involved in processing several GBs of initial data, courtesy from D2Asun, the goat, into my own historical database. This process is technically ETL processing, where I am reading from compressed datasources of historical Destiny 2 data, creating my own internal mappings and saving them in my database. I have been working on this processing for the past two months and I've come at a cross roads that whatever I'm doing looks like shit and is hard to test because of how much gibberish and scattered logic I've written.
Hence Chainmorph's existence. When I first saw ThePrimagen's video on why he hates Go now it intrigued me exactly what kind of change was Go 1.27 making that he hated it so much. Turns out that this change, lettings methods define their own generics constraints, was a banger of a change and made this whole library possible! Therefore, here it is.
Documentation
¶
Overview ¶
Package chainmorph is a small ETL pipeline library built around Go 1.27's generic methods. It lets you compose readers, filters, processors, and writers into a single fluent, lazily-evaluated chain, including stages that change the item's type as it flows through.
Overview ¶
A Pipeline is a lazily-pulled stream. Nothing runs until a terminal method — Pipeline.WriteTo — is called. Every stage before that just builds up a chain of deferred work.
err := From(reader). Filter(isEven). MapTo(double). WriteTo(ctx, writer)
Type-changing stages ¶
Pipeline.MapTo and Pipeline.MapFunc can change the pipeline's element type from T to R. This is only possible because Go 1.27 allows methods to declare their own type parameters, independent of the receiver's.
Index ¶
- Variables
- type ItemFilter
- type ItemMapper
- type ItemReader
- type ItemWriter
- type MapperFunc
- type Pipeline
- func (p *Pipeline[T]) Filter(itemFilter ItemFilter[T]) *Pipeline[T]
- func (p *Pipeline[T]) FilterFunc(f func(context.Context, T) (bool, error)) *Pipeline[T]
- func (p *Pipeline[T]) If(f func(item T) bool) *Pipeline[T]
- func (p *Pipeline[T]) MapFunc[R any](f func(context.Context, T) (R, error)) *Pipeline[R]
- func (p *Pipeline[T]) MapTo[R any](itemMapper ItemMapper[T, R]) *Pipeline[R]
- func (p *Pipeline[T]) Tap(tapper Tapper[T]) *Pipeline[T]
- func (p *Pipeline[T]) TapFunc(f func(context.Context, T) error) *Pipeline[T]
- func (p *Pipeline[T]) WriteTo(ctx context.Context, itemWriter ItemWriter[T]) error
- type Predicate
- type TapFunc
- type Tapper
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrEndOfStream = errors.New("end of stream")
ErrEndOfStream signals that an ItemReader has no more items to produce. Implementations of ItemReader.ReadFrom must return this error wrapped or unwrapped. Once exhausted, Pipeline has no other way to detect the end of a stream. Returning any other error is treated as a failure and stops the pipeline, returning nil forever will cause the pipeline to pull indefinitely.
Functions ¶
This section is empty.
Types ¶
type ItemFilter ¶
ItemFilter decides whether an item should continue through the pipeline. See Pipeline.Filter.
type ItemMapper ¶
ItemMapper transforms an item of type T into an item of type R. See Pipeline.MapTo, which is the methods that lets a pipeline's element type change mid-chain.
type ItemReader ¶
ItemReader produces a stream of items of type T, one per call to ReadFrom. Implementations must return ErrEndOfStream once exhausted.
type ItemWriter ¶
ItemWriter is a terminal sink that consumes a single item of type T, such as writing it to a database, file, or external service.
type MapperFunc ¶
MapperFunc adapts a plain function in an ItemMapper. It's the mechanism behind Pipeline.MapFunc
type Pipeline ¶
type Pipeline[T any] struct { // contains filtered or unexported fields }
Pipeline represents a chain of lazily-evaluated steps applied to a stream of items of type T. No item is read or transformed until a terminal operation such as Pipeline.WriteTo is called.
Example ¶
package main
import (
"context"
"fmt"
"github.com/deahtstroke/chainmorph"
)
type user struct {
Name string
Department string
BadgeNumber int
}
type sliceReader[T any] struct {
data []T
}
func (r *sliceReader[T]) ReadFrom(ctx context.Context) (T, error) {
var zero T
if len(r.data) <= 0 {
return zero, chainmorph.ErrEndOfStream
}
var item T
item, r.data = r.data[0], r.data[1:]
return item, nil
}
type stdWriter[T any] struct{}
func (w *stdWriter[T]) Write(ctx context.Context, item T) error {
fmt.Println(item)
return nil
}
type userFilter struct{}
func (f *userFilter) Accept(ctx context.Context, item user) (bool, error) {
return item.BadgeNumber == 1, nil
}
type userMapper struct{}
func (m *userMapper) Map(ctx context.Context, item user) (string, error) {
return item.Name, nil
}
func main() {
ctx := context.Background()
var reader *sliceReader[user] = &sliceReader[user]{
data: seedUsers(),
}
var filter *userFilter
var mapper *userMapper
var writer *stdWriter[string]
if err := chainmorph.From(reader).
Filter(filter).
MapTo(mapper).
WriteTo(ctx, writer); err != nil {
fmt.Println("error:", err)
}
}
func seedUsers() []user {
return []user{
{
Name: "Daniel",
Department: "Software Support",
BadgeNumber: 1,
},
{
Name: "Zac",
Department: "Apps Team",
BadgeNumber: 1,
},
{
Name: "Andre",
Department: "Software Development",
BadgeNumber: 2,
},
{
Name: "Jason",
Department: "Software Support",
BadgeNumber: 2,
},
}
}
Output: Daniel Zac
func From ¶
func From[T any](itemReader ItemReader[T]) *Pipeline[T]
From builds a pipeline sourced from itemReader. See ErrEndOfStream for the contract itemReader must follow to signal completion.
func Just ¶
Just builds a pipeline that yields each of the elements in order, then ends. Unlike From, it needs no ItemReader: elements are consumed directly.
func (*Pipeline[T]) Filter ¶
func (p *Pipeline[T]) Filter(itemFilter ItemFilter[T]) *Pipeline[T]
Filter keeps only items for which itemFilter.Accept returns true Rejected items are silently skipped. They are never surfaced to callers or to later pipeline stages.
func (*Pipeline[T]) FilterFunc ¶
FilterFunc wraps f as a Predicate and calls Filter with it, letting callers pass a plain function instead of an ItemFilter implementation.
func (*Pipeline[T]) If ¶
If is a simplified Pipeline.Filter for predicates that don't need context or the ability to fail
func (*Pipeline[T]) MapFunc ¶
MapFunc wraos f as a MapperFunc and calls MapTo with it, letting callers pass a plain function instead of an ItemMapper implementation.
func (*Pipeline[T]) MapTo ¶
func (p *Pipeline[T]) MapTo[R any](itemMapper ItemMapper[T, R]) *Pipeline[R]
MapTo transforms each item from T to R using itemMapper, changing the pipeline's element type. This method's own type parameter, R, is independent of the receiver's T. This capability is only available to methods since Go 1.27.
func (*Pipeline[T]) Tap ¶
Tap runs tapper.Tap on each item as a side effect, then passes the item through unchanged. Use it for logging, metrics, or publishing events without altering the pipeline's data.
func (*Pipeline[T]) TapFunc ¶
TapFunc wraps f as a TapFunc and calls Tap with it, letting callers pass a plain function instead of a Tapper implementation.
func (*Pipeline[T]) WriteTo ¶
func (p *Pipeline[T]) WriteTo(ctx context.Context, itemWriter ItemWriter[T]) error
WriteTo drains the pipeline, pulling and writing items one at a time until the source is exhausted or an error occurs. It returns nil once the pipeline ends cleanly, or the first error encountered, either from upstream or itemWriter.Write
type Predicate ¶
Predicate adapts a plain function into an ItemFilter. It's the mechanism behind Pipeline.FilterFunc and Pipeline.If.
type TapFunc ¶
TapFunc adapts a plain function into a Tapper. It's the mechanism behind Pipeline.TapFunc
Source Files