Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
6128f82149 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -1,2 +1,3 @@
|
||||
cmd
|
||||
coverage.out
|
||||
example
|
||||
|
22
LICENSE
Normal file
22
LICENSE
Normal file
@@ -0,0 +1,22 @@
|
||||
Copyright 2025 Dmitry Fedotov
|
||||
|
||||
Permission is hereby granted, free of charge, to any person
|
||||
obtaining a copy of this software and associated documentation
|
||||
files (the “Software”), to deal in the Software without
|
||||
restriction, including without limitation the rights to use,
|
||||
copy, modify, merge, publish, distribute, sublicense, and/or
|
||||
sell copies of the Software, and to permit persons to whom
|
||||
the Software is furnished to do so,
|
||||
subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall
|
||||
be included in all copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND,
|
||||
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
|
||||
OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
||||
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
|
||||
HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
|
||||
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
|
||||
OR OTHER DEALINGS IN THE SOFTWARE.
|
137
README.md
Normal file
137
README.md
Normal file
@@ -0,0 +1,137 @@
|
||||
# package script
|
||||
|
||||
This package is intended as a collection of helper functions and tools fow quick construction of scripts that process .csv tables.
|
||||
|
||||
Note that this is a work in progress. API is not guaranteed to be stable.
|
||||
|
||||
## Example usage
|
||||
|
||||
This demonstrates a very basic case: reading from stdin, checking whether array len is 2 and switching array elements.
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"code.uint32.ru/dmitry/script"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer stop()
|
||||
|
||||
reader := script.NewStdinReader()
|
||||
writer := script.NewStdoutWriter()
|
||||
|
||||
processor := func(_ context.Context, in []string) ([]string, error) {
|
||||
if len(in) != 2 {
|
||||
return nil, errors.New("incorrect input len")
|
||||
}
|
||||
|
||||
return []string{in[1], in[0]}, nil
|
||||
}
|
||||
|
||||
conf := script.RunConfig{
|
||||
Input: reader,
|
||||
Output: writer,
|
||||
Processor: processor,
|
||||
}
|
||||
|
||||
if err := script.Run(ctx, conf); err != nil {
|
||||
fmt.Println(err)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
A more conplicated example featuring a processor which is itself a struct with external dependency.
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
|
||||
"code.uint32.ru/dmitry/script"
|
||||
)
|
||||
|
||||
// ExternalDep represent an external dependency
|
||||
// which your script utilizes. This can be a DB connection etc.
|
||||
type ExternalDep interface {
|
||||
QueryID(string) ([]string, error)
|
||||
}
|
||||
|
||||
type Processor struct {
|
||||
dep ExternalDep
|
||||
}
|
||||
|
||||
func New(e ExternalDep) *Processor {
|
||||
|
||||
return &Processor{
|
||||
dep: e,
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Processor) Process(ctx context.Context, in []string) ([]string, error) {
|
||||
var out []string
|
||||
|
||||
result, err := p.dep.QueryID(in[0])
|
||||
if err != nil {
|
||||
// you can as well record the error and continue
|
||||
// processing without yielding the error here
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out = append(out, result...)
|
||||
out = append(out, "processed")
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer stop()
|
||||
|
||||
r, err := script.NewCSVReader("/tmp/some_file.csv")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := r.Close(); err != nil {
|
||||
fmt.Println("err closing reader", err)
|
||||
}
|
||||
}()
|
||||
|
||||
w, err := script.NewCSVWriter("/tmp/some_output.csv")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if err := w.Close(); err != nil {
|
||||
fmt.Println("err closing writer", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// intializing the processor
|
||||
p := New(ExternalDep(nil))
|
||||
|
||||
conf := script.RunConfig{
|
||||
Input: r,
|
||||
Output: w,
|
||||
Processor: p.Process, // Process implements script.Processor
|
||||
}
|
||||
|
||||
if err := script.Run(ctx, conf); err != nil {
|
||||
fmt.Println(err)
|
||||
}
|
||||
}
|
||||
```
|
@@ -1,7 +1,6 @@
|
||||
package script
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
)
|
||||
|
||||
@@ -16,7 +15,7 @@ func NewMemReader(records [][]string) *MemReader {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MemReader) Read(context.Context) ([]string, error) {
|
||||
func (m *MemReader) Read() ([]string, error) {
|
||||
if m.curr == len(m.rows) || len(m.rows) == 0 {
|
||||
return nil, io.EOF
|
||||
}
|
||||
|
@@ -1,7 +1,6 @@
|
||||
package script
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"os"
|
||||
)
|
||||
@@ -16,7 +15,7 @@ type CSVReader struct {
|
||||
rdr *csv.Reader
|
||||
}
|
||||
|
||||
func (c *CSVReader) Read(context.Context) ([]string, error) {
|
||||
func (c *CSVReader) Read() ([]string, error) {
|
||||
return c.rdr.Read()
|
||||
}
|
||||
|
||||
|
@@ -2,7 +2,6 @@ package script
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
@@ -23,7 +22,7 @@ func NewStdinReader() *StdinReader {
|
||||
// then splits result at "," separator and returns
|
||||
// the resulting slice.
|
||||
// It returns EOF when nothing left to read.
|
||||
func (s *StdinReader) Read(_ context.Context) ([]string, error) {
|
||||
func (s *StdinReader) Read() ([]string, error) {
|
||||
if !s.scanner.Scan() {
|
||||
err := s.scanner.Err()
|
||||
if err == nil {
|
||||
|
@@ -1,7 +1,6 @@
|
||||
package script
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"errors"
|
||||
"os"
|
||||
@@ -20,7 +19,7 @@ type CSVWriter struct {
|
||||
}
|
||||
|
||||
// Write writer row to csv. Writes are buffered
|
||||
func (c *CSVWriter) Write(_ context.Context, record []string) error {
|
||||
func (c *CSVWriter) Write(record []string) error {
|
||||
err := c.wr.Write(record)
|
||||
c.err = err
|
||||
|
||||
|
@@ -1,7 +1,5 @@
|
||||
package script
|
||||
|
||||
import "context"
|
||||
|
||||
type MemWriter struct {
|
||||
rows [][]string
|
||||
}
|
||||
@@ -10,7 +8,7 @@ func NewMemWriter() *MemWriter {
|
||||
return new(MemWriter)
|
||||
}
|
||||
|
||||
func (m *MemWriter) Write(_ context.Context, record []string) error {
|
||||
func (m *MemWriter) Write(record []string) error {
|
||||
m.rows = append(m.rows, record)
|
||||
|
||||
return nil
|
||||
|
@@ -1,13 +1,11 @@
|
||||
package script
|
||||
|
||||
import "context"
|
||||
|
||||
type NopWriter struct{}
|
||||
|
||||
func NewNopWriter() *NopWriter {
|
||||
return new(NopWriter)
|
||||
}
|
||||
|
||||
func (d *NopWriter) Write(context.Context, []string) error {
|
||||
func (d *NopWriter) Write([]string) error {
|
||||
return nil
|
||||
}
|
@@ -1,7 +1,6 @@
|
||||
package script
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
@@ -12,7 +11,7 @@ func NewStdoutWriter() *StdoutWriter {
|
||||
return new(StdoutWriter)
|
||||
}
|
||||
|
||||
func (s *StdoutWriter) Write(_ context.Context, in []string) error {
|
||||
func (s *StdoutWriter) Write(in []string) error {
|
||||
_, err := os.Stdout.WriteString(strings.Join(in, ",") + "\n")
|
||||
return err
|
||||
}
|
||||
|
12
runner.go
12
runner.go
@@ -28,13 +28,13 @@ type Reader interface {
|
||||
// Read is called by Processor to obtain input
|
||||
// for processing. Read must return EOF/io.EOF to indicate
|
||||
// that there is no more input.
|
||||
Read(context.Context) ([]string, error)
|
||||
Read() ([]string, error)
|
||||
}
|
||||
|
||||
type Writer interface {
|
||||
// Write is called by Processor to record
|
||||
// output.
|
||||
Write(context.Context, []string) error
|
||||
Write([]string) error
|
||||
}
|
||||
|
||||
// Processor accepts string slice, does what it should
|
||||
@@ -76,7 +76,7 @@ func Run(ctx context.Context, r RunConfig) error {
|
||||
defer close(rdch)
|
||||
|
||||
for range r.Offset {
|
||||
_, err := r.Input.Read(ctx)
|
||||
_, err := r.Input.Read()
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not advance to required offset: %w", err)
|
||||
}
|
||||
@@ -84,7 +84,7 @@ func Run(ctx context.Context, r RunConfig) error {
|
||||
|
||||
count := 0
|
||||
for {
|
||||
inp, err := r.Input.Read(ctx)
|
||||
inp, err := r.Input.Read()
|
||||
if err != nil && errors.Is(err, EOF) {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
@@ -111,8 +111,10 @@ func Run(ctx context.Context, r RunConfig) error {
|
||||
grp.Go(func() error {
|
||||
// not paying attention to context here
|
||||
// because we must complete writes
|
||||
// this is run within group so that write
|
||||
// error would cancel group context
|
||||
for outp := range wrch {
|
||||
if err := r.Output.Write(ctx, outp); err != nil {
|
||||
if err := r.Output.Write(outp); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
@@ -41,7 +41,7 @@ func TestBasicRun(t *testing.T) {
|
||||
|
||||
type infiniteReader struct{}
|
||||
|
||||
func (ir *infiniteReader) Read(_ context.Context) ([]string, error) {
|
||||
func (ir *infiniteReader) Read() ([]string, error) {
|
||||
return []string{"infinity", "looks", "like", "this"}, nil
|
||||
}
|
||||
|
||||
|
Reference in New Issue
Block a user