singer

Documentation

Index

• Constants
• func NewSinger(ctx context.Context, sourceConfig *base.SourceConfig, ...) (base.Driver, error)
• func TestSinger(sourceConfig *base.SourceConfig) error
• type Catalog
• type CommandCloser
  • func (cc *CommandCloser) Close() error
  • func (cc *CommandCloser) String() string
• type Config
  • func (sc *Config) Validate() error
• type Metadata
• type MetadataWrapper
• type Schema
• type SchemaRecord
• type SettingsExtractor
  • func NewFileBasedSingerSettingsExtractor(catalogPath, propertiesPath string) (*SettingsExtractor, error)
  • func (sse *SettingsExtractor) ExtractStreamReplicationMappings() (map[string]string, error)
  • func (sse *SettingsExtractor) ExtractTableNamesMappings(prefix string) (map[string]string, error)
  • func (sse *SettingsExtractor) LoadCatalog(jsonBytes []byte) error
  • func (sse *SettingsExtractor) LoadProperties(jsonBytes []byte) error
• type Singer
  • func (s *Singer) Close() (multiErr error)
  • func (s *Singer) Delete() error
  • func (s *Singer) EnsureTapAndCatalog() error
  • func (s *Singer) GetDriversInfo() *base.DriversInfo
  • func (s *Singer) IsClosed() bool
  • func (s *Singer) Load(config string, state string, taskLogger logging.TaskLogger, ...) error
  • func (s *Singer) Ready() (bool, error)
  • func (s *Singer) ReplaceTables() bool
  • func (s *Singer) Type() string
• type StreamCatalog

Constants

const (
	SINGER_REPLICATION_INCREMENTAL = "INCREMENTAL"
	SINGER_REPLICATION_FULL_TABLE  = "FULL_TABLE"
)

Functions

func NewSinger

func NewSinger(ctx context.Context, sourceConfig *base.SourceConfig, collection *base.Collection) (base.Driver, error)

NewSinger returns Singer driver and 1. writes json files (config, catalog, properties, state) if string/raw json was provided 2. runs discover and collects catalog.json 2. creates venv 3. in another goroutine: updates pip, install singer tap

func TestSinger

func TestSinger(sourceConfig *base.SourceConfig) error

TestSinger tests singer connection (runs discover) if tap has been installed otherwise returns nil

Types

type Catalog

type Catalog struct {
	Streams []StreamCatalog `json:"streams,omitempty"`
}

Catalog is a dto for Singer catalog partly serialization (only for extracting destination_table_name)

type CommandCloser

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

func (*CommandCloser) Close

func (cc *CommandCloser) Close() error

func (*CommandCloser) String

func (cc *CommandCloser) String() string

type Config

type Config struct {
	Tap                    string                     `mapstructure:"tap" json:"tap,omitempty" yaml:"tap,omitempty"`
	Config                 interface{}                `mapstructure:"config" json:"config,omitempty" yaml:"config,omitempty"`
	Catalog                interface{}                `mapstructure:"catalog" json:"catalog,omitempty" yaml:"catalog,omitempty"`
	Properties             interface{}                `mapstructure:"properties" json:"properties,omitempty" yaml:"properties,omitempty"`
	InitialState           interface{}                `mapstructure:"initial_state" json:"initial_state,omitempty" yaml:"initial_state,omitempty"`
	StreamTableNames       map[string]string          `mapstructure:"stream_table_names" json:"stream_table_names,omitempty" yaml:"stream_table_names,omitempty"`
	StreamTableNamesPrefix string                     `mapstructure:"stream_table_name_prefix" json:"stream_table_name_prefix,omitempty" yaml:"stream_table_name_prefix,omitempty"`
	SelectedStreams        []base.StreamConfiguration `mapstructure:"selected_streams" json:"selected_streams,omitempty" yaml:"selected_streams,omitempty"`
}

Config is a dto for Singer configuration serialization

func (*Config) Validate

func (sc *Config) Validate() error

Validate returns err if configuration is invalid

type Metadata

type Metadata struct {
	ReplicationMethod       string `json:"replication-method,omitempty"`
	ForcedReplicationMethod string `json:"forced-replication-method,omitempty"`
}

type MetadataWrapper

type MetadataWrapper struct {
	Breadcrumb []string `json:"breadcrumb,omitempty"`
	Metadata   Metadata `json:"metadata,omitempty"`
}

type Schema

type Schema struct {
	Properties map[string]*base.Property `json:"properties,omitempty"`
}

type SchemaRecord

type SchemaRecord struct {
	Type          string   `json:"type,omitempty"`
	Stream        string   `json:"stream,omitempty"`
	Schema        *Schema  `json:"schema,omitempty"`
	KeyProperties []string `json:"key_properties,omitempty"`
}

type SettingsExtractor

type SettingsExtractor struct {
	Catalog    *Catalog
	Properties *Catalog
}

func NewFileBasedSingerSettingsExtractor

func NewFileBasedSingerSettingsExtractor(catalogPath, propertiesPath string) (*SettingsExtractor, error)

func (*SettingsExtractor) ExtractStreamReplicationMappings

func (sse *SettingsExtractor) ExtractStreamReplicationMappings() (map[string]string, error)

func (*SettingsExtractor) ExtractTableNamesMappings

func (sse *SettingsExtractor) ExtractTableNamesMappings(prefix string) (map[string]string, error)

func (*SettingsExtractor) LoadCatalog

func (sse *SettingsExtractor) LoadCatalog(jsonBytes []byte) error

func (*SettingsExtractor) LoadProperties

func (sse *SettingsExtractor) LoadProperties(jsonBytes []byte) error

type Singer

type Singer struct {
	base.AbstractCLIDriver
	// contains filtered or unexported fields
}

Singer is a Singer CLI driver

func (*Singer) Close

func (s *Singer) Close() (multiErr error)

Close kills all commands and returns errors if occurred

func (*Singer) Delete

func (s *Singer) Delete() error

func (*Singer) EnsureTapAndCatalog

func (s *Singer) EnsureTapAndCatalog() error

EnsureTapAndCatalog ensures Singer tap via singer.Instance and does discover if catalog wasn't provided

func (*Singer) GetDriversInfo

func (s *Singer) GetDriversInfo() *base.DriversInfo

GetDriversInfo returns telemetry information about the driver

func (*Singer) IsClosed

func (s *Singer) IsClosed() bool

func (*Singer) Load

func (s *Singer) Load(config string, state string, taskLogger logging.TaskLogger, dataConsumer base.CLIDataConsumer, taskCloser base.CLITaskCloser) error

func (*Singer) Ready

func (s *Singer) Ready() (bool, error)

Ready returns true if catalog is discovered and tap is installed

func (*Singer) ReplaceTables

func (s *Singer) ReplaceTables() bool

func (*Singer) Type

func (s *Singer) Type() string

type StreamCatalog

type StreamCatalog struct {
	Stream               string            `json:"stream,omitempty"`
	TapStreamID          string            `json:"tap_stream_id,omitempty"`
	DestinationTableName string            `json:"destination_table_name,omitempty"`
	Metadata             []MetadataWrapper `json:"metadata,omitempty"`
}

StreamCatalog is a dto for Singer catalog Stream object serialization