feat - allow support to send event into data source.

Author: nicola-strappazzon

README.md

@@ -10,6 +10,7 @@ A [Tinybird](https://www.tinybird.co/) module for Go. Why need this module? It p
 - Test your code with mocks.
 - Allow JSON, [NDJSON](http://ndjson.org/) and CSV between tinybird and this module.
 - Parallelize HTTP requests.
+- Send [events](https://www.tinybird.co/docs/forward/get-data-in/events-api) to data source.
 - Add custom logic `after` and `before` execute request. For example a cache system.
 - Shared logger with [logrus](https://github.com/sirupsen/logrus).
 

event.go

@@ -0,0 +1,23 @@
+package tinybird
+
+import (
+	"fmt"
+)
+
+type Event struct {
+	Name       string
+	Datasource string
+	Workspace  Workspace
+}
+
+func (e Event) GetURL() string {
+	return URL_BASE
+}
+
+func (e Event) GetURI() string {
+	return fmt.Sprintf(
+		"%s/v0/events?name=%s",
+		e.GetURL(),
+		e.Datasource,
+	)
+}

example/events/README.md

@@ -0,0 +1,10 @@
+# Example - Send events to data source.
+
+This example demonstrates how to use append data on data source into Tinybird.
+
+To run this example:
+
+```shell
+export TB_TOKEN="<your-api-token>"
+go run .
+```

example/events/events

@@ -0,0 +1,12 @@
+module github.com/the-hotels-network/go-tinybird/example/events
+
+go 1.19
+
+replace github.com/the-hotels-network/go-tinybird => ../..
+
+require (
+	github.com/olivere/ndjson v1.0.1 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/the-hotels-network/go-tinybird v1.3.0 // indirect
+	golang.org/x/sys v0.9.0 // indirect
+)

example/events/go.mod

@@ -0,0 +1,24 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
+github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/olivere/ndjson v1.0.1 h1:q+rEa/MOpElAGj7W4IHmpY6VG7baHAugfs0MGx8DNA8=
+github.com/olivere/ndjson v1.0.1/go.mod h1:y3NXfLEBYQd+QGVQxzIFYnbRhdVKVBaLSCGAoyialk4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
+github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
+golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

example/events/go.sum

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+
+	"github.com/the-hotels-network/go-tinybird"
+)
+
+func main() {
+	req := tinybird.Request{
+		Method: http.MethodGet,
+		Event: tinybird.Event{
+			Datasource: "ep_quantum_disparities",
+			Workspace: tinybird.Workspace{
+				Token: os.Getenv("TB_TOKEN"),
+			},
+		},
+	}
+
+	err := req.Execute()
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+
+	// res := req.Response
+	// fmt.Println("Status code:", res.Status)
+	// fmt.Println("Elapsed time:", req.Elapsed)
+	// fmt.Println("Error:", res.Error)
+	// fmt.Println("Data:", res.Data)
+	// fmt.Println("URL:", req.URI())
+}

example/events/main.go

@@ -1,8 +1,18 @@
 package tinybird
 
 import (
+	"fmt"
 	"net/url"
 	"strings"
+
+	"github.com/the-hotels-network/go-tinybird/internal/env"
+)
+
+const (
+	Format string = "json"
+	JSON          = "json"
+	NDJSON        = "ndjson"
+	CSV           = "csv"
 )
 
 // Pipe is a object on tinybird, contains one or more SQL queries (Nodes) that
@@ -19,6 +29,8 @@ type Pipe struct {
 	// Is an area that contains a set of Tinybird resources, including Pipes,
 	// Nodes, APIs, Data Sources & Auth Tokens. Define by user.
 	Workspace Workspace
+	// Return format.
+	Format string
 }
 
 func (p *Pipe) GetParameters() string {
@@ -31,3 +43,65 @@ func (p *Pipe) GetParameters() string {
 
 	return strings.Replace(p.Parameters.Encode(), "+", "%20", -1)
 }
+
+// Build and return the pipe URL.
+func (p Pipe) GetURL() string {
+	baseUrl := URL_BASE
+
+	if p.URL != "" {
+		baseUrl = p.URL
+	}
+
+	return baseUrl
+
+	// if len(p.URL) > 0 {
+	// 	return p.URL
+	// 	// return fmt.Sprintf(
+	// 	// 	"%s/pipes/%s.%s",
+	// 	// 	p.URL,
+	// 	// 	p.Name,
+	// 	// 	p.GetFormat(),
+	// 	// )
+	// }
+
+	// return fmt.Sprintf(
+	// 	"%s/pipes/%s.%s",
+	// 	baseUrl,
+	// 	p.Name,
+	// 	p.GetFormat(),
+	// )
+}
+
+func (p Pipe) GetURI() string {
+	qs, _ := url.QueryUnescape(p.GetParameters())
+
+	if qs == "" {
+		return fmt.Sprintf(
+			"%s/v0/pipes/%s.%s",
+			p.GetURL(),
+			p.Name,
+			p.GetFormat(),
+		)
+	}
+
+	return fmt.Sprintf(
+		"%s/v0/pipes/%s.%s?%s",
+		p.GetURL(),
+		p.Name,
+		p.GetFormat(),
+		qs,
+	)
+}
+
+// Verify the variable Format value to return json, ndjson or csv.
+func (p Pipe) GetFormat() string {
+	if p.Format == JSON || p.Format == NDJSON || p.Format == CSV {
+		return p.Format
+	}
+
+	if env.GetBool("TB_NDJSON", false) {
+		return NDJSON
+	}
+
+	return JSON
+}

pipe.go

@@ -9,6 +9,14 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestGetParametersEmpty(t *testing.T) {
+	p := tinybird.Pipe{
+		Parameters: url.Values{},
+	}
+
+	assert.Equal(t, p.GetParameters(), "")
+}
+
 func TestGetParameters(t *testing.T) {
 	params := url.Values{}
 	params.Add("start_date", "2022-05-01")
@@ -30,3 +38,7 @@ func TestGetParameters(t *testing.T) {
 
 	assert.Equal(t, p.GetParameters(), "end_date=2022-05-30&id=1%2C2%2C3%2C4&request_at=2023-05-01%2023%3A59%3A59&start_date=2022-05-01")
 }
+
+func TestGetURI(t *testing.T) {
+
+}

pipe_test.go

@@ -1,33 +1,24 @@
 package tinybird
 
 import (
-	"fmt"
+	"bytes"
 	"net/http"
-	"net/url"
 	"time"
 
-	"github.com/the-hotels-network/go-tinybird/internal/env"
-
 	log "github.com/sirupsen/logrus"
 )
 
-const (
-	Format string = "json"
-	JSON          = "json"
-	NDJSON        = "ndjson"
-	CSV           = "csv"
-)
-
 // Basic request struct.
 type Request struct {
+	After    func(*Request)      // Run after execute request.
+	Before   func(*Request) bool // Run before execute request.
+	Data     []byte              // Data to send.
 	Elapsed  Duration            // Elapsed time of client request.
 	Error    error               // Error on client request.
+	Event    Event               // Send event to data source.
 	Method   string              // Define HTTP method.
 	Pipe     Pipe                // Pipe details.
 	Response Response            // Response data.
-	Format   string              // Return format.
-	Before   func(*Request) bool // Run before execute request.
-	After    func(*Request)      // Run after execute request.
 }
 
 // Custom HTTP client for this module.
@@ -83,7 +74,7 @@ func (r *Request) Execute() error {
 
 // Create new request.
 func (r *Request) newRequest() (*http.Request, error) {
-	req, err := http.NewRequest(r.Method, r.URL(), nil)
+	req, err := http.NewRequest(r.Method, r.URI(), bytes.NewBuffer(r.Data))
 	if err != nil {
 		return nil, err
 	}
@@ -94,8 +85,11 @@ func (r *Request) newRequest() (*http.Request, error) {
 	}).Debug("tinybird")
 
 	req.Header.Add("Accept", "application/json")
-	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", r.Pipe.Workspace.Token))
-	req.URL.RawQuery = r.Pipe.GetParameters()
+	req.Header.Add("Authorization", r.Token())
+
+	if r.Pipe.Workspace.IsSet() {
+		req.URL.RawQuery = r.Pipe.GetParameters()
+	}
 
 	return req, nil
 }
@@ -104,7 +98,7 @@ func (r *Request) newRequest() (*http.Request, error) {
 func (r *Request) readBody(resp *http.Response) (err error) {
 	defer resp.Body.Close()
 
-	r.Response.Format = r.Format
+	r.Response.Format = r.Format()
 	r.Response.Status = resp.StatusCode
 	r.Response.Header = resp.Header
 	r.Response.Raw = resp.Body
@@ -119,38 +113,40 @@ func (r *Request) readBody(resp *http.Response) (err error) {
 	return err
 }
 
-// Build and return the pipe URL.
-func (r *Request) URL() string {
-	var baseUrl string
-	if r.Pipe.URL != "" {
-		baseUrl = r.Pipe.URL
-	} else {
-		baseUrl = URL_BASE
+func (r *Request) URL() (out string) {
+	if r.Pipe.Workspace.IsSet() {
+		out = r.Pipe.GetURL()
+	} else if r.Event.Workspace.IsSet() {
+		out = r.Event.GetURL()
 	}
 
-	return fmt.Sprintf(
-		"%s/%s.%s",
-		baseUrl,
-		r.Pipe.Name,
-		r.GetFormat(),
-	)
+	return out
 }
 
-// Verify the variable Format value to return json, ndjson or csv.
-func (r *Request) GetFormat() string {
-	if r.Format == JSON || r.Format == NDJSON || r.Format == CSV {
-		return r.Format
+func (r *Request) Token() (out string) {
+	if r.Pipe.Workspace.IsSet() {
+		out = r.Pipe.Workspace.GetToken()
+	} else if r.Event.Workspace.IsSet() {
+		out = r.Event.Workspace.GetToken()
 	}
 
-	if env.GetBool("TB_NDJSON", false) {
-		return NDJSON
+	return out
+}
+
+func (r *Request) URI() (out string) {
+	if r.Pipe.Workspace.IsSet() {
+		out = r.Pipe.GetURI()
+	} else if r.Event.Workspace.IsSet() {
+		out = r.Event.GetURI()
 	}
 
-	return JSON
+	return out
 }
 
-// Return concatened URL and Query String to generate a URI.
-func (r *Request) URI() string {
-	qs, _ := url.QueryUnescape(r.Pipe.GetParameters())
-	return fmt.Sprintf("%s?%s", r.URL(), qs)
+func (r *Request) Format() (out string) {
+	if r.Pipe.Workspace.IsSet() {
+		out = r.Pipe.GetFormat()
+	}
+
+	return out
 }