pg

Documentation

Index

• Constants
• func BigintOrNull(n int64) pgtype.Int8
• func Date(t time.Time) pgtype.Date
• func IsConstraintError(err error, constraint string) bool
• func PBool(b *bool) pgtype.Bool
• func PInt2(n *int16) pgtype.Int2
• func PInt32(n *int32) pgtype.Int4
• func PInt64(n *int64) pgtype.Int8
• func PText(s *string) pgtype.Text
• func PTime(t *time.Time) pgtype.Timestamptz
• func PUUID(u *uuid.UUID) pgtype.UUID
• func Rollback(tx pgx.Tx, outErr *error)
• func SafeRollback(ctx context.Context, logger *slog.Logger, tx pgx.Tx, txName string)
• func SetConnStringVariables(conn string, vars url.Values) (string, error)
• func Subscribe(ctx context.Context, logger *slog.Logger, pool *pgxpool.Pool, ...)
• func Text(s string) pgtype.Text
• func TextOrNull(s string) pgtype.Text
• func Time(t time.Time) pgtype.Timestamptz
• func TimeOrNull(t time.Time) pgtype.Timestamptz
• func ToUUIDPointer(v pgtype.UUID) *uuid.UUID
• func UUID(u uuid.UUID) pgtype.UUID
• func WithTX(ctx context.Context, pool TransactionBeginner, fn func(tx pgx.Tx) error) (outErr error)
• type ChannelSubscription
• type FanOut
  • func NewFanOut[T any](channel string) *FanOut[T]
  • func (f *FanOut[T]) ChannelName() string
  • func (f *FanOut[T]) Listen(ctx context.Context, l chan T, test func(v T) bool)
  • func (f *FanOut[T]) ListenAll(ctx context.Context, l chan T)
  • func (f *FanOut[T]) Notify(msg T)
  • func (f *FanOut[T]) NotifyWithPayload(data []byte) error
• type JobLock
  • func NewJobLock(db *pgxpool.Pool, logger *slog.Logger, name string, opts JobLockOptions) (*JobLock, error)
  • func (jl *JobLock) Identity() string
  • func (jl *JobLock) RunWithContext(ctx context.Context, fn func(ctx context.Context) error) error
  • func (jl *JobLock) Stop()
• type JobLockOptions
• type JobLockState
• type TransactionBeginner

Constants

const (
	JobLockStateNone     = ""
	JobLockStateHeld     = "held"
	JobLockStateLost     = "lost"
	JobLockStateReleased = "released"
)

Functions

func BigintOrNull

func BigintOrNull(n int64) pgtype.Int8

BigintOrNull returns a pgtype.Int8 for the given value, but will return a Int8 value that represents null in the database if the value is zero.

func Date

func Date(t time.Time) pgtype.Date

Date converts a stdlib time.Time to a pgtype.Date.

func IsConstraintError

func IsConstraintError(err error, constraint string) bool

IsConstraintError checks if an error was caused by a specific constraint violation.

func PBool

func PBool(b *bool) pgtype.Bool

PBool converts a *bool to a pgtype.Bool.

func PInt2

func PInt2(n *int16) pgtype.Int2

PInt2 returns a pgtype.Int2 for the given value, but will return a Int2 value that represents null in the database if the value is nil.

func PInt32

func PInt32(n *int32) pgtype.Int4

PInt32 converts a *int32 to a pgtype.Int4.

func PInt64

func PInt64(n *int64) pgtype.Int8

PInt64 converts a *int64 to a pgtype.Int8.

func PText

func PText(s *string) pgtype.Text

PText converts a *string to a pgtype.Text.

func PTime

func PTime(t *time.Time) pgtype.Timestamptz

PTime converts a stdlib *time.Time to a pgtype.Timestamptz.

func PUUID

func PUUID(u *uuid.UUID) pgtype.UUID

PUUID converts a *uuid.UUID to a pgtype.UUID.

func Rollback

func Rollback(tx pgx.Tx, outErr *error)

Rollback rolls back a transaction and joins the rollback error to the outError if the rollback fails. If the transaction already has been committed/closed it's not treated as an error.

Defer a call to Rollback directly after a transaction has been created. That will give you the guarantee that everything you've done will be rolled back if you return early before committing.

func SafeRollback

func SafeRollback(
	ctx context.Context, logger *slog.Logger, tx pgx.Tx, txName string,
)

SafeRollback rolls back a transaction and logs if the rollback fails. If the transaction already has been closed it's not treated as an error.

Deprecated: use Rollback() instead.

func SetConnStringVariables

func SetConnStringVariables(conn string, vars url.Values) (string, error)

SetConnStringVariables parses a connection string URI and adds the given query string variables to it.

func Subscribe

func Subscribe(
	ctx context.Context,
	logger *slog.Logger,
	pool *pgxpool.Pool,
	channels ...ChannelSubscription,
)

Subscribe opens a connection to the database and subscribes to the provided channels. Blocks until the context is cancelled.

func Text

func Text(s string) pgtype.Text

Text converts a string to a pgtype.Text.

func TextOrNull

func TextOrNull(s string) pgtype.Text

TextOrNull returns a pgtype.Text for the given string, but will return a Text value that represents null in the database if the string is empty.

func Time

func Time(t time.Time) pgtype.Timestamptz

Time converts a stdlib time.Time to a pgtype.Timestamptz.

func TimeOrNull

func TimeOrNull(t time.Time) pgtype.Timestamptz

Time converts a stdlib time.Time to a pgtype.Timestamptz, but will return a Timestamptz that represents a null value in the database if t is zero.

func ToUUIDPointer

func ToUUIDPointer(v pgtype.UUID) *uuid.UUID

ToUUIDPointer converts a pgtype.UUID to a *uuid.UUID.

func UUID

func UUID(u uuid.UUID) pgtype.UUID

UUID converts a uuid.UUID to a pgtype.UUID.

func WithTX

func WithTX(
	ctx context.Context, pool TransactionBeginner,
	fn func(tx pgx.Tx) error,
) (outErr error)

WithTX starts a transaction and calls the given function with it. If the function returns an error or panics the transaction will be rolled back.

Types

type ChannelSubscription

type ChannelSubscription interface {
	ChannelName() string
	NotifyWithPayload(data []byte) error
}

type FanOut

type FanOut[T any] struct {
	// contains filtered or unexported fields
}

func NewFanOut

func NewFanOut[T any](channel string) *FanOut[T]

func (*FanOut[T]) ChannelName

func (f *FanOut[T]) ChannelName() string

func (*FanOut[T]) Listen

func (f *FanOut[T]) Listen(ctx context.Context, l chan T, test func(v T) bool)

func (*FanOut[T]) ListenAll

func (f *FanOut[T]) ListenAll(ctx context.Context, l chan T)

func (*FanOut[T]) Notify

func (f *FanOut[T]) Notify(msg T)

func (*FanOut[T]) NotifyWithPayload

func (f *FanOut[T]) NotifyWithPayload(data []byte) error

type JobLock

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

JobLock helps separate processes coordinate who should be performing a (background) task through postgres.

func NewJobLock

func NewJobLock(
	db *pgxpool.Pool, logger *slog.Logger, name string,
	opts JobLockOptions,
) (*JobLock, error)

NewJobLock creates a new job lock.

func (*JobLock) Identity

func (jl *JobLock) Identity() string

func (*JobLock) RunWithContext

func (jl *JobLock) RunWithContext(
	ctx context.Context,
	fn func(ctx context.Context) error,
) error

RunWithContext runs the provided function once the job lock has been acquired. The context provided to the function will be cancelled if the job lock is lost.

func (*JobLock) Stop

func (jl *JobLock) Stop()

Stop releases the job lock if held and stops all polling.

type JobLockOptions

type JobLockOptions struct {
	// PingInterval controls how often the job locked should be
	// pinged/renewed. Defaults to 10s.
	PingInterval time.Duration
	// StaleAfter controls after how long a time a held lock should be
	// considered stale and other clients will start attempting to steal
	// it. Must be longer than the ping interval. Defaults to four times the
	// ping interval.
	StaleAfter time.Duration
	// CheckInterval controls how often clients should check if a held lock
	// has become stale. Defaults to twice the ping interval.
	CheckInterval time.Duration
	// Timeout is the timeout that should be used for all lock
	// operations. Must be shorter than the ping interval. Defaults to half
	// the ping interval.
	Timeout time.Duration
}

JobLockOptions controls how a job lock should behave.

type JobLockState

type JobLockState string

type TransactionBeginner

type TransactionBeginner interface {
	Begin(context.Context) (pgx.Tx, error)
}

TransactionBeginner is the interface for something that can start a pgx transaction for use with WithTX().