Documentation
¶
Index ¶
Examples ¶
Constants ¶
const (
// MysqlDatetimeFormat is the Mysql date time format.
MysqlDatetimeFormat = "2006-01-02 15:04:05"
)
Variables ¶
var ErrInvalidSQLQueryType = errors.New("invalid SQL query type")
ErrInvalidSQLQueryType is the error for when the query passed in is isn't one of the know types.
var ErrStopExtract = errors.New("stop iterating over the *sqlx.Rows")
ErrStopExtract is the error to be used by a rowHandler to indicate that the iterating over the rows results should not continue.
Functions ¶
func ExtractRecordsFromRows ¶
func ExtractRecordsFromRows(rows *sqlx.Rows, rowHandler func(message.OrderedRecord, error) error) error
ExtractRecordsFromRows iterates over the *sqlx.Rows and calls the handler for each or error if something went wrong. The row handler can return it's own error in which case the iterating still stop. If the error is not the ErrStopExtract error, then it will be pass on through as the error of the overall func call. Keep in mind you will need to make sure the rows passed in are close elsewhere. (Ex: rows.Close() ) This func will not close the rows.
Types ¶
type Conn ¶
type Conn struct {
DSN string // connection string (required)
Driver string // used to make a new connection if DB is nil
DB *sqlx.DB
// contains filtered or unexported fields
}
Conn is a wrapper around a connection to handle creating the connection and shutting it down. This allows you to just set a DSN and this will handle connecting to it.
Example ¶
ExampleConn shows how to create a database connection that is used by the Query and Exec transformers. Conn is rarely used directly on it's own, but it wrapped by Exec and Query and used by them.
package main
import (
"github.com/Reisender/pipe/extras/sql"
)
func main() {
c := sql.Conn{Driver: "postgres", DSN: "postgres://user:pass@localhost/dbname?sslmode=verify-full"}
err := c.Open()
if err != nil {
panic(err)
}
defer c.Close()
// now the connection is ready to use
}
Output:
type Exec ¶
type Exec Conn
Exec executes a query
Example ¶
package main
import (
"github.com/Reisender/pipe/extras/sql"
"github.com/Reisender/pipe/line"
"github.com/Reisender/pipe/message"
)
func main() {
line.New().SetP(func(out chan<- interface{}, errs chan<- error) {
// using a string directly as the query
out <- "INSERT INTO foo (name) VALUES ('bar')"
// using message.Query
out <- message.Query{SQL: "INSERT INTO foo (name) VALUES ('bar')"}
// using message.Query with args (using sqlx under the hood for arg matching)
query := message.Query{SQL: "INSERT INTO foo (name) VALUES (?)"}
query.Args = append(query.Args, "bar")
out <- query
// using delta messages
record := message.NewRecordFromMSI(map[string]interface{}{"name": "bar"})
delta := message.InsertDelta{Table: "foo", Record: record}
out <- delta
}).Add(
sql.Exec{Driver: "postgres", DSN: "postgres://user:pass@localhost/dbname?sslmode=verify-full"}.T,
).Run()
}
Output:
type Get ¶
type Get struct {
Conn
SQL string // (required)
Table string
PageSize int
OrderBy string
BodyCol string // the column to put as the body of the message (blank is all as json)
}
Get gets records from a query or table
func (Get) T ¶
T will take in records and use them in a sql query. The input can be a single record or a batch of records. The predefined template functions can be used to extract individual keys from the metadata record, or the message body as a single string value. Or if this is a batch the keys can be used to extract one columns data as a comma seperated list e.g. for use in an IN clause.
The output by default will send one resulting record in one out message. Otherwise this can be configured to send all resulting records from one
type Query ¶
type Query Conn
Query runs an SQL query on a db connection
Example ¶
package main
import (
"github.com/Reisender/pipe/extras/sql"
"github.com/Reisender/pipe/line"
"github.com/Reisender/pipe/message"
_ "github.com/proullon/ramsql/driver"
)
func main() {
line.New().SetP(func(out chan<- interface{}, errs chan<- error) {
// using a string directly as the query
out <- "SELECT * FROM foo"
// using message.Query
out <- message.Query{SQL: "SELECT * FROM foo"}
// using message.Query with args (using sqlx under the hood for arg matching)
query := message.Query{SQL: "SELECT * FROM foo WHERE name=? AND email=?"}
query.Args = append(query.Args, "bar", "bar@foo.com")
out <- query
}).Add(
sql.Query{Driver: "postgres", DSN: "postgres://user:pass@localhost/dbname?sslmode=verify-full"}.T,
).Run()
}
Output:
Source Files