flow

package module
v1.2.16 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 30, 2026 License: MIT Imports: 36 Imported by: 0

README

Flow 🌊

flow is an SSIS-like and embeddable data pipeline orchestration and stream ETL library for Go. It allows developers to programmatically load, validate, and execute complex pipeline AST nodes (such as loops, parallel batches, and dynamic SQL/Go scripts) from XML configuration files.

go-flow is a fully functional implementation that takes and executes XML files, with config.xml overrides, and can be found at github.com/etl-madness/go-flow

FLOW Source code can be found at github.com/etl-madness/flow

Key Features

  • No Global State: Fully isolated execution contexts (Registry) allowing you to run multiple pipelines concurrently in the same process without interference.
  • Dynamic Configuration Decoder: Built-in support for XML parsers and optional XSD schema schema validations.
  • Direct Streaming ETL: Copy bulk datasets line-by-line across heterogeneous engines (PostgreSQL, SQLite, MySQL, Oracle, SQL Server) with automatic parameter placeholder syntax correction.
  • Flexible Flow Controls: Execution structures for Sequential steps, Parallel queues, If/Else branching, ForEach loops, and While loops.
  • Embedded Script Interpreter: Dynamic runtime Go evaluations via Yaegi with closure-bound environment state injection.

🐚 OS Shell & Command Execution

flow supports, in addition to database connections, the execution of native host shell commands and binaries directly on the operating system without passing through the Go interpreter[cite: 4]. Supported language options on <script> tags include:

  • shell: Cross-platform default shell (cmd /C on Windows, sh -c on Linux/macOS)[cite: 4].
  • cmd: Windows Command Prompt (cmd /C)[cite: 4].
  • powershell,pwsh: Windows PowerShell (powershell -NoProfile -NonInteractive -Command)[cite: 4].
  • bash,zsh,ksh,csh,tcsh,dash,fish,sh: Various Unix shells (bash -c, zsh -c, etc.)[cite: 4].
  • dotnet-script (or csx): Executed using C# script files with dotnet-script or dotnet script. Allows full inline C# execution including external NuGet package references (#r "nuget: ..."). This requires the dotnet-script tool to be installed on the host machine and the ability to create temporary files.
Key Capabilities
  • Variable Interpolation: Use {{var_name}} syntax inside script bodies to dynamically inject pipeline variables[cite: 3, 4].
  • Output Capture: Define the output_var attribute to save standard output/error into a pipeline variable for downstream consumption by SQL or Go steps[cite: 2, 4].
XML Examples
<pipeline>
    <variables>
        <variable name="export_dir" value="C:\exports" />
    </variables>
    <flow>
        <!-- Run external executable and store output in variable -->
        <script id="ExtractData" language="shell" output_var="GCLOUD_BILLING_JSON">
            ..\bqBilling.exe
        </script>

        <!-- PowerShell execution with variable interpolation -->
        <script id="PrepFolder" language="powershell">
            New-Item -ItemType Directory -Force -Path "{{export_dir}}"
        </script>

        <!-- Bash execution -->
        <script id="ArchiveLogs" language="bash" output_var="ARCHIVE_LOG">
            tar -czvf {{export_dir}}/archive.tar.gz {{export_dir}}/*.csv
        </script>
    </flow>
</pipeline>

Package API Reference

1. Parsing & Validation
// ParseXMLConfig parses a byte stream of XML into structured configuration blocks.
func ParseXMLConfig(xmlData []byte) ([]VariableConfig, []DatabaseConfig, []PipelineNode, error)

// ValidateAST performs semantic structure checks (uniqueness, reference integrity, loop bounds).
func ValidateAST(nodes []PipelineNode, registeredDBs []DatabaseConfig) error

// ValidateXSD invokes 'xmllint' to validate an XML configuration against schema standards.
func ValidateXSD(xmlPath string, xsdPath string) error

// GetSchemaXSD returns the compiled-in, embedded XSD schema file as a byte slice.
func GetSchemaXSD() []byte
2. State & Context Management
// Registry holds thread-safe variable registries and database connection pools.
type Registry struct { ... }

func NewRegistry() *Registry
func (r *Registry) InitVariables(configs []VariableConfig) error
func (r *Registry) InitDatabases(configs []DatabaseConfig) error
func (r *Registry) CloseDatabases()

// Variables getters & setters
func (r *Registry) SetVar(name string, value interface{})
func (r *Registry) GetVar(name string) interface{}
func (r *Registry) GetVarString(name string) string
func (r *Registry) GetVarInt(name string) int
func (r *Registry) GetVarBool(name string) bool
3. Pipeline Executor
// Executor orchestrates tree node executions.
type Executor struct { ... }

func NewExecutor(r *Registry) *Executor
func (e *Executor) Execute(ctx context.Context, nodes []PipelineNode) ([]ScriptResult, error)
func (e *Executor) SetVerbose(verbose bool)
func (e *Executor) SetGoPath(goPath string) // sets the GOPATH for the embedded Go interpreter (Yaegi) to resolve imports during script execution.

// ScriptResult represents the outcome of an executed script or loop step.
type ScriptResult struct {
	ScriptID      string `json:"script_id"`
	ReturnCode    any    `json:"return_code"`      // 0 on success, or an error string/code on failure
	ResultsString string `json:"results_string"`    // Output logs/data from query or script
	Duration      string `json:"duration,omitempty"` // Execution duration (e.g. "14.285ms")
}

Quick Start Example

The following example demonstrates how to load, parse, validate, and execute an XML pipeline programmatically from custom Go code.

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/etl-madness/flow"
)

func main() {
	xsdSchema := flow.GetSchemaXSD() // Load embedded XSD schema for validation
	xmlConfig := []byte(`<?xml version="1.0" encoding="UTF-8"?>
	<pipeline>
		<variables>
			<variable name="TargetTable" value="processed_logs" />
			<variable name="Threshold" type="int" value="100" />
		</variables>
	</pipeline>`)
	xmlScript := []byte(`<?xml version="1.0" encoding="UTF-8"?>
	<pipeline>
		<variables>
			<variable name="TargetTable" value="processed_logs" />
			<variable name="Threshold" type="int" value="100" />
		</variables>
		<databases>
			<database name="sqlite_db" driver="sqlite" connection_string="./mydb.db" />
		</databases>
		<flow>
			<script id="SetupTable" language="sql" db="sqlite_db">
				CREATE TABLE IF NOT EXISTS processed_logs (id INTEGER PRIMARY KEY, status TEXT);
			</script>
			<script id="VerifyGo" language="go">
				package main
				import (
					"fmt"
					"host/vars"
				)
				func main() {
					tbl := vars.GetString("TargetTable")
					thresh := vars.GetInt("Threshold")
					fmt.Printf("Configured target table: %s with limit: %d\n", tbl, thresh)
				}
			</script>
		</flow>
	</pipeline>`)

	// 1. Parse XML to Pipeline AST
	varConfigs, dbConfigs, nodes, err := flow.ParseXMLConfig(xmlConfig)
	if err != nil {
		log.Fatalf("Parsing failed: %v", err)
	}
	if err := flow.ValidateXSD(xmlConfig, string(xsdSchema)); err != nil {
		log.Fatalf("XSD validation failed: %v", err)
	}
	if err := flow.ValidateXSD(xmlScript, string(xsdSchema)); err != nil {
		log.Fatalf("XSD validation failed: %v", err)
	}
	// 2. Perform semantic checks on the AST
	if err := flow.ValidateAST(nodes, dbConfigs); err != nil {
		log.Fatalf("Validation failed: %v", err)
	}

	// 3. Instantiate Registry and Register Connection Pools / Variables
	registry := flow.NewRegistry()
	if err := registry.InitVariables(varConfigs); err != nil {
		log.Fatalf("Variables initialization failed: %v", err)
	}
	if err := registry.InitDatabases(dbConfigs); err != nil {
		log.Fatalf("Databases initialization failed: %v", err)
	}
	defer registry.CloseDatabases()

	// 4. Instantiate Executor and run the pipeline
	executor := flow.NewExecutor(registry)
	results, err := executor.Execute(context.Background(), nodes)
	if err != nil {
		log.Fatalf("Execution encountered errors: %v", err)
	}

	// 5. Inspect Results
	fmt.Println("--- Pipeline Execution Results ---")
	for _, res := range results {
		fmt.Printf("Script [%s]: Return Code: %v\n", res.ScriptID, res.ReturnCode)
		if res.ResultsString != "" {
			fmt.Printf("Output:\n%s\n", res.ResultsString)
		}
	}
}

Advanced: Shared Context Isolation

Since the state of connections and active variables is entirely held inside the *flow.Registry object rather than package globals, you can safely initialize multiple independent registries and run them in concurrent threads or separate executors:

registryA := flow.NewRegistry()
registryB := flow.NewRegistry()

// Run independent pipelines in parallel
go flow.NewExecutor(registryA).Execute(context.Background(), nodesA)
go flow.NewExecutor(registryB).Execute(context.Background(), nodesB)

🔧 Customizing Go Interpreter Options

When running dynamic Go scripts, flow uses the Yaegi interpreter under the hood. You can customize the interp.Options struct (e.g., to enable unrestricted code execution) by registering an interpreter configuration hook before executing your pipeline nodes.

Example: Unrestricted Execution

By default, Yaegi restricts some types of execution for safety. If your scripts require complete system access (like reading files, starting subprocesses, or bypassing other sandbox restrictions), you can enable Unrestricted: true:

import (
	"context"
	"github.com/traefik/yaegi/interp"
	"github.com/etl-madness/flow"
)

func main() {
	executor := flow.NewExecutor(registry)

	// Set a hook to modify Yaegi's interpreter options
	executor.SetInterpHook(func(opts *interp.Options) {
		opts.Unrestricted = true
	})

	results, err := executor.Execute(context.Background(), nodes)
}

📢 Verbose Execution Logging

To monitor the start, finish, duration, and outcome of each task in real-time as the pipeline processes them, you can enable verbose logging on the Executor.

Enabling Verbose Mode

By default, the executor runs silently. Use the SetVerbose(true) method before triggering your pipeline to output execution summaries directly to the console:

executor := flow.NewExecutor(registry)

// Enable verbose logging to console
executor.SetVerbose(true)

results, err := executor.Execute(context.Background(), nodes)
Console Output Format

When verbose mode is enabled, the executor logs task lifecycles in the following format:

Starting execution of script "SetupTable"
Finished execution of script "SetupTable" (duration: 4.812ms)
Starting execution of script "VerifyGo"
Finished execution of script "VerifyGo" (duration: 1.251ms)

If a task encounters an error during execution, the failure details are printed along with the elapsed time:

Starting execution of script "FailedQuery"
Finished execution of script "FailedQuery" with error: table 'non_existent_table' not found (duration: 3.125ms)

⚡ Parallel Execution Engine

The <parallel> block allows you to execute multiple child nodes concurrently. It includes a built-in semaphore-based throttle and thread-safe error-handling behaviors.

How it Works
  1. Concurrency Throttle (max_threads): You can specify the max_threads attribute on a <parallel> block. If unspecified or set to <= 0, it defaults to 4. The engine uses a buffered channel semaphore to guarantee that no more than max_threads goroutines run concurrently.
  2. Fail-Fast Error & Context Cancellation: If any concurrent child node encounters an error or if the pipeline context.Context is cancelled/timed out, execution halts immediately (fail-fast), preventing wasted compute resources.
  3. Variable Isolation & Conflict Resolution:
    • Each parallel worker is isolated with a cloned registry snapshot containing the state of variables at the moment of start.
    • Each worker's registry receives a thread-specific variable named _THREAD_ID.
    • To prevent stale overwrites, only variables actually modified (dirtyVars) by a worker are considered during the merge phase.
    • If multiple parallel workers mutate the same variable name, they are automatically namespaced as WORKER_<id>_<name> to prevent race conditions or collisions. Non-colliding keys merge directly back into the parent registry.
  4. Thread-Safe Results Accumulation: All script and task outcomes are safely accumulated into the final results list via internal mutex locking (resultsMu).
XML Configuration Example

The following XML segment configures a parallel block of 3 tasks running with a maximum concurrency limit of 2:

<pipeline>
    <databases>
        <database name="main_db" driver="sqlite" connection_string="./mydb.db" />
    </databases>
    <flow>
        <parallel max_threads="2">
            <script id="ProcessBatchA" language="sql" db="main_db">
                UPDATE transactions SET processed = 1 WHERE batch_id = 'A';
            </script>
            <script id="ProcessBatchB" language="sql" db="main_db">
                UPDATE transactions SET processed = 1 WHERE batch_id = 'B';
            </script>
            <script id="ProcessBatchC" language="sql" db="main_db">
                UPDATE transactions SET processed = 1 WHERE batch_id = 'C';
            </script>
        </parallel>
    </flow>
</pipeline>
Concurrently Nesting Loops (e.g. <foreach>)

Yes! Since the child elements in a <parallel> block are fully parsed as generic pipeline nodes, <parallel> natively supports concurrently running multiple loops (such as <foreach>) or nested structures.

When you nest <foreach> blocks inside <parallel>, each loop executes concurrently in parallel on its own thread, while the individual iterations within each loop run sequentially.

Example: Concurrently Running Independent Data Processors

Below is an XML pipeline configuring two <foreach> loops running simultaneously to import customers and products in parallel:

<pipeline>
    <databases>
        <database name="src_db" driver="sqlite" connection_string="./source.db" />
        <database name="target_db" driver="postgres" connection_string="postgresql://user:pass@localhost/db" />
    </databases>
	<script id="StreamData_MSSQL" language="sql" db="src_db" target_db="target_db" target_table="customers" batch_size="10000" tablock="true" check_constraints="false" fire_triggers="false" keep_nulls="true">
    SELECT id, name, email FROM source_customers;
    </script>
    <flow>
        <parallel max_threads="2">
            <!-- Loop 1: Import customer records -->
            <foreach id="SyncCustomers" db="src_db" var="customer_id">
                SELECT id FROM customers WHERE sync_pending = 1;
                <script id="MigrateCustomer" language="sql" db="src_db" target_db="target_db" target_table="customers" batch_size="100">
                    SELECT name, email, country FROM customers WHERE id = {{customer_id}};
                </script>
            </foreach>

            <!-- Loop 2: Import product records concurrently -->
            <foreach id="SyncProducts" db="src_db" var="product_id">
                SELECT id FROM products WHERE stock &gt; 0;
                <script id="MigrateProduct" language="sql" db="src_db" target_db="target_db" target_table="products" batch_size="50">
                    SELECT title, price, SKU FROM products WHERE id = {{product_id}};
                </script>
            </foreach>
        </parallel>
    </flow>
</pipeline>

[!TIP] Use parallel blocks for network-bound tasks, independent bulk database loads, or concurrent loops (like syncing separate data tables) where workflows do not rely on each other's outputs.

🚀 High-Performance MSSQL Bulk Copy Support

flow supports high-performance native bulk stream copy operations when transferring datasets into Microsoft SQL Server (sqlserver or mssql drivers). When streaming data to a SQL Server target, flow bypasses standard parameterized multi-row INSERT operations (which are subject to the 2,100 parameter limit) and instead utilizes native TDS Bulk Copy Streams (mssql.CopyIn).

XML Configuration Attributes

On any streaming <script> node (where both target_db and target_table are defined), you can configure the following bulk copy options:

  • tablock (boolean, optional, default true): Acquires a table-level lock during the bulk insert, drastically reducing transaction log overhead and boosting throughput.
  • check_constraints (boolean, optional, default false): Evaluates check and foreign key constraints on the target table during bulk insert.
  • fire_triggers (boolean, optional, default false): Executes any insert triggers defined on the target table during bulk execution.
  • keep_nulls (boolean, optional, default false): Retains explicit NULL values from the source dataset instead of utilizing target table default values.
XML Example
<pipeline>
    <databases>
        <database name="src_db" driver="sqlite" connection_string="./source.db" />
        <database name="dst_mssql" driver="sqlserver" connection_string="sqlserver://user:pass@localhost:1433?database=target_db" />
    </databases>
    <flow>
        <script id="BulkSync" 
                language="sql" 
                db="src_db" 
                target_db="dst_mssql" 
                target_table="customers" 
                batch_size="25000" 
                tablock="true" 
                check_constraints="true" 
                fire_triggers="false" 
                keep_nulls="true">
            SELECT id, name, email, signup_date FROM raw_users;
        </script>
    </flow>
</pipeline>
Fallback Driver Support

For non-MSSQL destination databases (e.g. PostgreSQL, MySQL, SQLite), flow automatically falls back to standard multi-row parameter-bound batch inserts. The batch sizes for fallback drivers are automatically throttled to ensure they never exceed the maximum 2,100 parameter limit (calculated dynamically as 2100 / column_count).

📂 Project Structure

.
├── .gitignore          # Git exclusion rules
├── LICENSE             # MIT License
├── README.md           # This document
├── go.mod              # Go module definition
├── go.sum              # Go dependencies checksums
├── config.go           # XML parsing and schema validation functions
├── etl.go              # Database stream copying implementation
├── etl_test.go         # Core placeholder unit tests
├── executor.go         # Core AST walker and script runner
├── registry.go         # Environment variable and connection pool registry
├── transactions.md     # Documentation on database transactions
├── validator.go        # Semantic AST validator rules
├── variables.md        # Documentation on variable management & usage
└── xsd/
    └── schema.xsd      # XML validation schema

Documentation

Overview

Package flow provides a high-performance, modular, and embeddable data pipeline orchestration and stream ETL library for Go. It allows developers to programmatically load, validate, and execute complex pipeline AST nodes (such as loops, parallel batches, and dynamic SQL/Go scripts) from XML configuration files.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func BuildClientAndRequest added in v1.2.11 

func BuildClientAndRequest(elem HTTPClientElement) (*http.Client, *http.Request, error)

BuildClientAndRequest constructs fully configured http.Client and http.Request instances.

func GetSchemaXSD added in v1.1.0 

func GetSchemaXSD() []byte

GetSchemaXSD returns the embedded XSD schema content as a byte slice.

func StreamETL 

func StreamETL(ctx context.Context, r *Registry, srcDBName, queryStr, dstDBName, targetTable string, opts ETLOptions) (int64, error)

StreamETL streams query results line-by-line from a source database into a target database table.

func ValidateAST 

func ValidateAST(preflightNodes []PipelineNode, flowNodes []PipelineNode, registeredDBs []DatabaseConfig) error

ValidateAST verifies script IDs, database names, and structural rules across preflight and flow ASTs.

func ValidateXSD 

func ValidateXSD(xmlPath string, xsdPath string) error

ValidateXSD invokes 'xmllint' to validate the XML file against the given XSD schema.

Types

type AssertElement added in v1.2.15 

type AssertElement struct {
	ID           string         `xml:"id,attr"`
	Var          string         `xml:"var,attr"`
	Equals       string         `xml:"equals,attr"`
	Value        string         `xml:"value,attr"`
	Operator     string         `xml:"operator,attr"`
	Message      string         `xml:"message,attr"`
	OnFailure    string         `xml:"on_failure,attr"` // "halt", "warn", "continue", "set_var"
	FailVar      string         `xml:"fail_var,attr"`
	FailVal      string         `xml:"fail_val,attr"`
	FailureNodes []PipelineNode // Nodes inside <on_failure> block
}

type DBHandle 

type DBHandle struct {
	Conn   *sql.DB // Connection pool handle
	Driver string  // Database driver name (e.g. "sqlite", "mysql")
}

DBHandle encapsulates an active sql.DB connection pool along with its driver name.

type DatabaseConfig 

type DatabaseConfig struct {
	Name             string // Unique identifier for the database
	Driver           string // Database driver name (e.g. postgres, mysql, sqlite)
	ConnectionString string // Driver-specific connection string
}

DatabaseConfig represents a database connection setup defined in the XML.

type ETLOptions added in v1.2.0 

type ETLOptions struct {
	BatchSize        int  // Number of rows per batch flush
	Tablock          bool // Acquire table lock for minimal logging on SQL Server
	CheckConstraints bool // Enforce target constraints during MSSQL bulk insert
	FireTriggers     bool // Execute target table triggers during MSSQL bulk insert
	KeepNulls        bool // Preserve explicit NULL values during MSSQL bulk insert
}

ETLOptions encapsulates batching and engine-specific performance tuning flags.

type ExcelReadElement added in v1.2.13 

type ExcelReadElement struct {
	ID        string `xml:"id,attr"`
	File      string `xml:"file,attr"`
	Sheet     string `xml:"sheet,attr"`
	Header    *bool  `xml:"header,attr"`
	Var       string `xml:"var,attr"`
	OutputVar string `xml:"output_var,attr"`
}

func (*ExcelReadElement) GetOutputVar added in v1.2.13 

func (e *ExcelReadElement) GetOutputVar() string

type ExcelWriteElement added in v1.2.13 

type ExcelWriteElement struct {
	ID     string `xml:"id,attr"`
	File   string `xml:"file,attr"`
	Sheet  string `xml:"sheet,attr"`
	DBName string `xml:"db,attr"`
	Var    string `xml:"var,attr"`
	Query  string `xml:",chardata"`
}

type Executor 

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

Executor orchestrates recursive pipeline AST node executions.

func NewExecutor 

func NewExecutor(r *Registry) *Executor

NewExecutor creates and returns a new Executor configured with the provided Registry.

func (*Executor) Execute 

func (e *Executor) Execute(ctx context.Context, nodes []PipelineNode) ([]ScriptResult, error)

Execute triggers sequential or parallel tree evaluation for a slice of PipelineNodes.

func (*Executor) SetGoPath added in v1.2.1 

func (e *Executor) SetGoPath(goPath string)

func (*Executor) SetInterpHook added in v1.2.4 

func (e *Executor) SetInterpHook(hook func(*interp.Options))

SetInterpHook registers a callback to customize Yaegi interpreter options.

func (*Executor) SetVerbose added in v1.1.0 

func (e *Executor) SetVerbose(verbose bool)

SetVerbose sets whether execution start and finish events should be printed to the console.

type FileReadElement added in v1.2.13 

type FileReadElement struct {
	ID             string `xml:"id,attr"`
	File           string `xml:"file,attr"`
	Path           string `xml:"path,attr"`
	Filename       string `xml:"filename,attr"`
	Var            string `xml:"var,attr"`
	Variable       string `xml:"variable,attr"`
	OutputVar      string `xml:"output_var,attr"`
	OutputVariable string `xml:"output_variable,attr"`
	OutVar         string `xml:"out_var,attr"`
}

func (*FileReadElement) GetFilePath added in v1.2.13 

func (f *FileReadElement) GetFilePath() string

func (*FileReadElement) GetOutputVar added in v1.2.13 

func (f *FileReadElement) GetOutputVar() string

type FileSaveElement added in v1.2.13 

type FileSaveElement struct {
	ID       string `xml:"id,attr"`
	File     string `xml:"file,attr"`
	Path     string `xml:"path,attr"`
	Filename string `xml:"filename,attr"`
	Var      string `xml:"var,attr"`
	Variable string `xml:"variable,attr"`
	Append   *bool  `xml:"append,attr"`
	Content  string `xml:",chardata"`
}

func (*FileSaveElement) GetFilePath added in v1.2.13 

func (f *FileSaveElement) GetFilePath() string

func (*FileSaveElement) GetInputVar added in v1.2.13 

func (f *FileSaveElement) GetInputVar() string

type HTTPClientElement added in v1.2.11 

type HTTPClientElement struct {
	XMLName xml.Name `xml:"-"`

	// Core Request Attributes
	ID          string `xml:"id,attr"`
	URI         string `xml:"uri,attr"`
	URL         string `xml:"url,attr"`
	Method      string `xml:"method,attr"`
	Data        string `xml:"data,attr"`
	BodyContent string `xml:",chardata"`
	Headers     string `xml:"headers,attr"`
	ContentType string `xml:"content_type,attr"`

	// Variable Output Assignments
	Var            string `xml:"var,attr"`
	Variable       string `xml:"variable,attr"`
	OutputVar      string `xml:"output_var,attr"`
	OutputVariable string `xml:"output_variable,attr"`
	OutVar         string `xml:"out_var,attr"`

	StatusCodeVar      string `xml:"status_code_var,attr"`
	StatusCodeVariable string `xml:"status_code_variable,attr"`
	StatusVar          string `xml:"status_var,attr"`
	StatusVariable     string `xml:"status_variable,attr"`

	// http.Client Attributes
	Timeout         string `xml:"timeout,attr"`
	MaxRedirects    *int   `xml:"max_redirects,attr"`
	FollowRedirects *bool  `xml:"follow_redirects,attr"`
	CookieJar       *bool  `xml:"cookie_jar,attr"`

	// http.Transport Attributes
	Proxy                  string `xml:"proxy,attr"`
	TLSInsecureSkipVerify  *bool  `xml:"tls_insecure_skip_verify,attr"`
	TLSHandshakeTimeout    string `xml:"tls_handshake_timeout,attr"`
	TLSServerName          string `xml:"tls_server_name,attr"`
	TLSMinVersion          string `xml:"tls_min_version,attr"`
	TLSMaxVersion          string `xml:"tls_max_version,attr"`
	DisableKeepAlives      *bool  `xml:"disable_keep_alives,attr"`
	DisableCompression     *bool  `xml:"disable_compression,attr"`
	MaxIdleConns           *int   `xml:"max_idle_conns,attr"`
	MaxIdleConnsPerHost    *int   `xml:"max_idle_conns_per_host,attr"`
	MaxConnsPerHost        *int   `xml:"max_conns_per_host,attr"`
	IdleConnTimeout        string `xml:"idle_conn_timeout,attr"`
	ResponseHeaderTimeout  string `xml:"response_header_timeout,attr"`
	ExpectContinueTimeout  string `xml:"expect_continue_timeout,attr"`
	MaxResponseHeaderBytes *int64 `xml:"max_response_header_bytes,attr"`
	WriteBufferSize        *int   `xml:"write_buffer_size,attr"`
	ReadBufferSize         *int   `xml:"read_buffer_size,attr"`
	ForceAttemptHTTP2      *bool  `xml:"force_attempt_http2,attr"`
}

HTTPClientElement maps all attributes from the HttpClientType XML schema[cite: 1].

func (*HTTPClientElement) GetOutputVariable added in v1.2.11 

func (e *HTTPClientElement) GetOutputVariable() string

Helper methods to identify target output variables

func (*HTTPClientElement) GetStatusCodeVariable added in v1.2.11 

func (e *HTTPClientElement) GetStatusCodeVariable() string

type JsonPathElement added in v1.2.13 

type JsonPathElement struct {
	ID        string `xml:"id,attr"`
	File      string `xml:"file,attr"`
	Var       string `xml:"var,attr"`
	Path      string `xml:"path,attr"`
	JSONPath  string `xml:"jsonpath,attr"`
	Content   string `xml:",chardata"` // Captures inner element body text
	Mode      string `xml:"mode,attr"` // "value", "json", "json_array"
	OutputVar string `xml:"output_var,attr"`
	OutVar    string `xml:"out_var,attr"`
}

func (*JsonPathElement) GetJSONPath added in v1.2.13 

func (j *JsonPathElement) GetJSONPath() string

func (*JsonPathElement) GetOutputVar added in v1.2.13 

func (j *JsonPathElement) GetOutputVar() string

type NodeKind 

type NodeKind int

NodeKind represents the structural type of a PipelineNode. 

const (
	// NodeScript represents a leaf script execution step.
	NodeScript NodeKind = iota
	// NodeGroup represents a simple sequence container of nodes.
	NodeGroup
	// NodeIf represents a conditional branching sequence.
	NodeIf
	// NodeForEach represents an iterative driver loop.
	NodeForEach
	// NodeParallel represents a concurrent block container.
	NodeParallel
	// NodeWhile represents a condition-controlled iteration loop.
	NodeWhile
	// NodeHTTPClient represents an HTTP client execution step.
	NodeHTTPClient // Added NodeHTTPClient enum
	// NodeTemplate represents a template inclusion step.
	NodeTemplate // New enum item
	// NodeFileSave represents a file save operation step.
	NodeFileSave // New enum item for file save operation
	// NodeFileRead represents a file read operation step.
	NodeFileRead   // New enum item for file read operation
	NodeExcelRead  // New enum item for Excel read operation
	NodeExcelWrite // New enum item for Excel write operation
	NodeXMLXPath   // New enum item for XML XPath extraction
	NodeJSONPath   // New enum item for JSON path extraction
	NodeYAMLPath   // New enum item for YAML path extraction
	NodeSQL        // New enum item for standard SQL execution
	NodeSQLBulk    // New enum item for bulk SQL execution
	NodeAssert     // New enum item for assert operation
)

type PipelineConfig added in v1.2.16 

type PipelineConfig struct {
	Variables      []VariableConfig
	Databases      []DatabaseConfig
	PreflightNodes []PipelineNode
	FlowNodes      []PipelineNode
}

PipelineConfig encapsulates the complete parsed AST structure.

func ParseXMLConfig 

func ParseXMLConfig(xmlData []byte) (PipelineConfig, error)

ParseXMLConfig parses XML pipeline config definitions into separate Preflight and Flow ASTs.

type PipelineNode 

type PipelineNode struct {
	Kind          NodeKind           // Struct/flow type of the node
	MaxThreads    int                // Concurrency limit (only used for NodeParallel)
	MaxIterations int                // Infinite loop safety limit (only used for NodeWhile)
	Script        *ScriptItem        // Leaf script item payload (only used for NodeScript)
	HTTPClient    *HTTPClientElement // Added HTTP payload
	GroupID       string             // Structural/group name or ID
	IfVar         string             // Condition driver variable name
	IfEquals      string             // Expected variable value to match
	ForEachScript *ScriptItem        // Iterator driver script config (only used for NodeForEach)
	Children      []PipelineNode     // List of sequential child execution steps
	ElseNodes     []PipelineNode     // Else branching steps (only used for NodeIf)
	Transaction   bool               // Start transaction for this group
	DBName        string             // Database name for the transaction
	Template      *TemplateElement   // New payload field for template inclusion step
	FileSave      *FileSaveElement   // New payload field for file save operation
	FileRead      *FileReadElement   // New payload field for file read operation
	ExcelRead     *ExcelReadElement  // New payload field for Excel read operation
	ExcelWrite    *ExcelWriteElement // New payload field for Excel write operation
	XmlXPath      *XmlXPathElement   // New payload field for XML XPath extraction
	JsonPath      *JsonPathElement   // New payload field for JSON path extraction
	YamlPath      *YamlPathElement   // New payload field for YAML path extraction
	Assert        *AssertElement     // New enum item for assert operation
}

PipelineNode is an AST node in the pipeline execution tree.

type Registry 

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

Registry is a thread-safe container that manages active database connection pools and dynamic pipeline environment variables.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry instantiates and returns an empty Registry context.

func (*Registry) CloseDatabases 

func (r *Registry) CloseDatabases()

CloseDatabases closes all open database connections tracked inside the registry and removes them.

func (*Registry) CopyVariables 

func (r *Registry) CopyVariables() map[string]interface{}

CopyVariables creates and returns a thread-safe snapshot map of all current environment variables.

func (*Registry) GetDB 

func (r *Registry) GetDB(name string) (*sql.DB, error)

GetDB returns the direct sql.DB pointer for the requested database name, if registered.

func (*Registry) GetDBHandle 

func (r *Registry) GetDBHandle(name string) (DBHandle, error)

GetDBHandle returns the DBHandle wrapper (containing sql.DB and Driver name) for the database.

func (*Registry) GetVar 

func (r *Registry) GetVar(name string) interface{}

GetVar retrieves an environment variable's raw interface value in a thread-safe manner.

func (*Registry) GetVarBool 

func (r *Registry) GetVarBool(name string) bool

func (*Registry) GetVarFloat 

func (r *Registry) GetVarFloat(name string) float64

func (*Registry) GetVarInt 

func (r *Registry) GetVarInt(name string) int

GetVarInt retrieves a variable and returns its value as an integer (parsing strings if necessary).

func (*Registry) GetVarString 

func (r *Registry) GetVarString(name string) string

GetVarString retrieves a variable and returns its value formatted as a string.

func (*Registry) GetVarTime added in v1.2.8 

func (r *Registry) GetVarTime(name string) time.Time

GetVarTime retrieves a variable and returns its value as time.Time (parsing string dates if necessary).

func (*Registry) InitDatabases 

func (r *Registry) InitDatabases(configs []DatabaseConfig) error

InitDatabases opens connection pools for all supplied DatabaseConfigs with variable interpolation in connection strings.

func (*Registry) InitVariables 

func (r *Registry) InitVariables(configs []VariableConfig) error

InitVariables registers and parses multiple environment variables based on type configuration.

func (*Registry) MergeVariables added in v1.2.8 

func (r *Registry) MergeVariables(src map[string]interface{})

MergeVariables copies variable key-value pairs into the parent registry.

func (*Registry) SetVar 

func (r *Registry) SetVar(name string, value interface{})

SetVar sets an environment variable value in a thread-safe manner.

func (*Registry) Snapshot added in v1.2.2 

func (r *Registry) Snapshot() *Registry

Snapshot returns a new Registry instance with isolated variable storage while sharing the underlying database connection handles.

type ScriptItem 

type ScriptItem struct {
	ID               string // Unique identifier of the script
	Language         string // Language identifier (sql or go)
	DBName           string // Target database identifier for SQL queries
	TargetDB         string // Destination database identifier for streaming ETL
	TargetTable      string // Destination table name for streaming ETL
	BatchSize        int    // Maximum rows loaded per batch
	VarName          string // Input environment variable to pull script code from dynamically
	OutputVar        string // Environment variable to store the command's outputs or logs into
	Code             string // Inner script text/payload
	Tablock          bool   // Acquire table lock for minimal logging on SQL Server
	CheckConstraints bool   // Evaluate constraints during MSSQL bulk insert
	FireTriggers     bool   // Execute target table triggers during MSSQL bulk insert
	KeepNulls        bool   // Preserve explicit NULL values during MSSQL bulk insert
}

ScriptItem represents an executable script payload (either SQL or Go) with metadata.

type ScriptResult 

type ScriptResult struct {
	ScriptID      string `json:"script_id"`          // Unique script identifier
	ReturnCode    any    `json:"return_code"`        // 0 on success, or error details on failure
	ResultsString string `json:"results_string"`     // Output logs, driver queries, or execution results
	Duration      string `json:"duration,omitempty"` // Cumulative execution time
}

ScriptResult represents the complete outcome of a single executed script or loop block.

type TemplateElement added in v1.2.13 

type TemplateElement struct {
	ID        string `xml:"id,attr"`
	Name      string `xml:"name,attr"`
	File      string `xml:"file,attr"`
	Engine    string `xml:"engine,attr"`
	OutputVar string `xml:"output_var,attr"`
	Var       string `xml:"var,attr"`
	Content   string `xml:",chardata"`
}

func (*TemplateElement) GetOutputVar added in v1.2.13 

func (t *TemplateElement) GetOutputVar() string

type VariableConfig 

type VariableConfig struct {
	Name  string // Name of the variable
	Type  string // Type of the variable (e.g. string, int, bool, float)
	Value string // Value of the variable as a raw string
}

VariableConfig represents an individual environment variable loaded from XML.

type XmlXPathElement added in v1.2.13 

type XmlXPathElement struct {
	ID        string `xml:"id,attr"`
	File      string `xml:"file,attr"`
	Var       string `xml:"var,attr"`
	XPath     string `xml:"xpath,attr"`
	Content   string `xml:",chardata"` // Captures inner element body text
	Mode      string `xml:"mode,attr"` // "text", "xml", "json_array"
	OutputVar string `xml:"output_var,attr"`
}

func (*XmlXPathElement) GetXPath added in v1.2.13 

func (x *XmlXPathElement) GetXPath() string

type YamlPathElement added in v1.2.13 

type YamlPathElement struct {
	ID        string `xml:"id,attr"`
	File      string `xml:"file,attr"`
	Var       string `xml:"var,attr"`
	Path      string `xml:"path,attr"`
	YAMLPath  string `xml:"yamlpath,attr"`
	Content   string `xml:",chardata"` // Captures inner element body text
	Mode      string `xml:"mode,attr"` // "value", "json", "json_array", "yaml"
	OutputVar string `xml:"output_var,attr"`
	OutVar    string `xml:"out_var,attr"`
}

func (*YamlPathElement) GetOutputVar added in v1.2.13 

func (y *YamlPathElement) GetOutputVar() string

func (*YamlPathElement) GetYAMLPath added in v1.2.13 

func (y *YamlPathElement) GetYAMLPath() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.