type APIConnection struct {NamestringSpecSpecState   *APIStateContext *g.Context// contains filtered or unexported fields}

funcNewAPIConnection¶

func NewAPIConnection(ctxcontext.Context, specSpec, data map[string]any) (ac *APIConnection, errerror)

NewAPIConnection creates an

func (*APIConnection)Authenticate¶

func (ac *APIConnection) Authenticate() (errerror)

Authenticate performs the auth workflow if needed. Like a Connect step.Header based auths (such as Basic, or Bearer) don't need this stepsave payload in APIState.Auth

func (*APIConnection)Close¶

func (ac *APIConnection) Close()error

Close performs cleanup of all resources

func (*APIConnection)CloseAllQueues¶

func (ac *APIConnection) CloseAllQueues()

CloseAllQueues closes all queues associated with this connection

func (*APIConnection)CompileEndpoints¶added inv1.4.25

func (ac *APIConnection) CompileEndpoints() (compiledEndpointsEndpoints, errerror)

compile all spec endpoints

func (*APIConnection)EnsureAuthenticated¶added inv1.4.14

func (ac *APIConnection) EnsureAuthenticated()error

EnsureAuthenticated checks if authentication is valid and re-authenticates if neededThis method ensures thread-safe authentication checks and re-authentication

func (*APIConnection)GetQueue¶

func (ac *APIConnection) GetQueue(namestring) (*iop.Queue,bool)

GetQueue retrieves a queue by name

func (*APIConnection)GetReplicationStore¶added inv1.4.20

func (ac *APIConnection) GetReplicationStore() (store map[string]any)

func (*APIConnection)GetSyncedState¶

func (ac *APIConnection) GetSyncedState(endpointNamestring) (data map[string]any, errerror)

GetSyncedState cycles through each endpoint, and collects the valuesfor each of the Endpoint.Sync values. Output is amap[Sync.value] = Endpoint.syncMap[Sync.value]

func (*APIConnection)IsAuthExpired¶added inv1.4.14

func (ac *APIConnection) IsAuthExpired()bool

IsAuthExpired checks if the authentication has expired

func (*APIConnection)ListEndpoints¶

func (ac *APIConnection) ListEndpoints(patterns ...string) (endpointsEndpoints, errerror)

func (*APIConnection)MakeAuthenticator¶added inv1.4.25

func (ac *APIConnection) MakeAuthenticator() (authenticatorAuthenticator, errerror)

func (*APIConnection)PutSyncedState¶

func (ac *APIConnection) PutSyncedState(endpointNamestring, data map[string]any) (errerror)

PutSyncedState restores the state from previous run in each endpointusing the Endpoint.Sync values.Inputs is map[Sync.value] = Endpoint.syncMap[Sync.value]

func (*APIConnection)ReadDataflow¶

func (ac *APIConnection) ReadDataflow(endpointNamestring, sCfgAPIStreamConfig) (df *iop.Dataflow, errerror)

func (*APIConnection)RegisterQueue¶

func (ac *APIConnection) RegisterQueue(namestring) (*iop.Queue,error)

RegisterQueue creates a new queue with the given nameIf a queue with the same name already exists, it is returned

func (*APIConnection)RemoveQueue¶

func (ac *APIConnection) RemoveQueue(namestring)error

RemoveQueue closes and removes a queue

func (*APIConnection)RenderDynamicEndpoints¶added inv1.4.14

func (ac *APIConnection) RenderDynamicEndpoints() (errerror)

RenderDynamicEndpoints will render the dynamic objectsbasically mutating the spec endpoints.Needs to authenticate first

func (*APIConnection)SetReplicationStore¶added inv1.4.20

func (ac *APIConnection) SetReplicationStore(store map[string]any)

type APIState struct {Env     map[string]string     `json:"env,omitempty"`State   map[string]any        `json:"state,omitempty"`Store   map[string]any        `json:"store,omitempty"`Inputs  map[string]any        `json:"inputs,omitempty"`Secrets map[string]any        `json:"secrets,omitempty"`Queues  map[string]*iop.Queue `json:"queues,omitempty"`// appends to fileAuthAPIStateAuth          `json:"auth,omitempty"`}

type APIStateAuth struct {Authenticatedbool              `json:"authenticated,omitempty"`Tokenstring            `json:"token,omitempty"`// refresh token?Headers       map[string]string `json:"-"`// to injectExpiresAtint64             `json:"expires_at,omitempty"`// Unix timestamp when auth expiresSign  func(*Iteration, *http.Request, []byte)error `json:"-"`// for AWS Sigv4, HMACMutex *sync.Mutex                                   `json:"-" yaml:"-"`// Mutex for auth operations}

type APIStreamConfig struct {Flattenint// levels of flattening. 0 is infiniteJmesPathstringSelect      []string// select specific columnsLimitintMetadataiop.MetadataModestringRangestringDsConfigMap map[string]any// stream processor options}

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

funcNewAggregateState¶added inv1.5.1

func NewAggregateState()AggregateState

type AggregationTypestring

type AuthTypestring

func (AuthType)String¶added inv1.4.25

func (atAuthType) String()string

type Authentication map[string]any

func (Authentication)Expires¶added inv1.4.14

func (aAuthentication) Expires()int

func (Authentication)Type¶

func (aAuthentication) Type()AuthType

type Authenticator interface {AuthType()AuthTypeAuthenticate(ctxcontext.Context, state *APIStateAuth)errorRefresh(ctxcontext.Context, state *APIStateAuth)error}

type AuthenticatorAWSSigV4 struct {AuthenticatorBaseAwsServicestring `yaml:"aws_service,omitempty" json:"aws_service,omitempty"`AwsAccessKeyIDstring `yaml:"aws_access_key_id,omitempty" json:"aws_access_key_id,omitempty"`AwsSecretAccessKeystring `yaml:"aws_secret_access_key,omitempty" json:"aws_secret_access_key,omitempty"`AwsSessionTokenstring `yaml:"aws_session_token,omitempty" json:"aws_session_token,omitempty"`AwsRegionstring `yaml:"aws_region,omitempty" json:"aws_region,omitempty"`AwsProfilestring `yaml:"aws_profile,omitempty" json:"aws_profile,omitempty"`}

func (*AuthenticatorAWSSigV4)Authenticate¶added inv1.4.25

func (a *AuthenticatorAWSSigV4) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

type AuthenticatorBase struct {TypeAuthType `yaml:"type" json:"type"`Expiresint      `yaml:"expires" json:"expires,omitempty"`// when set, re-auth after number of seconds// contains filtered or unexported fields}

func (*AuthenticatorBase)AuthType¶added inv1.4.25

func (a *AuthenticatorBase) AuthType()AuthType

func (*AuthenticatorBase)Authenticate¶added inv1.4.25

func (a *AuthenticatorBase) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

func (*AuthenticatorBase)Refresh¶added inv1.4.25

func (a *AuthenticatorBase) Refresh(ctxcontext.Context, state *APIStateAuth) (errerror)

type AuthenticatorBasic struct {AuthenticatorBaseUsernamestring `yaml:"username,omitempty" json:"username,omitempty"`Passwordstring `yaml:"password,omitempty" json:"password,omitempty"`}

func (*AuthenticatorBasic)Authenticate¶added inv1.4.25

func (a *AuthenticatorBasic) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

type AuthenticatorHMAC struct {AuthenticatorBaseAlgorithmstring            `yaml:"algorithm" json:"algorithm"`Secretstring            `yaml:"secret" json:"secret"`SigningStringstring            `yaml:"signing_string" json:"signing_string"`ReqHeaders    map[string]string `yaml:"request_headers" json:"request_headers"`NonceLengthint               `yaml:"nonce_length" json:"nonce_length,omitempty"`// Length of random nonce bytes (0 to disable)}

func (*AuthenticatorHMAC)Authenticate¶added inv1.4.25

func (a *AuthenticatorHMAC) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

type AuthenticatorOAuth2 struct {AuthenticatorBaseClientIDstring    `yaml:"client_id,omitempty" json:"client_id,omitempty"`ClientSecretstring    `yaml:"client_secret,omitempty" json:"client_secret,omitempty"`RedirectURIstring    `yaml:"redirect_uri,omitempty" json:"redirect_uri,omitempty"`AuthenticationURLstring    `yaml:"authentication_url,omitempty" json:"authentication_url,omitempty"`// Token endpointAuthorizationURLstring    `yaml:"authorization_url,omitempty" json:"authorization_url,omitempty"`// Auth endpointDeviceAuthURLstring    `yaml:"device_auth_url,omitempty" json:"device_auth_url,omitempty"`// Device auth endpoint (optional)Scopes            []string  `yaml:"scopes,omitempty" json:"scopes,omitempty"`FlowOAuthFlow `yaml:"flow,omitempty" json:"flow,omitempty"`// e.g., "authorization_code", "client_credentials", "device_code"}

func (*AuthenticatorOAuth2)Authenticate¶added inv1.4.25

func (a *AuthenticatorOAuth2) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

Authenticate performs OAuth2 authentication based on the configured flow.

func (*AuthenticatorOAuth2)LoadToken¶added inv1.4.25

func (a *AuthenticatorOAuth2) LoadToken() (*oauth2.Token,error)

LoadToken loads the stored token.

func (*AuthenticatorOAuth2)SaveToken¶added inv1.4.25

func (a *AuthenticatorOAuth2) SaveToken(tok *oauth2.Token)error

SaveToken securely saves the token (use keyring for better security).

func (*AuthenticatorOAuth2)TokenFile¶added inv1.4.25

func (a *AuthenticatorOAuth2) TokenFile()string

TokenFile returns a secure path for token storage.

type AuthenticatorSequence struct {AuthenticatorBaseSequenceSequence `yaml:"sequence" json:"sequence,omitempty"`}

func (*AuthenticatorSequence)Authenticate¶added inv1.4.25

func (a *AuthenticatorSequence) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

type AuthenticatorStatic struct {AuthenticatorBaseHeaders map[string]string `yaml:"headers,omitempty" json:"headers,omitempty"`}

func (*AuthenticatorStatic)Authenticate¶added inv1.5.1

func (a *AuthenticatorStatic) Authenticate(ctxcontext.Context, state *APIStateAuth) (errerror)

type BackoffTypestring

type Call struct {Ifstring          `yaml:"if" json:"if"`RequestRequest         `yaml:"request" json:"request"`PaginationPagination      `yaml:"pagination" json:"pagination"`ResponseResponse        `yaml:"response" json:"response"`Authentication *Authentication `yaml:"authentication,omitempty" json:"authentication,omitempty"`Iteratestring          `yaml:"iterate" json:"iterate"`Intostring          `yaml:"into" json:"into"`}

type DynamicEndpoint struct {SetupSequence `yaml:"setup" json:"setup"`Iteratestring   `yaml:"iterate" json:"iterate"`Intostring   `yaml:"into" json:"into"`EndpointEndpoint `yaml:"endpoint" json:"endpoint"`}

type DynamicEndpoints []DynamicEndpoint

type Endpoint struct {Namestring     `yaml:"name" json:"name"`Descriptionstring     `yaml:"description" json:"description,omitempty"`Docsstring     `yaml:"docs" json:"docs,omitempty"`Disabledbool       `yaml:"disabled" json:"disabled"`Dynamicbool       `yaml:"dynamic,omitempty" json:"dynamic,omitempty"`// is generatedStateStateMap   `yaml:"state" json:"state"`Sync        []string   `yaml:"sync" json:"sync,omitempty"`RequestRequest    `yaml:"request" json:"request"`PaginationPagination `yaml:"pagination" json:"pagination"`ResponseResponse   `yaml:"response" json:"response"`IterateIterate    `yaml:"iterate" json:"iterate,omitempty"`// state expression to use to loopSetupSequence   `yaml:"setup" json:"setup,omitempty"`TeardownSequence   `yaml:"teardown" json:"teardown,omitempty"`DependsOn   []string   `yaml:"depends_on" json:"depends_on"`// upstream endpointsOverridesany        `yaml:"overrides" json:"overrides"`// stream overridesAuthentication *Authentication `yaml:"authentication,omitempty" json:"authentication,omitempty"`// contains filtered or unexported fields}

func (*Endpoint)Authenticate¶added inv1.5.1

func (ep *Endpoint) Authenticate() (errerror)

Authenticate performs the auth workflow if needed. Like a Connect step.Header based auths (such as Basic, or Bearer) don't need this stepsave payload in APIState.Auth

func (*Endpoint)EnsureAuthenticated¶added inv1.5.1

func (ep *Endpoint) EnsureAuthenticated()error

func (*Endpoint)Evaluate¶added inv1.4.25

func (ep *Endpoint) Evaluate(exprstring, state map[string]interface{}) (resultany, errerror)

func (*Endpoint)IsAuthExpired¶added inv1.5.1

func (ep *Endpoint) IsAuthExpired()bool

IsAuthExpired checks if the authentication has expired

func (*Endpoint)MakeAuthenticator¶added inv1.5.1

func (ep *Endpoint) MakeAuthenticator() (authenticatorAuthenticator, errerror)

func (*Endpoint)SetStateVal¶

func (ep *Endpoint) SetStateVal(keystring, valany)

func (*Endpoint)UnmarshalYAML¶added inv1.5.1

func (ep *Endpoint) UnmarshalYAML(unmarshal func(interface{})error)error

UnmarshalYAML handles +setup/setup+ and +teardown/teardown+ modifier syntax for Endpoint

type EndpointMap map[string]Endpoint

type Endpoints []Endpoint

func (Endpoints)DAG¶added inv1.4.23

func (epsEndpoints) DAG() [][]string

DAG returns groups of interdependent endpoints

func (Endpoints)HasUpstreams¶added inv1.4.23

func (epsEndpoints) HasUpstreams(endpointNamestring) (upstreams []string)

HasUpstreams returns all upstream endpoint names that the specified endpoint depends on.Returns empty slice if the endpoint doesn't exist or has no dependencies.

func (Endpoints)Names¶added inv1.4.23

func (epsEndpoints) Names() (names []string)

Names returns names in alphabetical order

func (Endpoints)Sort¶

func (epsEndpoints) Sort()

type HTTPMethodstring

type Iterate struct {Overany    `yaml:"over" json:"iterate,omitempty"`// expressionIntostring `yaml:"into" json:"into,omitempty"`// state variableConcurrencyint    `yaml:"concurrency" json:"concurrency,omitempty"`// contains filtered or unexported fields}

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

func (*Iteration)Debug¶added inv1.4.20

func (iter *Iteration) Debug(textstring, args ...any)

func (*Iteration)DetermineStateRenderOrder¶added inv1.4.14

func (iter *Iteration) DetermineStateRenderOrder() (order []string, errerror)

type OAuthFlowstring

type Pagination struct {NextState     map[string]any `yaml:"next_state" json:"next_state,omitempty"`StopConditionstring         `yaml:"stop_condition" json:"stop_condition,omitempty"`}

type Processor struct {AggregationAggregationType `yaml:"aggregation" json:"aggregation"`Expressionstring          `yaml:"expression" json:"expression"`Outputstring          `yaml:"output" json:"output"`Ifstring          `yaml:"if" json:"if,omitempty"`// if we should evaluate processor}

type Records struct {JmesPathstring   `yaml:"jmespath" json:"jmespath,omitempty"`// for json or xmlPrimaryKey []string `yaml:"primary_key" json:"primary_key,omitempty"`UpdateKeystring   `yaml:"update_key" json:"update_key,omitempty"`Limitint      `yaml:"limit" json:"limit,omitempty"`// to limit the records, useful for testingCasingstring   `yaml:"casing" json:"casing,omitempty"`// "snake" or "camel"Select     []string `yaml:"select" json:"select,omitempty"`// include/exclude/renameDuplicateTolerancestring `yaml:"duplicate_tolerance" json:"duplicate_tolerance,omitempty"`}

type Request struct {URLstring         `yaml:"url" json:"url,omitempty"`Timeoutint            `yaml:"timeout" json:"timeout,omitempty"`MethodHTTPMethod     `yaml:"method" json:"method,omitempty"`Headers     map[string]any `yaml:"headers" json:"headers,omitempty"`Parameters  map[string]any `yaml:"parameters" json:"parameters,omitempty"`Payloadany            `yaml:"payload" json:"payload,omitempty"`Ratefloat64        `yaml:"rate" json:"rate,omitempty"`// maximum request per secondConcurrencyint            `yaml:"concurrency" json:"concurrency,omitempty"`// maximum concurrent requests}

type RequestState struct {Methodstring         `yaml:"method" json:"method"`URLstring         `yaml:"url" json:"url"`Headers  map[string]any `yaml:"headers" json:"headers"`Payloadany            `yaml:"payload" json:"payload"`Attemptsint            `yaml:"attempts" json:"attempts"`}

type Response struct {Formatdbio.FileType `yaml:"format" json:"format,omitempty"`// force response formatRecordsRecords       `yaml:"records" json:"records"`Processors []Processor   `yaml:"processors" json:"processors,omitempty"`Rules      []Rule        `yaml:"rules" json:"rules,omitempty"`// Modifier fields for prepending/appending to defaults (populated by custom unmarshal)// Use +rules to prepend rules (evaluated before defaults)// Use rules+ to append rules (evaluated after defaults, before hardcoded)PrependRules      []Rule      `yaml:"-" json:"-"`AppendRules       []Rule      `yaml:"-" json:"-"`PrependProcessors []Processor `yaml:"-" json:"-"`AppendProcessors  []Processor `yaml:"-" json:"-"`}

func (*Response)UnmarshalYAML¶added inv1.5.1

func (r *Response) UnmarshalYAML(unmarshal func(interface{})error)error

UnmarshalYAML handles +rules/rules+ and +processors/processors+ modifier syntax

type ResponseState struct {Statusint              `yaml:"status" json:"status"`Headers map[string]any   `yaml:"headers" json:"headers"`Textstring           `yaml:"text" json:"text"`JSONany              `yaml:"json" json:"json"`Records []map[string]any `yaml:"records" json:"records"`}

type Rule struct {ActionRuleType    `yaml:"action" json:"action"`Conditionstring      `yaml:"condition" json:"condition"`// an expressionMaxAttemptsint         `yaml:"max_attempts" json:"max_attempts"`BackoffBackoffType `yaml:"backoff" json:"backoff"`BackoffBaseint         `yaml:"backoff_base" json:"backoff_base"`// base duration, number of seconds. default is 1Messagestring      `yaml:"message" json:"message"`}

type RuleTypestring

type Sequence []Call

type SingleRequest struct {Request  *RequestState  `yaml:"request" json:"request"`Response *ResponseState `yaml:"response" json:"response"`// contains filtered or unexported fields}

funcNewSingleRequest¶

func NewSingleRequest(iter *Iteration) *SingleRequest

func (*SingleRequest)Debug¶

func (lrs *SingleRequest) Debug(textstring, args ...any)

func (*SingleRequest)Map¶

func (lrs *SingleRequest) Map() map[string]any

func (*SingleRequest)Records¶

func (lrs *SingleRequest) Records() []map[string]any

func (*SingleRequest)Trace¶added inv1.4.6

func (lrs *SingleRequest) Trace(textstring, args ...any)

type Spec struct {Namestring           `yaml:"name" json:"name"`Descriptionstring           `yaml:"description" json:"description"`Queues           []string         `yaml:"queues" json:"queues"`DefaultsEndpoint         `yaml:"defaults" json:"defaults"`AuthenticationAuthentication   `yaml:"authentication" json:"authentication"`EndpointMapEndpointMap      `yaml:"endpoints" json:"endpoints"`DynamicEndpointsDynamicEndpoints `yaml:"dynamic_endpoints" json:"dynamic_endpoints"`// contains filtered or unexported fields}

funcLoadSpec¶

func LoadSpec(specBodystring) (specSpec, errerror)

func (*Spec)IsDynamic¶

func (s *Spec) IsDynamic()bool

func (*Spec)MD5¶added inv1.5.1

func (s *Spec) MD5()string

type StateMap map[string]any