db

Documentation

Index

• Variables
• type CursorError
• type DSN
• type Loader
  • func NewLoader(psqlDsn string, logger *zap.Logger, tracer logging.Tracer) (*Loader, error)
  • func (l *Loader) Delete(tableName string, primaryKey string) error
  • func (l *Loader) Flush(ctx context.Context, moduleHash string, cursor *sink.Cursor) (err error)
  • func (l *Loader) GetAvailableTablesInSchema() string
  • func (l *Loader) GetCursor(ctx context.Context, outputModuleHash string) (*sink.Cursor, error)
  • func (l *Loader) GetIdentifier() string
  • func (l *Loader) GetSchema() string
  • func (l *Loader) HasTable(tableName string) bool
  • func (l *Loader) Insert(tableName string, primaryKey string, data map[string]string) error
  • func (l *Loader) LoadTables() error
  • func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error
  • func (l *Loader) Setup(ctx context.Context, schemaFile string) error
  • func (l *Loader) Update(tableName string, primaryKey string, data map[string]string) error
  • func (l *Loader) UpdateCursorQuery(moduleHash string, c *sink.Cursor) string
  • func (l *Loader) WriteCursor(ctx context.Context, moduleHash string, c *sink.Cursor) error
• type Operation
  • func (o *Operation) String() string
• type OperationType
• type Queryable
• type TypeGetter

Variables

var ErrCursorNotFound = errors.New("cursor not found")

Types

type CursorError

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

type DSN

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

type Loader

type Loader struct {
	*sql.DB

	EntriesCount uint64
	// contains filtered or unexported fields
}

func NewLoader

func NewLoader(psqlDsn string, logger *zap.Logger, tracer logging.Tracer) (*Loader, error)

func (*Loader) Delete

func (l *Loader) Delete(tableName string, primaryKey string) error

Delete a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) Flush

func (l *Loader) Flush(ctx context.Context, moduleHash string, cursor *sink.Cursor) (err error)

func (*Loader) GetAvailableTablesInSchema

func (l *Loader) GetAvailableTablesInSchema() string

func (*Loader) GetCursor

func (l *Loader) GetCursor(ctx context.Context, outputModuleHash string) (*sink.Cursor, error)

func (*Loader) GetIdentifier

func (l *Loader) GetIdentifier() string

GetIdentifier returns <database>/<schema> suitable for user presentation

func (*Loader) GetSchema

func (l *Loader) GetSchema() string

func (*Loader) HasTable

func (l *Loader) HasTable(tableName string) bool

func (*Loader) Insert

func (l *Loader) Insert(tableName string, primaryKey string, data map[string]string) error

Insert a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) LoadTables

func (l *Loader) LoadTables() error

func (*Loader) MarshalLogObject

func (l *Loader) MarshalLogObject(encoder zapcore.ObjectEncoder) error

func (*Loader) Setup

func (l *Loader) Setup(ctx context.Context, schemaFile string) error

func (*Loader) Update

func (l *Loader) Update(tableName string, primaryKey string, data map[string]string) error

Update a row in the DB, it is assumed the table exists, you can do a check before with HasTable()

func (*Loader) UpdateCursorQuery

func (l *Loader) UpdateCursorQuery(moduleHash string, c *sink.Cursor) string

func (*Loader) WriteCursor

func (l *Loader) WriteCursor(ctx context.Context, moduleHash string, c *sink.Cursor) error

type Operation

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

func (*Operation) String

func (o *Operation) String() string

type OperationType

type OperationType string

const (
	OperationTypeInsert OperationType = "INSERT"
	OperationTypeUpdate OperationType = "UPDATE"
	OperationTypeDelete OperationType = "DELETE"
)

type Queryable

type Queryable interface {
	// contains filtered or unexported methods
}

type TypeGetter

type TypeGetter func(tableName string, columnName string) (reflect.Type, error)