plugin

Documentation

Index

• Variables
• func MatchesTable(name string, includeTablesPattern []string, skipTablesPattern []string) bool
• func RecordDiff(l, r arrow.Record) string
• func TestWriterSuiteRunner(t *testing.T, p *Plugin, tests WriterTestSuiteTests, ...)
• func TotalRows(records []arrow.Record) int64
• func WithTestDataOptions(opts schema.TestSourceOptions) func(o *WriterTestSuite)
• func WithTestIgnoreNullsInLists() func(o *WriterTestSuite)
• type Client
• type DestinationClient
• type MigrateMode
  • func (m MigrateMode) String() string
• type NewClientFunc
• type NewPluginFunc
• type NewSourceClientFunc
• type Option
• type Plugin
  • func NewPlugin(name string, version string, newClient NewClientFunc, options ...Option) *Plugin
  • func NewSourcePlugin(name string, version string, newClient NewSourceClientFunc, options ...Option) *Plugin
  • func (p *Plugin) Close(ctx context.Context) error
  • func (p *Plugin) Init(ctx context.Context, spec []byte) error
  • func (p *Plugin) Name() string
  • func (p *Plugin) SetLogger(logger zerolog.Logger)
  • func (p *Plugin) Sync(ctx context.Context, options SyncOptions, res chan<- message.Message) error
  • func (p *Plugin) SyncAll(ctx context.Context, options SyncOptions) (message.Messages, error)
  • func (p *Plugin) Tables(ctx context.Context) (schema.Tables, error)
  • func (p *Plugin) Version() string
  • func (p *Plugin) Write(ctx context.Context, options WriteOptions, res <-chan message.Message) error
  • func (p *Plugin) WriteAll(ctx context.Context, options WriteOptions, resources []message.Message) error
• type SafeMigrations
• type SourceClient
• type SyncOptions
• type UnimplementedDestination
  • func (UnimplementedDestination) Read(context.Context, *schema.Table, chan<- arrow.Record) error
  • func (UnimplementedDestination) Write(context.Context, WriteOptions, <-chan message.Message) error
• type UnimplementedSource
  • func (UnimplementedSource) Sync(context.Context, SyncOptions, chan<- message.Message) error
  • func (UnimplementedSource) Tables(context.Context) (schema.Tables, error)
• type WriteOptions
• type WriterTestSuite
• type WriterTestSuiteTests

Variables

var ErrNotImplemented = fmt.Errorf("not implemented")

Functions

func MatchesTable

func MatchesTable(name string, includeTablesPattern []string, skipTablesPattern []string) bool

func RecordDiff

func RecordDiff(l, r arrow.Record) string

func TestWriterSuiteRunner

func TestWriterSuiteRunner(t *testing.T, p *Plugin, tests WriterTestSuiteTests, opts ...func(o *WriterTestSuite))

func TotalRows

func TotalRows(records []arrow.Record) int64

func WithTestDataOptions

func WithTestDataOptions(opts schema.TestSourceOptions) func(o *WriterTestSuite)

func WithTestIgnoreNullsInLists

func WithTestIgnoreNullsInLists() func(o *WriterTestSuite)

Types

type Client

type Client interface {
	SourceClient
	DestinationClient
}

type DestinationClient

type DestinationClient interface {
	Close(ctx context.Context) error
	Read(ctx context.Context, table *schema.Table, res chan<- arrow.Record) error
	Write(ctx context.Context, options WriteOptions, res <-chan message.Message) error
}

type MigrateMode

type MigrateMode int

const (
	MigrateModeSafe MigrateMode = iota
	MigrateModeForce
)

func (MigrateMode) String

func (m MigrateMode) String() string

type NewClientFunc

type NewClientFunc func(context.Context, zerolog.Logger, []byte) (Client, error)

type NewPluginFunc

type NewPluginFunc func() *Plugin

type NewSourceClientFunc

type NewSourceClientFunc func(context.Context, zerolog.Logger, any) (SourceClient, error)

type Option

type Option func(*Plugin)

type Plugin

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

Plugin is the base structure required to pass to sdk.serve We take a declarative approach to API here similar to Cobra

func NewPlugin

func NewPlugin(name string, version string, newClient NewClientFunc, options ...Option) *Plugin

NewPlugin returns a new CloudQuery Plugin with the given name, version and implementation. Depending on the options, it can be a write-only plugin, read-only plugin, or both.

func NewSourcePlugin

func NewSourcePlugin(name string, version string, newClient NewSourceClientFunc, options ...Option) *Plugin

NewSourcePlugin returns a new CloudQuery Plugin with the given name, version and implementation. Source plugins only support read operations. For Read & Write plugin use NewPlugin.

func (*Plugin) Close

func (p *Plugin) Close(ctx context.Context) error

func (*Plugin) Init

func (p *Plugin) Init(ctx context.Context, spec []byte) error

Init initializes the plugin with the given spec.

func (*Plugin) Name

func (p *Plugin) Name() string

Name return the name of this plugin

func (*Plugin) SetLogger

func (p *Plugin) SetLogger(logger zerolog.Logger)

func (*Plugin) Sync

func (p *Plugin) Sync(ctx context.Context, options SyncOptions, res chan<- message.Message) error

Sync is syncing data from the requested tables in spec to the given channel

func (*Plugin) SyncAll

func (p *Plugin) SyncAll(ctx context.Context, options SyncOptions) (message.Messages, error)

func (*Plugin) Tables

func (p *Plugin) Tables(ctx context.Context) (schema.Tables, error)

func (*Plugin) Version

func (p *Plugin) Version() string

Version returns the version of this plugin

func (*Plugin) Write

func (p *Plugin) Write(ctx context.Context, options WriteOptions, res <-chan message.Message) error

func (*Plugin) WriteAll

func (p *Plugin) WriteAll(ctx context.Context, options WriteOptions, resources []message.Message) error

WriteAll is currently used mostly for testing, so it's not a public api

type SafeMigrations

type SafeMigrations struct {
	AddColumn           bool
	AddColumnNotNull    bool
	RemoveColumn        bool
	RemoveColumnNotNull bool
	ChangeColumn        bool
}

SafeMigrations defines which migrations are supported by the plugin in safe migrate mode

type SourceClient

type SourceClient interface {
	Close(ctx context.Context) error
	Tables(ctx context.Context) (schema.Tables, error)
	Sync(ctx context.Context, options SyncOptions, res chan<- message.Message) error
}

type SyncOptions

type SyncOptions struct {
	Tables            []string
	SkipTables        []string
	DeterministicCQID bool
}

type UnimplementedDestination

type UnimplementedDestination struct{}

func (UnimplementedDestination) Read

func (UnimplementedDestination) Read(context.Context, *schema.Table, chan<- arrow.Record) error

func (UnimplementedDestination) Write

func (UnimplementedDestination) Write(context.Context, WriteOptions, <-chan message.Message) error

type UnimplementedSource

type UnimplementedSource struct{}

func (UnimplementedSource) Sync

func (UnimplementedSource) Sync(context.Context, SyncOptions, chan<- message.Message) error

func (UnimplementedSource) Tables

func (UnimplementedSource) Tables(context.Context) (schema.Tables, error)

type WriteOptions

type WriteOptions struct {
	MigrateForce bool
}

type WriterTestSuite

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

type WriterTestSuiteTests

type WriterTestSuiteTests struct {
	// SkipUpsert skips testing with message.Insert and Upsert=true.
	// Usually when a destination is not supporting primary keys
	SkipUpsert bool

	// SkipDeleteStale skips testing message.Delete events.
	SkipDeleteStale bool

	// SkipAppend skips testing message.Insert and Upsert=false.
	SkipInsert bool

	// SkipMigrate skips testing migration
	SkipMigrate bool

	// SafeMigrations defines which tests should work with force migration
	// and which should pass with safe migration
	SafeMigrations SafeMigrations
}