csv

Documentation

Overview

const CSV = "foo,bar,10,01/02/2006"

type Row struct {
	Foo string
	Bar string
	Amount float44
	Date time.Time `csv:"01/02/2006"`
}

reader := csv.NewReader(strings.NewReader(CSV))

var rows []Row err := NewDecoder(reader).Decode(&rows)

Package to interact with CSV files, finding, reading, processing. Uses pipeline methodoligy to implement funcations that can be composed together to create complex pipelines

Index

• Constants
• func IsCSV(filename string) bool
• func IsTSV(filename string) bool
• func ParseS3Path(s3Path string) (string, string, error)
• func WalkDirectory(directory string, logger *zl.Logger) (<-chan string, error)
• func WalkDirectoryFilter(directory string, filter FilterFunc, logger *zl.Logger) (<-chan string, error)
• type Decoder
  • func NewDecoder(reader *csv.Reader) *Decoder
  • func (t *Decoder) Decode(rows interface{}) error
  • func (t *Decoder) HasHeader()
  • func (t *Decoder) SetDateFormat(format string)
• type FilterFunc
• type PipelineFunc
• type Reader
  • func NewReader[T any]() *Reader[T]
  • func (t *Reader[T]) Stream(reader io.Reader) (<-chan T, <-chan error, error)
  • func (t Reader[T]) Walk(path string) (<-chan T, <-chan error)
• type S3Reader
  • func NewS3Reader(sess *session.Session, logger *zl.Logger) *S3Reader
  • func (t *S3Reader) List(bucket, prefix string) (<-chan string, error)
  • func (t *S3Reader) Stream(bucket, path string, header bool) (<-chan []string, error)
  • func (t *S3Reader) WithFilter(fileFilter FilterFunc) *S3Reader
  • func (t *S3Reader) WithPipeline(fns ...PipelineFunc) *S3Reader
  • func (t *S3Reader) WithTerminator(terminator rune) *S3Reader

Constants

const (
	Comma = ','
	Tab   = '\t'

	CSVTag = "csv"
)

Functions

func IsCSV

func IsCSV(filename string) bool

IsCSV FilterFunc implementation

func IsTSV

func IsTSV(filename string) bool

IsTSV FilterFunc implementation

func ParseS3Path

func ParseS3Path(s3Path string) (string, string, error)

func WalkDirectory

func WalkDirectory(directory string, logger *zl.Logger) (<-chan string, error)

WalkDirectory looking for CSV files as defined by IsCSV function. Use WalkDirectoryFilter if you need to specify a custom extension.

func WalkDirectoryFilter

func WalkDirectoryFilter(directory string, filter FilterFunc, logger *zl.Logger) (<-chan string, error)

WalkDirectoryFilter walks directory looking for files that match the FilterFunc

Types

type Decoder

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

func NewDecoder

func NewDecoder(reader *csv.Reader) *Decoder

func (*Decoder) Decode

func (t *Decoder) Decode(rows interface{}) error

func (*Decoder) HasHeader

func (t *Decoder) HasHeader()

func (*Decoder) SetDateFormat

func (t *Decoder) SetDateFormat(format string)

type FilterFunc

type FilterFunc func(string) bool

FilterFunc

type PipelineFunc

type PipelineFunc func(filename string, row []string) []string

PipelineFunc process a row to add information based on filename

type Reader

type Reader[T any] struct {
	Comma      rune
	DateFormat string
	HasHeader  bool
}

func NewReader

func NewReader[T any]() *Reader[T]

func (*Reader[T]) Stream

func (t *Reader[T]) Stream(reader io.Reader) (<-chan T, <-chan error, error)

Stream reads a CSV file and returns a channel of the generic type T representing each row, and a channel for errors.

func (Reader[T]) Walk

func (t Reader[T]) Walk(path string) (<-chan T, <-chan error)

type S3Reader

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

func NewS3Reader

func NewS3Reader(sess *session.Session, logger *zl.Logger) *S3Reader

func (*S3Reader) List

func (t *S3Reader) List(bucket, prefix string) (<-chan string, error)

listS3 List files at S3 path

func (*S3Reader) Stream

func (t *S3Reader) Stream(bucket, path string, header bool) (<-chan []string, error)

Stream - streams CSV from S3 row by row

func (*S3Reader) WithFilter

func (t *S3Reader) WithFilter(fileFilter FilterFunc) *S3Reader

func (*S3Reader) WithPipeline

func (t *S3Reader) WithPipeline(fns ...PipelineFunc) *S3Reader

func (*S3Reader) WithTerminator

func (t *S3Reader) WithTerminator(terminator rune) *S3Reader