spec

package
v0.1.0-alpha.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 13, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Overview

Package spec owns the canonical pipeline specification shape and YAML/JSON file decoding.

Index

Constants

View Source 
const (
	OperationTypeSQL       = catalog.OperationTypeSQL
	OperationTypeBigQuery  = catalog.OperationTypeBigQuery
	OperationTypeSnowflake = catalog.OperationTypeSnowflake
	OperationTypeHTTP      = catalog.OperationTypeHTTP
	OperationTypeFTP       = catalog.OperationTypeFTP
	OperationTypeGit       = catalog.OperationTypeGit
	OperationTypeMongoDB   = catalog.OperationTypeMongoDB
	OperationTypeHubSpot   = catalog.OperationTypeHubSpot
	OperationTypeS3        = catalog.OperationTypeS3
)

Variables

View Source 
var ErrUnsupportedFormat = xerror.UnsupportedPipelineFormat()

Functions

func IsCanonicalFile 

func IsCanonicalFile(path string) bool

IsCanonicalFile reports whether path should be considered a committed pipeline spec during repository-wide validation.

func IsSupportedFile 

func IsSupportedFile(path string) bool

IsSupportedFile reports whether path has a supported spec extension.

func IsSupportedQueryOperationType 

func IsSupportedQueryOperationType(operationType string) bool

IsSupportedQueryOperationType reports whether the catalog knows the operation type.

func IsSupportedSinkOperationType 

func IsSupportedSinkOperationType(operationType string) bool

IsSupportedSinkOperationType reports whether the operation type can write to a sink.

func SinkBackendForOperationType 

func SinkBackendForOperationType(operationType string) (model.SinkBackend, bool)

SinkBackendForOperationType returns the sink backend attached to an operation type.

func SinkIdempotencyClassStrings 

func SinkIdempotencyClassStrings() []string

SinkIdempotencyClassStrings returns a slice of all String values of the enum

func SupportsIncrementalSourceCheckpoint 

func SupportsIncrementalSourceCheckpoint(
	operationType string,
	checkpointKind model.IncrementalCheckpointKind,
) bool

SupportsIncrementalSourceCheckpoint reports whether an operation type accepts the checkpoint kind.

func SupportsIncrementalSourceCheckpointKind 

func SupportsIncrementalSourceCheckpointKind(operationType, checkpointKind string) bool

SupportsIncrementalSourceCheckpointKind parses a checkpoint kind before capability lookup.

func SupportsSinkIdempotencyClass 

func SupportsSinkIdempotencyClass(
	dataSourceType model.DataSourceType,
	idempotencyClass SinkIdempotencyClass,
) bool

SupportsSinkIdempotencyClass reports whether the concrete sink connector supports the replay mechanism declared by a pipeline sink step.

Types

type ConnectorCapability 

type ConnectorCapability struct {
	DataSourceType                   model.DataSourceType
	SchemaFamily                     string
	QueryOperationType               string
	SinkBackend                      model.SinkBackend
	SupportsPipelineSource           bool
	SupportsPipelineSink             bool
	SupportedSinkReplayStrategies    []SinkIdempotencyClass
	IncrementalSourceCheckpointKinds []model.IncrementalCheckpointKind
	ForcedReadOnly                   bool
}

ConnectorCapability is the pipeline-facing subset of connector catalog facts. It lets same-repository frontend and builder packages consume connector behavior without importing the connector implementation boundary directly. If the frontend becomes a separately versioned artifact or repository, this descriptor contract belongs in OpenAPI or generated API metadata instead.

func AllConnectorCapabilities 

func AllConnectorCapabilities() []ConnectorCapability

AllConnectorCapabilities returns catalog connector facts in a pipeline-owned shape. Slice fields are copied so callers can sort or filter without mutating catalog state.

type IncrementalSourceContract 

type IncrementalSourceContract struct {
	CheckpointKind *model.IncrementalCheckpointKind `json:"checkpoint_kind,omitempty" yaml:"checkpoint_kind,omitempty"`
	Field          string                           `json:"field,omitempty"           yaml:"field,omitempty"`
	Cursor         string                           `json:"cursor,omitempty"          yaml:"cursor,omitempty"`
	LookbackWindow string                           `json:"lookback_window,omitempty" yaml:"lookback_window,omitempty"`
	Ordering       []string                         `json:"ordering,omitempty"        yaml:"ordering,omitempty"`
	DedupeKey      string                           `json:"dedupe_key,omitempty"      yaml:"dedupe_key,omitempty"`
}

IncrementalSourceContract declares source-step checkpoint behavior.

type InputSpec 

type InputSpec struct {
	OutputContract *payload.Contract `json:"output_contract" yaml:"output_contract" validate:"required"`
}

InputSpec declares the one run-scoped JSON payload accepted by a pipeline.

type Metadata 

type Metadata struct {
	Slug slug.Slug `json:"slug" yaml:"slug" validate:"required,slug"`
	Name string    `json:"name" yaml:"name" validate:"required"`
}

Metadata holds the pipeline identity fields.

type Pipeline 

type Pipeline struct {
	Metadata Metadata `json:"metadata" yaml:"metadata" validate:"required"`
	Steps    []Step   `json:"steps"    yaml:"steps"    validate:"min=1,dive"`
}

Pipeline is the top-level spec loaded from a pipeline .yaml or .json file.

func Load 

func Load(path string, data []byte) (*Pipeline, error)

Load decodes a YAML or JSON pipeline spec from data. path is used only to detect file format and produce useful error messages. Semantic validation belongs to internal/pipeline/validate.

type QueryRef 

type QueryRef struct {
	DataSourceSlug  slug.Slug                  `json:"data_source_ref"            yaml:"data_source_ref"            validate:"required,slug"`
	OperationType   string                     `json:"operation_type"             yaml:"operation_type"             validate:"required"`
	OperationRef    slug.Slug                  `json:"operation_ref,omitempty"    yaml:"operation_ref,omitempty"    validate:"omitempty,slug"`
	OperationInline map[string]any             `json:"operation_inline,omitempty" yaml:"operation_inline,omitempty" validate:"-"`
	OutputContract  *payload.Contract          `json:"output_contract,omitempty"  yaml:"output_contract,omitempty"  validate:"-"`
	Parameters      map[string]any             `json:"parameters,omitempty"       yaml:"parameters,omitempty"`
	Incremental     *IncrementalSourceContract `json:"incremental,omitempty"      yaml:"incremental,omitempty"      validate:"-"`
}

QueryRef identifies the operation to execute. Specs must define operation_type plus exactly one of operation_ref or operation_inline.

Parameter values may be plain values or step output references of the form $steps.<step-slug>.<field> (e.g. $steps.fetch-users.rows). References are resolved at runtime by the Runner before the operation is executed.

type SinkIdempotencyClass 

type SinkIdempotencyClass int

SinkIdempotencyClass declares the durable replay strategy for one sink operation. Pipeline authors must choose a strategy deliberately; the runtime uses it to decide whether a delivery can be repeated after an uncertain crash-after-effect outcome. 

const (
	SinkIdempotencyClassUnknown              SinkIdempotencyClass = iota // unknown
	SinkIdempotencyClassNaturallyIdempotent                              // naturally_idempotent
	SinkIdempotencyClassIdempotentWithKey                                // idempotent_with_key
	SinkIdempotencyClassDeduplicatedByLedger                             // deduplicated_by_ledger
	SinkIdempotencyClassUnsafe                                           // unsafe
)

func SinkIdempotencyClassString 

func SinkIdempotencyClassString(s string) (SinkIdempotencyClass, error)

SinkIdempotencyClassString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.

func SinkIdempotencyClassValues 

func SinkIdempotencyClassValues() []SinkIdempotencyClass

SinkIdempotencyClassValues returns all values of the enum

func (SinkIdempotencyClass) IsASinkIdempotencyClass 

func (i SinkIdempotencyClass) IsASinkIdempotencyClass() bool

IsASinkIdempotencyClass returns "true" if the value is listed in the enum definition. "false" otherwise

func (SinkIdempotencyClass) IsReplaySafe 

func (class SinkIdempotencyClass) IsReplaySafe() bool

IsReplaySafe reports whether repeating the provider call after a durable attempt record is known to preserve the external effect.

func (SinkIdempotencyClass) MarshalJSON 

func (i SinkIdempotencyClass) MarshalJSON() ([]byte, error)

MarshalJSON implements the json.Marshaler interface for SinkIdempotencyClass

func (SinkIdempotencyClass) MarshalText 

func (i SinkIdempotencyClass) MarshalText() ([]byte, error)

MarshalText implements the encoding.TextMarshaler interface for SinkIdempotencyClass

func (SinkIdempotencyClass) String 

func (i SinkIdempotencyClass) String() string

func (*SinkIdempotencyClass) UnmarshalJSON 

func (i *SinkIdempotencyClass) UnmarshalJSON(data []byte) error

UnmarshalJSON implements the json.Unmarshaler interface for SinkIdempotencyClass

func (*SinkIdempotencyClass) UnmarshalText 

func (i *SinkIdempotencyClass) UnmarshalText(text []byte) error

UnmarshalText implements the encoding.TextUnmarshaler interface for SinkIdempotencyClass

type SinkSpec 

type SinkSpec struct {
	Input            *StepInput           `json:"input,omitempty"            yaml:"input,omitempty"            validate:"omitempty"`
	InputContract    *payload.Contract    `json:"input_contract,omitempty"   yaml:"input_contract,omitempty"   validate:"-"`
	DataSourceSlug   slug.Slug            `json:"data_source_ref,omitempty"  yaml:"data_source_ref,omitempty"  validate:"omitempty,slug"`
	OperationType    string               `json:"operation_type,omitempty"   yaml:"operation_type,omitempty"`
	OperationRef     slug.Slug            `json:"operation_ref,omitempty"    yaml:"operation_ref,omitempty"    validate:"omitempty,slug"`
	OperationInline  map[string]any       `json:"operation_inline,omitempty" yaml:"operation_inline,omitempty" validate:"-"`
	Config           map[string]any       `json:"config,omitempty"           yaml:"config,omitempty"`
	IdempotencyClass SinkIdempotencyClass `json:"idempotency_class,omitempty" yaml:"idempotency_class,omitempty" validate:"sinkidempotencyclass"`
}

SinkSpec configures pipeline output delivery for kind=sink steps.

type Step 

type Step struct {
	Kind      string         `json:"kind"                yaml:"kind"                validate:"required"`
	Slug      slug.Slug      `json:"slug"                yaml:"slug"                validate:"required,slug"`
	Name      string         `json:"name,omitempty"      yaml:"name,omitempty"`
	Input     *InputSpec     `json:"input,omitempty"     yaml:"input,omitempty"`
	Query     *QueryRef      `json:"query,omitempty"     yaml:"query,omitempty"`
	Transform *TransformSpec `json:"transform,omitempty" yaml:"transform,omitempty"`
	Sink      *SinkSpec      `json:"sink,omitempty"      yaml:"sink,omitempty"`
}

Step is a pipeline step. The Kind field determines which block is populated: kind=query populates Query; kind=transform populates Transform; kind=sink populates Sink.

type StepInput 

type StepInput struct {
	FromStep slug.Slug `validate:"required,slug"`
}

StepInput declares the upstream step consumed by a transform or sink step.

func (StepInput) MarshalJSON 

func (input StepInput) MarshalJSON() ([]byte, error)

MarshalJSON emits the canonical scalar step input form.

func (StepInput) MarshalYAML 

func (input StepInput) MarshalYAML() (any, error)

MarshalYAML emits the canonical scalar step input form.

func (*StepInput) UnmarshalJSON 

func (input *StepInput) UnmarshalJSON(data []byte) error

UnmarshalJSON accepts the canonical scalar step input form.

func (*StepInput) UnmarshalYAML 

func (input *StepInput) UnmarshalYAML(value *yaml.Node) error

UnmarshalYAML accepts the canonical scalar step input form.

type TransformSpec 

type TransformSpec struct {
	SQL            string                    `json:"sql"                       yaml:"sql"                       validate:"required"`
	OutputKind     model.TransformOutputKind `json:"output_kind"               yaml:"output_kind"               validate:"required,transformoutputkind"`
	Inputs         map[string]StepInput      `json:"inputs"                    yaml:"inputs"                    validate:"-"`
	InputContract  *payload.Contract         `json:"input_contract,omitempty"  yaml:"input_contract,omitempty"  validate:"-"`
	OutputContract *payload.Contract         `json:"output_contract,omitempty" yaml:"output_contract,omitempty" validate:"-"`
}

TransformSpec configures the sql_v1 transform behavior.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.