etl

Documentation

Index

• func ToViaJSON[T any](source any) (target T, err error)
• func Upsert(columns ...string) clause.OnConflict
• func WithRequestInputFunc(ctx context.Context, fn RequestInputFunc) context.Context
• func WithTimeout(ctx context.Context, timeout time.Duration, fn func(ctx context.Context))
• type BatchIterator
  • func (it *BatchIterator[V]) Run(fn func(log *Logger, value V, batchSize int) (*V, error)) (errs error)
• type Logger
  • func (l *Logger) Error(errs *error, err error, desc string) bool
  • func (l *Logger) With(args ...any) *Logger
  • func (l *Logger) WithDesc(key, value string) *Logger
• type Pipeline
• type Registry
  • func (r *Registry) Register(id string, pipeline Pipeline)
  • func (r *Registry) Run(ctx context.Context, log *slog.Logger, username string) error
• type RequestInputFunc
  • func GetRequestInputFunc(ctx context.Context) RequestInputFunc

Functions

func ToViaJSON

func ToViaJSON[T any](source any) (target T, err error)

func Upsert

func Upsert(columns ...string) clause.OnConflict

func WithRequestInputFunc

func WithRequestInputFunc(ctx context.Context, fn RequestInputFunc) context.Context

func WithTimeout

func WithTimeout(ctx context.Context, timeout time.Duration, fn func(ctx context.Context))

Types

type BatchIterator

type BatchIterator[V any] struct {
	BatchSize int
	Log       *Logger
	Key       string
	Value     V
}

func (*BatchIterator[V]) Run

func (it *BatchIterator[V]) Run(fn func(log *Logger, value V, batchSize int) (*V, error)) (errs error)

type Logger

type Logger struct {
	*slog.Logger
	// contains filtered or unexported fields
}

func (*Logger) Error

func (l *Logger) Error(errs *error, err error, desc string) bool

func (*Logger) With

func (l *Logger) With(args ...any) *Logger

func (*Logger) WithDesc

func (l *Logger) WithDesc(key, value string) *Logger

type Pipeline

type Pipeline interface {
	Run(ctx context.Context, log *Logger, username string) error
}

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

func (*Registry) Register

func (r *Registry) Register(id string, pipeline Pipeline)

func (*Registry) Run

func (r *Registry) Run(ctx context.Context, log *slog.Logger, username string) error

type RequestInputFunc

type RequestInputFunc func(ctx context.Context, text string) (string, error)

var RequestInputStdin RequestInputFunc = func(ctx context.Context, text string) (string, error) {
	reader := bufio.NewReader(os.Stdin)
	fmt.Printf("%s: ", text)
	text, err := reader.ReadString('\n')
	if err != nil {
		return "", errors.Wrap(err, "read line from stdin")
	}

	return strings.Trim(text, " \n\t\v"), nil
}

func GetRequestInputFunc

func GetRequestInputFunc(ctx context.Context) RequestInputFunc