mihaitodor
commented
1 year ago Hey @loicalleyne, that sounds useful! I took a stab at using https://github.com/simpleforce/simpleforce for an internal demo, but I don't have much experience with Salesforce, so I didn't spend much time refining it. I'm also not sure what kind of SOQL queries people would like to run, so I hope you have some more familiarity with this to test it. Here's the demo code I put together:
package input
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/url"
"github.com/benthosdev/benthos/v4/public/service"
"github.com/simpleforce/simpleforce"
)
var salesforceInputConfigSpec = service.NewConfigSpec().
Summary("Creates an input that reads Salesforce data.").
Field(service.NewStringField("url")).
Field(service.NewStringField("user")).
Field(service.NewStringField("password")).
Field(service.NewStringField("token")).
Field(service.NewStringField("query"))
func newSalesforceInput(conf *service.ParsedConfig) (service.Input, error) {
input := salesforceInput{}
var err error
if input.url, err = conf.FieldString("url"); err != nil {
return nil, err
}
if _, err = url.ParseRequestURI(input.url); err != nil {
return nil, fmt.Errorf("invalid URL %q: %s", input.url, err)
}
if input.user, err = conf.FieldString("user"); err != nil {
return nil, err
}
if input.password, err = conf.FieldString("password"); err != nil {
return nil, err
}
if input.token, err = conf.FieldString("token"); err != nil {
return nil, err
}
if input.query, err = conf.FieldString("query"); err != nil {
return nil, err
}
return service.AutoRetryNacks(&input), nil
}
func init() {
err := service.RegisterInput(
"salesforce_select", salesforceInputConfigSpec,
func(conf *service.ParsedConfig, mgr *service.Resources) (service.Input, error) {
return newSalesforceInput(conf)
})
if err != nil {
panic(err)
}
}
//------------------------------------------------------------------------------
type salesforceInput struct {
url string
user string
password string
token string
query string
client *simpleforce.Client
iterator *simpleforce.QueryResult
currentRecord int
}
func (si *salesforceInput) Connect(ctx context.Context) error {
si.client = simpleforce.NewClient(si.url, simpleforce.DefaultClientID, simpleforce.DefaultAPIVersion)
if si.client == nil {
return errors.New("failed to create client")
}
err := si.client.LoginPassword(si.user, si.password, si.token)
if err != nil {
return fmt.Errorf("failed to login: %s", err)
}
si.iterator, err = si.client.Query(si.query) // Note: for Tooling API, use client.Tooling().Query(q)
if err != nil {
return fmt.Errorf("failed to run SOQL query: %s", err)
}
si.currentRecord = 0
return nil
}
func (si *salesforceInput) Read(ctx context.Context) (*service.Message, service.AckFunc, error) {
if si.currentRecord >= len(si.iterator.Records) {
return nil, nil, service.ErrEndOfInput
}
record := si.iterator.Records[si.currentRecord]
b, err := json.Marshal(record)
if err != nil {
return nil, nil, fmt.Errorf("failed to marshal record: %s", err)
}
si.currentRecord++
return service.NewMessage(b), func(ctx context.Context, err error) error {
// Nacks are retried automatically when we use service.AutoRetryNacks
return nil
}, nil
}
func (si *salesforceInput) Close(ctx context.Context) error {
return nil
}Note that the above code is quite crude and one might want to add an args_mapping field similar to the sql_* components, such as https://www.benthos.dev/docs/components/inputs/sql_select#args_mapping. I guess it's also missing pagination...
Also, this simpleforce library doesn't do certain things properly, like, for example, it emits logs using the log package from the standard library, which I think can't be supressed.
Finally, it would be nice to have a few integration tests for it. This Salesforce API is HTTP-based (not sure if they have any alternative ones), so maybe the https://pkg.go.dev/net/http/httptest package can help.
loicalleyne
I'd like to suggest adding support for Salesforce Object Query Language (SOQL) as an input/output. I'm willing to take a stab at implementing this if I get the spare time to do it.