Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

pipeline

package

Go to main page

Versions in this module

v1

v1.2.6

Nov 7, 2025

v1.2.5

Nov 6, 2025

v1.2.5-rc1

Nov 6, 2025

v1.2.4

Oct 28, 2025

v1.2.3

Oct 27, 2025

v1.2.2

Jul 22, 2025

v1.2.1

Jul 21, 2025

v1.2.0

Jul 21, 2025

v1.2.0-rc0

May 21, 2025

v1.1.1

Feb 19, 2025

v1.1.1-RC.1

Feb 17, 2025

v1.1.0

Feb 14, 2025

v1.0.6

Feb 5, 2025

v1.0.5

Feb 1, 2025

v1.0.4

Jan 28, 2025

v1.0.4-RC.1

Jan 28, 2025

v1.0.3

Jan 24, 2025

v1.0.3-RC.1

Jan 23, 2025

v1.0.2

Jan 21, 2025

v1.0.1

Jan 20, 2025

v1.0.0

Jan 7, 2025

Changes in this version

+ const InputTaskKey

+ var ErrBadInput = errors.New("bad input for task")

+ var ErrCancelled = errors.New("task run cancelled (fail early)")

+ var ErrDisallowedIP = errors.New("disallowed IP")

+ var ErrInputTaskErrored = errors.New("input task errored")

+ var ErrKeypathNotFound = errors.New("keypath not found")

+ var ErrKeypathTooDeep = errors.New("keypath too deep (maximum 2 keys)")

+ var ErrOverflow = errors.New("overflow")

+ var ErrParameterEmpty = errors.New("parameter is empty")

+ var ErrTaskRunFailed = errors.New("task run failed")

+ var ErrTimeout = errors.New("timeout")

+ var ErrTooManyErrors = errors.New("too many errors")

+ var ErrVarsRoot = errors.New("cannot get/set the root of a pipeline.Vars")

+ var ErrWrongInputCardinality = errors.New("wrong number of task inputs")

+ func CheckInputs(inputs []Result, minLen, maxLen, maxErrors int) ([]interface{}, error)

+ func ConnectWebSocket(ctx context.Context, websocketUrl, urlHeader string, maxRetries int) (conn *websocket.Conn, err error)

+ func NewRunner(lggr log.Logger) *runner

+ func ParseETHABIArgsString(theABI []byte, isLog bool) (args abi.Arguments, indexedArgs abi.Arguments, _ error)

+ func ResolveParam(out PipelineParamUnmarshaler, getters []GetterFunc) error

+ func ToDecimal(input interface{}) (decimal.Decimal, error)

+ func WrapRecoverHandle(lggr log.Logger, fn func(), onPanic func(interface{}))

+ type AddressParam common.Address

+ func (a *AddressParam) UnmarshalPipelineParam(val interface{}) error

+ type AddressSliceParam []common.Address

+ func (s *AddressSliceParam) UnmarshalPipelineParam(val interface{}) error

+ type AnyTask struct

+ func (t *AnyTask) Run(_ context.Context, _ log.Logger, _ Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *AnyTask) Type() TaskType

+ type BaseTask struct

+ FailEarly bool

+ Index int32

+ MaxBackoff time.Duration

+ MinBackoff time.Duration

+ Retries null.Uint32

+ Timeout *time.Duration

+ func NewBaseTask(id int, dotID string, inputs []TaskDependency, outputs []Task, index int32) BaseTask

+ func (t *BaseTask) Base() *BaseTask

+ func (t BaseTask) DotID() string

+ func (t BaseTask) ID() int

+ func (t BaseTask) Inputs() []TaskDependency

+ func (t BaseTask) OutputIndex() int32

+ func (t BaseTask) Outputs() []Task

+ func (t BaseTask) TaskMaxBackoff() time.Duration

+ func (t BaseTask) TaskMinBackoff() time.Duration

+ func (t BaseTask) TaskRetries() uint32

+ func (t BaseTask) TaskTimeout() (time.Duration, bool)

+ type BoolParam bool

+ func (b *BoolParam) UnmarshalPipelineParam(val interface{}) error

+ type BytesParam []byte

+ func (b *BytesParam) UnmarshalPipelineParam(val interface{}) error

+ type DecimalParam decimal.Decimal

+ func (d *DecimalParam) UnmarshalPipelineParam(val interface{}) error

+ func (d DecimalParam) Decimal() decimal.Decimal

+ type DecimalSliceParam []decimal.Decimal

+ func (s *DecimalSliceParam) UnmarshalPipelineParam(val interface{}) error

+ type DivideTask struct

+ Divisor string

+ Input string

+ Precision string

+ func (t *DivideTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *DivideTask) Type() TaskType

+ type ETHABIDecodeLogTask struct

+ ABI string

+ Data string

+ Topics string

+ func (t *ETHABIDecodeLogTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ETHABIDecodeLogTask) Type() TaskType

+ type ETHABIDecodeTask struct

+ ABI string

+ Data string

+ func (t *ETHABIDecodeTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ETHABIDecodeTask) Type() TaskType

+ type ETHABIEncodeTask struct

+ ABI string

+ Data string

+ func (t *ETHABIEncodeTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ETHABIEncodeTask) Type() TaskType

+ type ETHABIEncodeTask2 struct

+ ABI string

+ Data string

+ func (t *ETHABIEncodeTask2) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (Result, RunInfo)

+ func (t *ETHABIEncodeTask2) Type() TaskType

+ type ErrRunPanicked struct

+ func (err ErrRunPanicked) Error() string

+ type FailTask struct

+ Msg string

+ func (t *FailTask) Run(_ context.Context, _ log.Logger, vars Vars, _ []Result) (Result, RunInfo)

+ func (t *FailTask) Type() TaskType

+ type FinalResult struct

+ AllErrors []error

+ FatalErrors []error

+ Values []interface{}

+ func (result FinalResult) HasErrors() bool

+ func (result FinalResult) HasFatalErrors() bool

+ func (result FinalResult) SingularResult() (Result, error)

+ type GetterFunc func() (interface{}, error)

+ func From(getters ...interface{}) []GetterFunc

+ func Input(inputs []Result, index int) GetterFunc

+ func Inputs(inputs []Result) GetterFunc

+ func JSONWithVarExprs(s string, vars Vars, allowErrors bool) GetterFunc

+ func NonemptyString(s string) GetterFunc

+ func VarExpr(s string, vars Vars) GetterFunc

+ type Graph struct

+ func NewGraph() *Graph

+ func (g *Graph) AddImplicitDependenciesAsEdges()

+ func (g *Graph) IsImplicitEdge(uid, vid int64) bool

+ func (g *Graph) NewEdge(from, to graph.Node) graph.Edge

+ func (g *Graph) NewNode() graph.Node

+ func (g *Graph) UnmarshalText(bs []byte) (err error)

+ type GraphEdge struct

+ func (e *GraphEdge) IsImplicit() bool

+ func (e *GraphEdge) SetIsImplicit(isImplicit bool)

+ type GraphNode struct

+ func (n *GraphNode) Attributes() []encoding.Attribute

+ func (n *GraphNode) DOTID() string

+ func (n *GraphNode) SetAttribute(attr encoding.Attribute) error

+ func (n *GraphNode) SetDOTID(id string)

+ func (n *GraphNode) String() string

+ type HTTPRequest struct

+ Logger log.Logger

+ Request *http.Request

+ func (h *HTTPRequest) SendRequest() (responseBody []byte, statusCode int, headers http.Header, err error)

+ type HTTPTask struct

+ HeaderMap string

+ Method string

+ RequestData string

+ URL string

+ func (t *HTTPTask) Run(ctx context.Context, lggr log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *HTTPTask) Type() TaskType

+ type HashSliceParam []common.Hash

+ func (s *HashSliceParam) UnmarshalPipelineParam(val interface{}) error

+ type JSONParseTask struct

+ Data string

+ Lax string

+ Path string

+ func (t *JSONParseTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *JSONParseTask) Type() TaskType

+ type JSONPathParam []string

+ func (p *JSONPathParam) UnmarshalPipelineParam(val interface{}) error

+ type JSONSerializable struct

+ Val interface{}

+ Valid bool

+ func (js *JSONSerializable) Empty() bool

+ func (js *JSONSerializable) Scan(value interface{}) error

+ func (js *JSONSerializable) UnmarshalJSON(bs []byte) error

+ func (js JSONSerializable) MarshalJSON() ([]byte, error)

+ func (js JSONSerializable) Value() (driver.Value, error)

+ type Keypath [2][]byte

+ func (keypath Keypath) NumParts() int

+ func (keypath Keypath) String() string

+ type LowercaseTask struct

+ Input string

+ func (t *LowercaseTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *LowercaseTask) Type() TaskType

+ type MapParam map[string]interface

+ func (m *MapParam) UnmarshalPipelineParam(val interface{}) error

+ func (m MapParam) Map() map[string]interface{}

+ type MaybeBigIntParam struct

+ func (p *MaybeBigIntParam) UnmarshalPipelineParam(val interface{}) error

+ func (p MaybeBigIntParam) BigInt() *big.Int

+ type MaybeInt32Param struct

+ func (p *MaybeInt32Param) UnmarshalPipelineParam(val interface{}) error

+ func (p MaybeInt32Param) Int32() (int32, bool)

+ type MaybeUint64Param struct

+ func (p *MaybeUint64Param) UnmarshalPipelineParam(val interface{}) error

+ func (p MaybeUint64Param) Uint64() (uint64, bool)

+ type MeanTask struct

+ AllowedFaults string

+ Precision string

+ Values string

+ func (t *MeanTask) Run(ctx context.Context, lggr log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *MeanTask) Type() TaskType

+ type MedianTask struct

+ AllowedFaults string

+ Values string

+ func (t *MedianTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *MedianTask) Type() TaskType

+ type MemoTask struct

+ Value string

+ func (t *MemoTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (Result, RunInfo)

+ func (t *MemoTask) Type() TaskType

+ type MergeTask struct

+ Left string

+ Right string

+ func (t *MergeTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *MergeTask) Type() TaskType

+ type Method struct

+ Inputs abi.Arguments

+ Name string

+ type ModeTask struct

+ AllowedFaults string

+ Values string

+ func (t *ModeTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *ModeTask) Type() TaskType

+ type MultiplyTask struct

+ Input string

+ Times string

+ func (t *MultiplyTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *MultiplyTask) Type() TaskType

+ type ObjectParam struct

+ BoolValue BoolParam

+ DecimalValue DecimalParam

+ MapValue MapParam

+ SliceValue SliceParam

+ StringValue StringParam

+ Type ObjectType

+ func MustNewObjectParam(val interface{}) *ObjectParam

+ func (o *ObjectParam) UnmarshalPipelineParam(val interface{}) error

+ func (o ObjectParam) Marshal() (string, error)

+ func (o ObjectParam) MarshalJSON() ([]byte, error)

+ func (o ObjectParam) String() string

+ type ObjectType int

+ const BoolType

+ const DecimalType

+ const MapType

+ const NilType

+ const SliceType

+ const StringType

+ type PanicTask struct

+ Msg string

+ func (t *PanicTask) Run(_ context.Context, _ log.Logger, vars Vars, _ []Result) (result Result, runInfo RunInfo)

+ func (t *PanicTask) Type() TaskType

+ type Pipeline struct

+ Source string

+ Tasks []Task

+ func Parse(text string) (*Pipeline, error)

+ func (p *Pipeline) ByDotID(id string) Task

+ func (p *Pipeline) MinTimeout() (time.Duration, bool, error)

+ func (p *Pipeline) UnmarshalText(bs []byte) (err error)

+ type PipelineParamUnmarshaler interface

+ UnmarshalPipelineParam func(val interface{}) error

+ type PossibleErrorResponses struct

+ Error string

+ ErrorMessage string

+ type Result struct

+ Error error

+ Value interface{}

+ func (result Result) ErrorDB() null.String

+ func (result Result) OutputDB() JSONSerializable

+ type ResumeRequest struct

+ Error null.String

+ Value json.RawMessage

+ func (rr ResumeRequest) ToResult() (Result, error)

+ type Run struct

+ AllErrors RunErrors

+ CreatedAt time.Time

+ FailEarly bool

+ FatalErrors RunErrors

+ FinishedAt null.Time

+ ID int64

+ Inputs JSONSerializable

+ Meta JSONSerializable

+ Outputs JSONSerializable

+ Pending bool

+ PipelineSpec Spec

+ PipelineSpecID int32

+ PipelineTaskRuns []TaskRun

+ State RunStatus

+ func NewRun(spec Spec, vars Vars) Run

+ func (r *Run) ByDotID(id string) *TaskRun

+ func (r *Run) SetID(value string) error

+ func (r *Run) Status() RunStatus

+ func (r *Run) StringAllErrors() []*string

+ func (r *Run) StringFatalErrors() []*string

+ func (r *Run) StringOutputs() ([]*string, error)

+ func (r Run) GetID() string

+ func (r Run) HasErrors() bool

+ func (r Run) HasFatalErrors() bool

+ type RunErrors []null.String

+ func (re *RunErrors) Scan(value interface{}) error

+ func (re RunErrors) HasError() bool

+ func (re RunErrors) ToError() error

+ func (re RunErrors) Value() (driver.Value, error)

+ type RunInfo struct

+ IsPending bool

+ IsRetryable bool

+ type RunStatus string

+ const RunStatusCompleted

+ const RunStatusErrored

+ const RunStatusRunning

+ const RunStatusSuspended

+ const RunStatusUnknown

+ func (s RunStatus) Completed() bool

+ func (s RunStatus) Errored() bool

+ func (s RunStatus) Finished() bool

+ type Runner interface

+ ExecuteRun func(ctx context.Context, spec Spec, vars Vars, l log.Logger) (run Run, trrs TaskRunResults, err error)

+ type SliceParam []interface

+ func (s *SliceParam) UnmarshalPipelineParam(val interface{}) error

+ func (s SliceParam) FilterErrors() (SliceParam, int)

+ type Spec struct

+ CreatedAt time.Time

+ DotDagSource string

+ ID int32

+ JobID int32

+ JobName string

+ func (s Spec) Pipeline() (*Pipeline, error)

+ type StringParam string

+ func (s *StringParam) UnmarshalPipelineParam(val interface{}) error

+ type SumTask struct

+ AllowedFaults string

+ Values string

+ func (t *SumTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *SumTask) Type() TaskType

+ type Task interface

+ Base func() *BaseTask

+ DotID func() string

+ ID func() int

+ Inputs func() []TaskDependency

+ OutputIndex func() int32

+ Outputs func() []Task

+ Run func(ctx context.Context, lggr log.Logger, vars Vars, inputs []Result) (Result, RunInfo)

+ TaskMaxBackoff func() time.Duration

+ TaskMinBackoff func() time.Duration

+ TaskRetries func() uint32

+ TaskTimeout func() (time.Duration, bool)

+ Type func() TaskType

+ func UnmarshalTaskFromMap(taskType TaskType, taskMap interface{}, ID int, dotID string) (_ Task, err error)

+ type TaskDependency struct

+ InputTask Task

+ PropagateResult bool

+ type TaskRun struct

+ CreatedAt time.Time

+ DotID string

+ Error null.String

+ FinishedAt null.Time

+ ID uuid.UUID

+ Index int32

+ Output JSONSerializable

+ PipelineRun Run

+ PipelineRunID int64

+ Type TaskType

+ func (tr *TaskRun) IsPending() bool

+ func (tr *TaskRun) SetID(value string) error

+ func (tr TaskRun) GetDotID() string

+ func (tr TaskRun) GetID() string

+ func (tr TaskRun) Result() Result

+ type TaskRunResult struct

+ Attempts uint

+ CreatedAt time.Time

+ FinishedAt null.Time

+ ID uuid.UUID

+ Result Result

+ Task Task

+ TaskRun TaskRun

+ func (result *TaskRunResult) IsPending() bool

+ func (result *TaskRunResult) IsTerminal() bool

+ type TaskRunResults []TaskRunResult

+ func (trrs TaskRunResults) FinalResult(l log.Logger) FinalResult

+ type TaskType string

+ const TaskTypeAny

+ const TaskTypeDivide

+ const TaskTypeETHABIDecode

+ const TaskTypeETHABIDecodeLog

+ const TaskTypeETHABIEncode

+ const TaskTypeETHABIEncode2

+ const TaskTypeFail

+ const TaskTypeHTTP

+ const TaskTypeJSONParse

+ const TaskTypeLowercase

+ const TaskTypeMean

+ const TaskTypeMedian

+ const TaskTypeMemo

+ const TaskTypeMerge

+ const TaskTypeMode

+ const TaskTypeMultiply

+ const TaskTypePanic

+ const TaskTypeSum

+ const TaskTypeUppercase

+ func (t TaskType) String() string

+ type URLParam url.URL

+ func (u *URLParam) String() string

+ func (u *URLParam) UnmarshalPipelineParam(val interface{}) error

+ type Uint64Param uint64

+ func (u *Uint64Param) UnmarshalPipelineParam(val interface{}) error

+ type UppercaseTask struct

+ Input string

+ func (t *UppercaseTask) Run(_ context.Context, _ log.Logger, vars Vars, inputs []Result) (result Result, runInfo RunInfo)

+ func (t *UppercaseTask) Type() TaskType

+ type Vars struct

+ func NewVarsFrom(m map[string]interface{}) Vars

+ func (vars Vars) Copy() Vars

+ func (vars Vars) Get(keypathStr string) (interface{}, error)

+ func (vars Vars) Set(dotID string, value interface{})