batchiterator

README

go-batch-iterator

go-batch-iterator provides an iterator to sequentially iterate over datasources utilizing a batch approach.
It utilizes generics to achieve an independence from underlying datastructures. Therefore, it is ready to serve rows from a database or objects from an api endpoint with pagination. All the user has to do is provide the logic to fetch the next batch from their datasource.

go get -u github.com/Eun/go-batch-iterator

Example Usage

package batchiterator_test

import (
	"context"
	"database/sql"
	"errors"
	"log"
	"testing"

	batchiterator "github.com/Eun/go-batch-iterator"
	_ "github.com/glebarez/go-sqlite"
)

func ExampleIterator() {
	db, err := sql.Open("sqlite", ":memory:")
	if err != nil {
		log.Print(err)
		return
	}
	defer db.Close()

	_, err = db.Exec(`
CREATE TABLE users (name TEXT);
INSERT INTO users (name) VALUES ("Alice"), ("Bob"), ("Charlie"), ("George"), ("Gerald"), ("Joe"), ("John"), ("Zoe");
`)
	if err != nil {
		log.Print(err)
		return
	}

	maxRowsPerQuery := 3
	offset := 0
	iter := batchiterator.Iterator[string]{
		NextBatchFunc: func(ctx context.Context) (items []string, hasMoreItems bool, err error) {
			rows, err := db.QueryContext(ctx, "SELECT name FROM users LIMIT ? OFFSET ?", maxRowsPerQuery, offset)
			if err != nil {
				if errors.Is(err, sql.ErrNoRows) {
					return nil, false, nil
				}
				return nil, false, err
			}
			defer rows.Close()

			var names []string
			for rows.Next() {
				var name string
				if err := rows.Scan(&name); err != nil {
					return nil, false, err
				}
				names = append(names, name)
			}
			if err := rows.Err(); err != nil {
				return nil, false, err
			}
			offset += maxRowsPerQuery
			moreRowsAvailable := len(names) == maxRowsPerQuery
			return names, moreRowsAvailable, nil
		},
	}
	for iter.Next(context.Background()) {
		log.Print(*iter.Value())
	}
	if err := iter.Error(); err != nil {
		log.Print(err)
		return
	}
}

func TestExamples(t *testing.T) {
	ExampleIterator()
}

example_test.go

Helpers

StaticSlice
Sometimes a developer wants to use the iterator but the items are already fetched.

iter := batchiterator.Iterator[string]{
    NextBatchFunc: batchiterator.StaticSlice([]string{"A", "B", "C"}),
}

RateLimit
RateLimit takes a rate.Limiter and wraps the NextBatchFunc with this limit.

iter := batchiterator.Iterator[string]{
    NextBatchFunc: batchiterator.RateLimit(
		rate.NewLimiter(rate.Every(time.Second), 1), 
		func (ctx context.Context) (items []string, hasMoreItems bool, err error) {
			panic("not implemented")
		}, 
	),
}

Documentation

Overview

Package batchiterator implements an Iterator that fetches batches of items in the background transparently from the user.

Index

• type Iterator
  • func (iter *Iterator[T]) Error() error
  • func (iter *Iterator[T]) Next(ctx context.Context) bool
  • func (iter *Iterator[T]) Value() *T
• type NextBatchFunction
  • func RateLimit[T any](limiter *rate.Limiter, nextBatchFunction NextBatchFunction[T]) NextBatchFunction[T]
  • func StaticSlice[T any](s []T) NextBatchFunction[T]

Examples

• Iterator

Types

type Iterator

type Iterator[T any] struct {
	// NextBatchFunc is the function that will be called by the iterator to fetch the next batch.
	// For more details see NextBatchFunction.
	NextBatchFunc NextBatchFunction[T]
	// contains filtered or unexported fields
}

Iterator is the main object for the iterator.

Example

package main

import (
	"context"
	"database/sql"
	"errors"
	"log"

	batchiterator "github.com/Eun/go-batch-iterator"
	_ "github.com/glebarez/go-sqlite"
)

func main() {
	db, err := sql.Open("sqlite", ":memory:")
	if err != nil {
		log.Print(err)
		return
	}
	defer db.Close()

	_, err = db.Exec(`
CREATE TABLE users (name TEXT);
INSERT INTO users (name) VALUES ("Alice"), ("Bob"), ("Charlie"), ("George"), ("Gerald"), ("Joe"), ("John"), ("Zoe");
`)
	if err != nil {
		log.Print(err)
		return
	}

	maxRowsPerQuery := 3
	offset := 0
	iter := batchiterator.Iterator[string]{
		NextBatchFunc: func(ctx context.Context) (items []string, hasMoreItems bool, err error) {
			rows, err := db.QueryContext(ctx, "SELECT name FROM users LIMIT ? OFFSET ?", maxRowsPerQuery, offset)
			if err != nil {
				if errors.Is(err, sql.ErrNoRows) {
					return nil, false, nil
				}
				return nil, false, err
			}
			defer rows.Close()

			var names []string
			for rows.Next() {
				var name string
				if err := rows.Scan(&name); err != nil {
					return nil, false, err
				}
				names = append(names, name)
			}
			if err := rows.Err(); err != nil {
				return nil, false, err
			}
			offset += maxRowsPerQuery
			moreRowsAvailable := len(names) == maxRowsPerQuery
			return names, moreRowsAvailable, nil
		},
	}
	for iter.Next(context.Background()) {
		log.Print(*iter.Value())
	}
	if err := iter.Error(); err != nil {
		log.Print(err)
		return
	}
}

func (*Iterator[T]) Error

func (iter *Iterator[T]) Error() error

Error returns the current error.

func (*Iterator[T]) Next

func (iter *Iterator[T]) Next(ctx context.Context) bool

Next prepares the next item for reading with the Value method. It returns true on success, or false if there is no next value or an error happened while preparing it. Error should be consulted to distinguish between the two cases.

It calls the NextBatchFunc after all items in the batch have been depleted.

func (*Iterator[T]) Value

func (iter *Iterator[T]) Value() *T

Value returns the current value as a pointer in the batch.

type NextBatchFunction

type NextBatchFunction[T any] func(ctx context.Context) (items []T, hasMoreItems bool, err error)

NextBatchFunction is the function that will be called by the Iterator.Next function to fetch the next batch. It returns the items for the batch and indicates whether there are more items available or not. If it is unclear whether there are more items or not, it should return true and the Iterator will call the function again. In this case it can be useful to delay the execution by using the rate limit helper (RateLimit). If all items are depleted it returns the items (or nil) and false for hasMoreItems. If an error occurred the iterator will stop the execution.

func RateLimit

func RateLimit[T any](limiter *rate.Limiter, nextBatchFunction NextBatchFunction[T]) NextBatchFunction[T]

RateLimit wraps the NextBatchFunc with a rate.Limiter.

func StaticSlice

func StaticSlice[T any](s []T) NextBatchFunction[T]

StaticSlice sets the next batch function to use a static slice.