Documentation
¶
Overview ¶
Package database provides deterministic node executors for querying external databases from a dataflow graph — Postgres, MySQL, SQLite (pkg/database/sql, three thin drivers sharing one executor), Redis, and MongoDB (native drivers), and Elasticsearch (its own REST API, no client SDK needed). Each node resolves its DSN/credential through dataflow.Runtime.Secrets rather than taking a raw connection string parameter, so a workflow definition never embeds a plaintext password.
Index ¶
- func NewMySQL() *sqlNode
- func NewPostgres() *sqlNode
- func NewSQLite() *sqlNode
- func Register(registry *dataflow.Registry)
- type Elasticsearch
- func (n *Elasticsearch) Description() dataflow.NodeDescription
- func (n *Elasticsearch) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, ...) (dataflow.Output, error)
- func (n *Elasticsearch) TestConnection(ctx context.Context, rt *dataflow.Runtime, params map[string]any) error
- func (n *Elasticsearch) ValidateParameters(params map[string]any) error
- type MongoDB
- func (MongoDB) Description() dataflow.NodeDescription
- func (MongoDB) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, ...) (dataflow.Output, error)
- func (MongoDB) TestConnection(ctx context.Context, rt *dataflow.Runtime, params map[string]any) error
- func (MongoDB) ValidateParameters(params map[string]any) error
- type Redis
- func (Redis) Description() dataflow.NodeDescription
- func (Redis) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, ...) (dataflow.Output, error)
- func (Redis) TestConnection(ctx context.Context, rt *dataflow.Runtime, params map[string]any) error
- func (Redis) ValidateParameters(params map[string]any) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewMySQL ¶
func NewMySQL() *sqlNode
NewMySQL creates the mysql node, which runs queries against a MySQL database.
func NewPostgres ¶
func NewPostgres() *sqlNode
NewPostgres creates the postgres node, which runs queries against a PostgreSQL database.
Types ¶
type Elasticsearch ¶
type Elasticsearch struct {
// contains filtered or unexported fields
}
Elasticsearch is the "elasticsearch" node type. Elasticsearch's own API is already plain HTTP/JSON, so — matching m9m's own approach here, which is the right call for this one service — this talks to it directly rather than pulling in the official client SDK as a dependency.
Parameters: baseURLSecretRef (required, resolves to e.g. "https://es.internal:9200"), index (required), operation (search/index/get/delete), id (for index/get/delete), document (for index), query (map[string]any, ES query DSL body, for search).
func NewElasticsearch ¶
func NewElasticsearch() *Elasticsearch
NewElasticsearch creates an Elasticsearch node with a 30-second timeout.
func (*Elasticsearch) Description ¶
func (n *Elasticsearch) Description() dataflow.NodeDescription
Description returns the catalog entry of the elasticsearch node: it runs one operation (search, index, get or delete) on an index. search returns one item per hit, the others one item with the raw response. The base URL is a reference to a dataflow secret.
func (*Elasticsearch) Execute ¶
func (n *Elasticsearch) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute resolves the base URL from the secrets of the run, sends the request for the operation and returns the items described by Description. It fails when the run has no secret resolver.
func (*Elasticsearch) TestConnection ¶
func (n *Elasticsearch) TestConnection(ctx context.Context, rt *dataflow.Runtime, params map[string]any) error
TestConnection verifies baseURLSecretRef alone resolves and reaches a real Elasticsearch cluster — it needs none of the other parameters (index/operation/...). A GET on the cluster root is the cheapest request every Elasticsearch deployment answers regardless of index. See dataflow.TestableExecutor.
func (*Elasticsearch) ValidateParameters ¶
func (n *Elasticsearch) ValidateParameters(params map[string]any) error
ValidateParameters requires baseURLSecretRef, an index and a supported operation, and a document id for get and delete.
type MongoDB ¶
type MongoDB struct{}
MongoDB is the "mongodb" node type. Unlike m9m's MongoDB node (which only works against MongoDB Atlas's Data API — a cloud-only REST proxy, not available for a plain self-hosted MongoDB), this uses the official native driver so it works against any MongoDB deployment.
Parameters: uriSecretRef (required), database, collection (required), operation (find/insertOne/updateOne/deleteOne), filter (map[string]any, query for find/updateOne/deleteOne), document (map[string]any, for insertOne), update (map[string]any, for updateOne — wrapped in $set).
func (MongoDB) Description ¶
func (MongoDB) Description() dataflow.NodeDescription
Description returns the catalog entry of the mongodb node: it runs one operation (find, insertOne, updateOne or deleteOne) on a collection. find returns one item per document found, the others one item that summarises the result. The connection URI is a reference to a dataflow secret.
func (MongoDB) Execute ¶
func (MongoDB) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute resolves the connection URI from the secrets of the run, connects, runs the operation and disconnects. It fails when the run has no secret resolver.
func (MongoDB) TestConnection ¶
func (MongoDB) TestConnection(ctx context.Context, rt *dataflow.Runtime, params map[string]any) error
TestConnection verifies uriSecretRef alone resolves and connects — it needs none of the other parameters (database/collection/operation/...). mongo.Connect itself is lazy (it doesn't verify connectivity), so this explicitly Pings — see dataflow.TestableExecutor.
type Redis ¶
type Redis struct{}
Redis is the "redis" node type. Unlike m9m's Redis node (an HTTP call to a REST-to-Redis gateway like Upstash — a real limitation for a plain self-hosted Redis, which exposes no such gateway by default), this uses go-redis's native client so it works against any standard Redis instance.
Parameters: addrSecretRef (required, resolves to "host:port"), passwordSecretRef (optional), operation (get/set/del/keys/hget/hset/lpush/ rpush), key, value (for set/hset/lpush/rpush), field (for hget/hset).
func (Redis) Description ¶
func (Redis) Description() dataflow.NodeDescription
Description returns the catalog entry of the redis node: it runs one Redis command (get, set, del, keys, hget, hset, lpush or rpush) and returns one item with the result. The address and the password are given as references to dataflow secrets.
func (Redis) Execute ¶
func (Redis) Execute(ctx context.Context, rt *dataflow.Runtime, input []dataflow.Item, params map[string]any) (dataflow.Output, error)
Execute resolves the address (and the password, when given) from the secrets of the run, connects, runs the operation and returns one item with its result. It fails when the run has no secret resolver.
func (Redis) TestConnection ¶
TestConnection verifies addrSecretRef (+ passwordSecretRef, if set) alone resolve and connect — see dataflow.TestableExecutor.
Source Files