primitive

package
v0.10.0-beta.202409251526 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 25, 2024 License: AGPL-3.0 Imports: 61 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DriverPostgres   = "postgres"
	DriverPostgresql = "postgresql"
	DriverMySQL      = "mysql"
	DriverDuckDB     = "duckdb"
	DriverSQLite3    = "sqlite3"
	DriverSQLite     = "sqlite"
)
View Source 
const EnvFlowpipeInputRouter = "FLOWPIPE_INPUT_ROUTER"

Variables

This section is empty.

Functions

func FlowpipeMetadataOutput added in v0.3.0 

func FlowpipeMetadataOutput(startedAt, finshedAt time.Time) map[string]interface{}

func GetInputRouter added in v1.0.0 

func GetInputRouter() (string, bool)

func IsInputRouted added in v1.0.0 

func IsInputRouted() bool

Types

type Container 

type Container struct {
	FullyQualifiedStepName string
}

func (*Container) Run 

func (cp *Container) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Container) ValidateInput 

func (cp *Container) ValidateInput(ctx context.Context, i modconfig.Input) error

type Email 

type Email struct {
	Input modconfig.Input
}

func (*Email) Run 

func (h *Email) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Email) ValidateInput 

func (h *Email) ValidateInput(ctx context.Context, i modconfig.Input) error

type Exec 

type Exec struct{}

func (*Exec) Run 

func (e *Exec) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Exec) ValidateInput 

func (e *Exec) ValidateInput(ctx context.Context, i modconfig.Input) error

type FileBasedQueryReader added in v0.2.1 

type FileBasedQueryReader struct {
	QueryReaderImpl
}

func (*FileBasedQueryReader) Initialize added in v0.2.1 

func (s *FileBasedQueryReader) Initialize() error

type Function 

type Function struct {
	ModPath string
}

func (*Function) Run 

func (e *Function) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Function) ValidateInput 

func (e *Function) ValidateInput(ctx context.Context, i modconfig.Input) error

type HTTPInput 

type HTTPInput struct {
	URL            string
	Method         string
	RequestBody    string
	RequestHeaders map[string]interface{}
	CaCertPem      string
	Insecure       bool
	Timeout        time.Duration
}

type HTTPRequest 

type HTTPRequest struct {
	Input modconfig.Input
}

func (*HTTPRequest) Run 

func (h *HTTPRequest) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*HTTPRequest) ValidateInput 

func (h *HTTPRequest) ValidateInput(ctx context.Context, i modconfig.Input) error

type Input 

type Input struct {
	ExecutionID         string
	PipelineExecutionID string
	StepExecutionID     string
	PipelineName        string
	StepName            string
}

func NewInputPrimitive added in v0.3.0 

func NewInputPrimitive(executionId, pipelineExecutionId, stepExecutionId, pipelineName, stepName string) *Input

func (*Input) Run 

func (ip *Input) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Input) ValidateInput 

func (ip *Input) ValidateInput(ctx context.Context, i modconfig.Input) error

type InputIntegration 

type InputIntegration interface {
	PostMessage(ctx context.Context, inputType string, prompt string, options []InputIntegrationResponseOption) (*modconfig.Output, error)
}

type InputIntegrationBase 

type InputIntegrationBase struct {
	ExecutionID         string
	PipelineExecutionID string
	StepExecutionID     string
}

func NewInputIntegrationBase added in v0.3.0 

func NewInputIntegrationBase(input *Input) InputIntegrationBase

type InputIntegrationConsole added in v0.6.0 

type InputIntegrationConsole struct {
	InputIntegrationBase
}

func NewInputIntegrationConsole added in v0.6.0 

func NewInputIntegrationConsole(base InputIntegrationBase) InputIntegrationConsole

func (*InputIntegrationConsole) PostMessage added in v0.6.0 

func (ip *InputIntegrationConsole) PostMessage(_ context.Context, mc MessageCreator, options []InputIntegrationResponseOption) (*modconfig.Output, error)

type InputIntegrationEmail 

type InputIntegrationEmail struct {
	InputIntegrationBase

	Host       *string
	Port       *int64
	SecurePort *int64
	Tls        *string
	To         []string
	Cc         []string
	Bcc        []string
	From       string
	Subject    string
	User       *string
	Pass       *string
	FormUrl    string
}

func NewInputIntegrationEmail added in v0.3.0 

func NewInputIntegrationEmail(base InputIntegrationBase) InputIntegrationEmail

func (*InputIntegrationEmail) PostMessage 

func (ip *InputIntegrationEmail) PostMessage(ctx context.Context, mc MessageCreator, options []InputIntegrationResponseOption) (*modconfig.Output, error)

func (*InputIntegrationEmail) ValidateInputIntegrationEmail 

func (ip *InputIntegrationEmail) ValidateInputIntegrationEmail(ctx context.Context, i modconfig.Input) error

type InputIntegrationEmailMessage added in v0.3.0 

type InputIntegrationEmailMessage interface {
	Message() (string, error)
}

type InputIntegrationMsTeams added in v0.4.0 

type InputIntegrationMsTeams struct {
	InputIntegrationBase
	IntegrationName string
	WebhookUrl      *string
}

func NewInputIntegrationMsTeams added in v0.4.0 

func NewInputIntegrationMsTeams(base InputIntegrationBase, name string) InputIntegrationMsTeams

func (*InputIntegrationMsTeams) PostMessage added in v0.4.0 

func (ip *InputIntegrationMsTeams) PostMessage(_ context.Context, mc MessageCreator, options []InputIntegrationResponseOption) (*modconfig.Output, error)

type InputIntegrationResponseOption added in v0.3.0 

type InputIntegrationResponseOption struct {
	Label    *string `json:"label,omitempty"`
	Value    *string `json:"value,omitempty"`
	Selected *bool   `json:"selected,omitempty"`
	Style    *string `json:"style,omitempty"`
}

type InputIntegrationSlack 

type InputIntegrationSlack struct {
	InputIntegrationBase
	Token         *string
	SigningSecret *string
	WebhookUrl    *string
	Channel       *string
}

func NewInputIntegrationSlack added in v0.3.0 

func NewInputIntegrationSlack(base InputIntegrationBase) InputIntegrationSlack

func (*InputIntegrationSlack) PostMessage 

func (ip *InputIntegrationSlack) PostMessage(ctx context.Context, mc MessageCreator, options []InputIntegrationResponseOption) (*modconfig.Output, error)

type InputStepMessageCreator added in v0.3.0 

type InputStepMessageCreator struct {
	Prompt    string
	InputType string
	StepName  string
}

func (*InputStepMessageCreator) ConsoleMessage added in v0.6.0 

func (icm *InputStepMessageCreator) ConsoleMessage(ip *InputIntegrationConsole, options []InputIntegrationResponseOption) (*string, *huh.Form, any, error)

func (*InputStepMessageCreator) EmailMessage added in v0.3.0 

func (icm *InputStepMessageCreator) EmailMessage(iim *InputIntegrationEmail, options []InputIntegrationResponseOption) (string, error)

func (*InputStepMessageCreator) MsTeamsMessage added in v0.4.0 

func (icm *InputStepMessageCreator) MsTeamsMessage(ip *InputIntegrationMsTeams, options []InputIntegrationResponseOption) (*messagecard.MessageCard, error)

func (*InputStepMessageCreator) SlackMessage added in v0.3.0 

func (icm *InputStepMessageCreator) SlackMessage(ip *InputIntegrationSlack, options []InputIntegrationResponseOption) (slack.Blocks, error)

type JSONPayload 

type JSONPayload struct {
	PipelineExecutionID string `json:"pipeline_execution_id"`
	StepExecutionID     string `json:"step_execution_id"`
	ExecutionID         string `json:"execution_id"`
}

type Message added in v0.3.0 

type Message struct {
	Input
}

func NewMessagePrimitive added in v0.3.0 

func NewMessagePrimitive(executionId, pipelineExecutionId, stepExecutionId, pipelineName, stepName string) *Message

func (*Message) Run added in v0.3.0 

func (mp *Message) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Message) ValidateInput added in v0.3.0 

func (mp *Message) ValidateInput(ctx context.Context, input modconfig.Input) error

type MessageCreator added in v0.3.0 

type MessageCreator interface {
	EmailMessage(*InputIntegrationEmail, []InputIntegrationResponseOption) (string, error)
	SlackMessage(*InputIntegrationSlack, []InputIntegrationResponseOption) (slack.Blocks, error)
	MsTeamsMessage(*InputIntegrationMsTeams, []InputIntegrationResponseOption) (*messagecard.MessageCard, error)
	ConsoleMessage(*InputIntegrationConsole, []InputIntegrationResponseOption) (*string, *huh.Form, any, error)
}

type MessageStepMessageCreator added in v0.3.0 

type MessageStepMessageCreator struct {
	Text string
}

func (*MessageStepMessageCreator) ConsoleMessage added in v0.6.0 

func (icm *MessageStepMessageCreator) ConsoleMessage(ip *InputIntegrationConsole, _ []InputIntegrationResponseOption) (*string, *huh.Form, any, error)

func (*MessageStepMessageCreator) EmailMessage added in v0.3.0 

func (icm *MessageStepMessageCreator) EmailMessage(iim *InputIntegrationEmail, _ []InputIntegrationResponseOption) (string, error)

func (*MessageStepMessageCreator) MsTeamsMessage added in v0.4.0 

func (icm *MessageStepMessageCreator) MsTeamsMessage(iit *InputIntegrationMsTeams, _ []InputIntegrationResponseOption) (*messagecard.MessageCard, error)

func (*MessageStepMessageCreator) SlackMessage added in v0.3.0 

func (icm *MessageStepMessageCreator) SlackMessage(ip *InputIntegrationSlack, options []InputIntegrationResponseOption) (slack.Blocks, error)

type MySQLQueryReader added in v0.2.1 

type MySQLQueryReader struct {
	QueryReaderImpl
}

func (*MySQLQueryReader) Query added in v0.2.1 

func (m *MySQLQueryReader) Query(ctx context.Context, queryString string, args ...interface{}) ([]map[string]interface{}, map[string]*sql.ColumnType, error)

func (*MySQLQueryReader) RowsToCty added in v0.2.1 

func (m *MySQLQueryReader) RowsToCty(rows []map[string]interface{}, columnTypes map[string]*sql.ColumnType) ([]cty.Value, error)

type MySQLRowReader added in v0.2.1 

type MySQLRowReader struct {
	RowReaderImpl
}

func (*MySQLRowReader) Read added in v0.2.1 

func (m *MySQLRowReader) Read(rows *sql.Rows, columnTypeMap map[string]*sql.ColumnType) ([]map[string]interface{}, error)

func (*MySQLRowReader) RowToCty added in v0.2.1 

func (m *MySQLRowReader) RowToCty(row map[string]interface{}, columnTypes map[string]*sql.ColumnType) (cty.Value, error)

type PostgresQueryReader added in v0.2.1 

type PostgresQueryReader struct {
	QueryReaderImpl
}

func (*PostgresQueryReader) Initialize added in v0.2.1 

func (p *PostgresQueryReader) Initialize() error

type Primitive 

type Primitive interface {
	Run(modconfig.Input) (*modconfig.Output, error)
}

type Query 

type Query struct {
	QueryReader QueryReader
}

func (*Query) Run 

func (e *Query) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Query) RunWithMetadata added in v0.2.1 

func (e *Query) RunWithMetadata(ctx context.Context, input modconfig.Input) (*modconfig.Output, map[string]*sql.ColumnType, error)

func (*Query) ValidateInput 

func (e *Query) ValidateInput(ctx context.Context, i modconfig.Input) error

type QueryReader added in v0.2.1 

type QueryReader interface {
	GetConnectionString() string
	Initialize() error
	Query(context.Context, string, ...interface{}) ([]map[string]interface{}, map[string]*sql.ColumnType, error)
	RowsToCty(rows []map[string]interface{}, columnTypes map[string]*sql.ColumnType) ([]cty.Value, error)
	Close()
}

func NewQueryReader added in v0.2.1 

func NewQueryReader(dbConnectionString string) (QueryReader, error)

type QueryReaderImpl added in v0.2.1 

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

func (*QueryReaderImpl) Close added in v0.2.1 

func (q *QueryReaderImpl) Close()

func (*QueryReaderImpl) GetConnectionString added in v0.2.1 

func (q *QueryReaderImpl) GetConnectionString() string

func (*QueryReaderImpl) Initialize added in v0.2.1 

func (q *QueryReaderImpl) Initialize() error

func (*QueryReaderImpl) Query added in v0.2.1 

func (q *QueryReaderImpl) Query(ctx context.Context, queryString string, args ...interface{}) ([]map[string]interface{}, map[string]*sql.ColumnType, error)

func (*QueryReaderImpl) RowsToCty added in v0.2.1 

func (q *QueryReaderImpl) RowsToCty(rows []map[string]interface{}, columnTypes map[string]*sql.ColumnType) ([]cty.Value, error)

type RoutedInput added in v1.0.0 

type RoutedInput struct {
	ExecutionID         string
	PipelineExecutionID string
	StepExecutionID     string
	PipelineName        string
	StepName            string
	RoutedUrl           string
	// contains filtered or unexported fields
}

func NewRoutedInput added in v1.0.0 

func NewRoutedInput(executionID, pipelineExecutionID, stepExecutionID, pipelineName, stepName, url string, endStepFunc RoutedInputEndStepFunc) *RoutedInput

func (*RoutedInput) GetShortStepName added in v1.0.0 

func (r *RoutedInput) GetShortStepName() string

func (*RoutedInput) Poll added in v1.0.0 

func (r *RoutedInput) Poll(ctx context.Context, client *http.Client, token string, inputID string)

func (*RoutedInput) Run added in v1.0.0 

func (r *RoutedInput) Run(ctx context.Context, i modconfig.Input) (*modconfig.Output, error)

func (*RoutedInput) ValidateInput added in v1.0.0 

func (r *RoutedInput) ValidateInput(ctx context.Context, i modconfig.Input) error

type RoutedInputCreatePayload added in v1.0.0 

type RoutedInputCreatePayload struct {
	// TODO: #refactor - can we use a shared struct with pipes for this?
	ExecutionID         string                         `json:"execution_id"`
	PipelineExecutionID string                         `json:"pipeline_execution_id"`
	StepExecutionID     string                         `json:"step_execution_id"`
	NotifierName        string                         `json:"notifier_name"`
	Inputs              map[string]RoutedInputFormData `json:"inputs"`
}

func NewRoutedInputHttpPayload added in v1.0.0 

func NewRoutedInputHttpPayload(executionID, pipelineExecutionID, stepExecutionID, notifierName string, inputs map[string]RoutedInputFormData) *RoutedInputCreatePayload

type RoutedInputEndStepFunc added in v1.0.0 

type RoutedInputEndStepFunc func(stepExecution *execution.StepExecution, out *modconfig.Output) error

RoutedInputEndStepFunc is a function that ends a step

type RoutedInputFormData added in v1.0.0 

type RoutedInputFormData struct {
	// TODO: #refactor - can we use a shared struct with pipes for this?
	Prompt    string                           `json:"prompt"`
	InputType string                           `json:"input_type"`
	Options   []InputIntegrationResponseOption `json:"options,omitempty"`
	Response  any                              `json:"response,omitempty"`
}

func NewRoutedInputHttpPayloadInput added in v1.0.0 

func NewRoutedInputHttpPayloadInput(prompt, inputType string, options []InputIntegrationResponseOption) *RoutedInputFormData

type RoutedInputResponse added in v1.0.0 

type RoutedInputResponse struct {
	// TODO: #refactor - can we use a shared struct with pipes for this?
	ID              string                         `json:"id"`
	TenantID        string                         `json:"tenant_id"`
	IdentityID      string                         `json:"identity_id"`
	WorkspaceID     string                         `json:"workspace_id"`
	NotifierID      string                         `json:"notifier_id"`
	Notifier        map[string]any                 `json:"notifier"`
	ProcessID       string                         `json:"process_id"`
	StepExecutionID string                         `json:"step_execution_id"`
	RandomID        string                         `json:"random_id"`
	State           string                         `json:"state"`
	StateReason     string                         `json:"state_reason"`
	Inputs          map[string]RoutedInputFormData `json:"inputs"`
}

type RowReader added in v0.2.1 

type RowReader interface {
	Read(*sql.Rows, map[string]*sql.ColumnType) ([]map[string]interface{}, error)
	RowToCty(row map[string]interface{}, columnTypes map[string]*sql.ColumnType) (cty.Value, error)
}

type RowReaderImpl added in v0.2.1 

type RowReaderImpl struct {
}

func (*RowReaderImpl) Read added in v0.2.1 

func (r *RowReaderImpl) Read(rows *sql.Rows, columnTypeMap map[string]*sql.ColumnType) ([]map[string]interface{}, error)

func (*RowReaderImpl) RowToCty added in v0.2.1 

func (r *RowReaderImpl) RowToCty(row map[string]interface{}, columnTypes map[string]*sql.ColumnType) (cty.Value, error)

Attempt to have a generic function to convert a row to cty. It may not work for all the database that Flowpipe will support, the types are structured so it can be extended to various different DB Row Readers

type RunPipeline 

type RunPipeline struct{}

func (*RunPipeline) Run 

func (e *RunPipeline) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*RunPipeline) ValidateInput 

func (e *RunPipeline) ValidateInput(ctx context.Context, input modconfig.Input) error

type SQLiteQueryReader added in v0.2.1 

type SQLiteQueryReader struct {
	QueryReaderImpl
}

func (*SQLiteQueryReader) Initialize added in v0.2.1 

func (s *SQLiteQueryReader) Initialize() error

type Sleep 

type Sleep struct{}

func (*Sleep) Run 

func (e *Sleep) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Sleep) ValidateInput 

func (e *Sleep) ValidateInput(ctx context.Context, input modconfig.Input) error

type Transform 

type Transform struct{}

func (*Transform) Run 

func (e *Transform) Run(ctx context.Context, input modconfig.Input) (*modconfig.Output, error)

func (*Transform) ValidateInput 

func (e *Transform) ValidateInput(ctx context.Context, i modconfig.Input) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.