pgstore

Documentation

Overview

Package pgstore is a PostgreSQL-backed checklist or "attempt list".

Sample Use Case

Let's say you want to download a million items, and you suspect not all items will get successfully downloaded on the first attempt.

Start with a list of the items to be downloaded.

connectionURL := "postgresql://iidy:password@localhost:5432/iidy?pool_max_conns=5&application_name=iidy"
s, _ := pgstore.NewPgStore(connectionURL)
listName := "downloads"
listItems := []string{"a.txt", "b.txt", "c.txt", "d.txt", "e.txt", "f.txt"}
s.AddBatch(context.Background(), listName, listItems)

A worker can get a certain number of items to work on:

// gets "a.txt", "b.txt", "c.txt"
items, _ := s.GetBatch(context.Background(), listName, "", 3)

For items that were unsuccessfully downloaded, the number of failed attempts is incremented for that item. (A business rule can be set to abandon downloading an item after a certain number of attempts.)

count, _ := s.IncrementBatch(context.Background(), ListName, []string{"a.txt", "c.txt"})

Items that were successfully downloaded can be removed from the list.

count, _ := s.DeleteBatch(context.Background(), ListName, []string{"b.txt"})

A worker can get more items from the list, starting past the last item in the previously-worked-on batch:

// gets "d.txt", "e.txt", "f.txt"
items, _ := s.GetBatch(context.Background(), listName, "c.txt", 3)

And the cycle can continue.

Index

• Constants
• type ListEntry
• type PgStore
  • func NewPgStore(connectionURL string) (*PgStore, error)
  • func (p *PgStore) DeleteBatch(ctx context.Context, list string, items []string) (int64, error)
  • func (p *PgStore) DeleteOne(ctx context.Context, list string, item string) (int64, error)
  • func (p *PgStore) GetBatch(ctx context.Context, list string, startID string, count int) ([]ListEntry, error)
  • func (p *PgStore) GetOne(ctx context.Context, list string, item string) (int, bool, error)
  • func (p *PgStore) IncrementBatch(ctx context.Context, list string, items []string) (int64, error)
  • func (p *PgStore) IncrementOne(ctx context.Context, list string, item string) (int64, error)
  • func (p *PgStore) InsertBatch(ctx context.Context, list string, items []string) (int64, error)
  • func (p *PgStore) InsertOne(ctx context.Context, list string, item string) (int64, error)
  • func (p *PgStore) Nuke(ctx context.Context) error
  • func (p *PgStore) String() string
• type Store

Constants

const DefaultConnectionURL string = "postgresql://postgres:postgres@localhost:5432/postgres?pool_max_conns=5&application_name=iidy"

DefaultConnectionURL is the default connection URL to the PostgreSQL database, including connection pool config and application_name config.

const TernDefaultMigrationTable string = "public.schema_version"

Types

type ListEntry

type ListEntry struct {
	Item     string `json:"item"`
	Attempts int    `json:"attempts"`
}

ListEntry is a list item and the number of times an attempt has been made to complete it.

type PgStore

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

PgStore is the backend store where lists and list items are kept.

func NewPgStore

func NewPgStore(connectionURL string) (*PgStore, error)

NewPgStore returns a pointer to a new PgStore. It's best to treat an instance of PgStore like a singleton, and have only one per process. connectionURL is a connection string is formatted like so,

postgresql://[user[:password]@][netloc][:port][,...][/dbname][?param1=value1&...]

according to https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING.

If connectionURL is the empty string, DefaultConnectionURL will be used.

func (*PgStore) DeleteBatch

func (p *PgStore) DeleteBatch(ctx context.Context, list string, items []string) (int64, error)

DeleteBatch deletes a slice of items (strings) from the specified list. The first return value is the number of items successfully deleted, generally len(items) or 0.

func (*PgStore) DeleteOne

func (p *PgStore) DeleteOne(ctx context.Context, list string, item string) (int64, error)

DeleteOne deletes an item from a list. The first return value is the number of items that were successfully deleted (1 or 0).

func (*PgStore) GetBatch

func (p *PgStore) GetBatch(ctx context.Context, list string, startID string, count int) ([]ListEntry, error)

GetBatch gets a slice of ListEntries from the specified list (alphabetically sorted), starting after the startID, or from the beginning of the list, if startID is an empty string. If there is nothing to be found, an empty slice is returned.

The general pattern being followed here is explained very well at http://use-the-index-luke.com/sql/partial-results/fetch-next-page

func (*PgStore) GetOne

func (p *PgStore) GetOne(ctx context.Context, list string, item string) (int, bool, error)

GetOne returns the number of attempts that were made to complete an item in a list. When a list or list item is missing, the number of attempts will be returned as 0, but the second return argument (commonly assiged to "ok") will be false.

func (*PgStore) IncrementBatch

func (p *PgStore) IncrementBatch(ctx context.Context, list string, items []string) (int64, error)

IncrementBatch increments the attempts count for each item in the items slice for the specified list. The first return value is the number of items successfully incremented, generally len(items) or 0.

func (*PgStore) IncrementOne

func (p *PgStore) IncrementOne(ctx context.Context, list string, item string) (int64, error)

IncrementOne increments the number of attempts to complete an item from a list. The first return value is the number of items found and incremented (1 or 0).

func (*PgStore) InsertBatch

func (p *PgStore) InsertBatch(ctx context.Context, list string, items []string) (int64, error)

InsertBatch adds a slice of items (strings) to the specified list, and sets their completion attempt counts to 0. The first return value is the number of items successfully inserted, generally len(items) or 0.

func (*PgStore) InsertOne

func (p *PgStore) InsertOne(ctx context.Context, list string, item string) (int64, error)

InsertOne adds an item to a list. If the list does not already exist, it will be created.

func (*PgStore) Nuke

func (p *PgStore) Nuke(ctx context.Context) error

Nuke destroys every list in the data store. Mostly used for testing. Use with caution.

func (*PgStore) String

func (p *PgStore) String() string

String gives us a string representation of the config for the data store. This is handy for debugging, or just for printing the connection info at program startup.

type Store

type Store interface {
	InsertOne(ctx context.Context, list string, item string) (int64, error)
	GetOne(ctx context.Context, list string, item string) (int, bool, error)
	DeleteOne(ctx context.Context, list string, item string) (int64, error)
	IncrementOne(ctx context.Context, list string, item string) (int64, error)
	InsertBatch(ctx context.Context, list string, items []string) (int64, error)
	GetBatch(ctx context.Context, list string, startID string, count int) ([]ListEntry, error)
	DeleteBatch(ctx context.Context, list string, items []string) (int64, error)
	IncrementBatch(ctx context.Context, list string, items []string) (int64, error)
}

Store describes list storage methods, in case we want to have a different implementation than the pg implementation.