fix: load raw sql from query / file

Author: realdatadriven

examples/raw_sql_file.sql

@@ -0,0 +1,21 @@
+-- Create a table
+CREATE TABLE employees (
+    id INT PRIMARY KEY,
+    name VARCHAR(100),
+    position VARCHAR(50),
+    salary DECIMAL(10, 2)
+);
+
+-- Insert data
+INSERT INTO employees (id, name, position, salary) VALUES
+(1, 'Alice Smith', 'Developer', 75000.00),
+(2, 'Bob Johnson', 'Manager', 90000.00);
+
+-- Select data
+SELECT * FROM employees;
+
+-- Update data
+UPDATE employees SET salary = 80000.00 WHERE id = 1;
+
+-- Delete data
+DELETE FROM employees WHERE id = 2;

examples/require_non_md.md

@@ -0,0 +1,38 @@
+# ETL
+
+```yaml metadata
+name: DB
+description: "Example extrating from S3 to a local sqlite3 file"
+connection: "duckdb:"
+active: true
+```
+
+## VERSION
+
+```yaml metadata
+name: VERSION
+description: "DDB Version"
+table: VERSION
+load_conn: "duckdb:"
+load_before_sql: "ATTACH 'database/DB.db' AS DB (TYPE SQLITE)"
+load_sql: 'CREATE OR REPLACE TABLE DB."<table>" AS SELECT version() AS "VERSION";'
+load_after_sql: "DETACH DB;"
+rows_sql: 'SELECT COUNT(*) AS "nrows" FROM DB."<table>"'
+active: true
+```
+
+# REQUIRES
+
+```yaml metadata
+name: REQUIRES               
+description: load dependencies
+active: true
+```
+
+## RAW_SQL
+```yaml
+name: RAW_SQL
+description: load raw sql from file
+path: examples/raw_sql_file.sql
+active: true
+```

internal/etlx/etlxlib.go

@@ -18,7 +18,8 @@ import (
 )
 
 type ETLX struct {
-	Config map[string]any
+	Config           map[string]any
+	autoLogsDisabled bool
 }
 
 func addAutoLoggs(md string) string {
@@ -93,10 +94,15 @@ func (etlx *ETLX) ConfigFromFile(filePath string) error {
 			return fmt.Errorf("failed convert the Notebook to MDText: %w", err)
 		}
 		// fmt.Println(mdText)
-		data = []byte(addAutoLoggs(mdText))
+		data = []byte(mdText)
+	}
+	if etlx.autoLogsDisabled {
+		//
+	} else {
+		data = []byte(addAutoLoggs(string(data)))
 	}
 	// Parse the Markdown content into an AST
-	reader := text.NewReader([]byte(addAutoLoggs(string(data))))
+	reader := text.NewReader(data)
 	return etlx.ParseMarkdownToConfig(reader, mdText)
 }
 

internal/etlx/load_requirements.go

@@ -2,6 +2,7 @@ package etlxlib
 
 import (
 	"fmt"
+	"os"
 	"time"
 )
 
@@ -75,7 +76,8 @@ func (etlx *ETLX) LoadREQUIRES(conf map[string]any, keys ...string) ([]map[strin
 		column, okColumn := itemMetadata["column"]
 		afterSQL, okAfter := itemMetadata["after_sql"]
 		config := make(map[string]any)
-		etl := &ETLX{Config: config}
+		etl := &ETLX{Config: config, autoLogsDisabled: true}
+		var mdConf any
 		if okQuery && query != "" {
 			conn, okCon := itemMetadata["connection"]
 			if !okCon {
@@ -130,7 +132,6 @@ func (etlx *ETLX) LoadREQUIRES(conf map[string]any, keys ...string) ([]map[strin
 				return nil
 			}
 			if len(*rows) > 0 {
-				var mdConf any
 				okConf := false
 				if column != nil && okColumn {
 					mdConf, okConf = (*rows)[0][column.(string)]
@@ -204,14 +205,30 @@ func (etlx *ETLX) LoadREQUIRES(conf map[string]any, keys ...string) ([]map[strin
 				return nil
 			}
 		}
-		for newConfKey, value := range etl.Config {
-			if newConfKey == "metadata" || newConfKey == "__order" || newConfKey == "order" {
-				continue
+		//fmt.Println("LOADED ETLX CONF:", etl.Config)
+		if len(etl.Config) == 1 && etl.Config["__order"] != nil {
+			etlx.Config[itemKey] = map[string]any{}
+			if okQuery && query != "" && mdConf != nil {
+				//etlx.Config[itemKey].(map[string]any)[itemKey] = mdConf.(string)
+				etlx.Config[itemKey] = mdConf.(string)
+			} else if path != nil && okPath {
+				data, err := os.ReadFile(path.(string))
+				if err != nil {
+					fmt.Printf("LOAD RAW FILE: failed to read file: %s", err)
+				} else {
+					etlx.Config[itemKey] = string(data)
+				}
 			}
-			if _, ok := etlx.Config[newConfKey]; !ok {
-				etlx.Config[newConfKey] = value
-			} else {
-				fmt.Println(newConfKey, "Already exists!")
+		} else {
+			for newConfKey, value := range etl.Config {
+				if newConfKey == "metadata" || newConfKey == "__order" || newConfKey == "order" {
+					continue
+				}
+				if _, ok := etlx.Config[newConfKey]; !ok {
+					etlx.Config[newConfKey] = value
+				} else {
+					fmt.Println(newConfKey, "Already exists!")
+				}
 			}
 		}
 		_log2["success"] = true

internal/etlx/run_etl.go

@@ -200,6 +200,10 @@ func (etlx *ETLX) ReplacePlaceholders(sql string, item map[string]any) (string,
 				_sql, _, _, err := etlx.QueryBuilder(nil, queryName)
 				if err != nil {
 					fmt.Printf("QUERY DOC ERR ON KEY %s: %v\n", queryName, err)
+					_q, _e := etlx.Config[queryName].(string)
+					if !_e {
+						return _q
+					}
 				} else {
 					return _sql
 				}
@@ -256,6 +260,10 @@ func (etlx *ETLX) getDynamicQueriesIfAny(conn db.DBInterface, sqlData any, item
 				_sql, _, _, err := etlx.QueryBuilder(nil, name)
 				if err != nil {
 					fmt.Printf("QUERY DOC ERR ON KEY %s: %v\n", name, err)
+					_q, _e := etlx.Config[name].(string)
+					if !_e {
+						query = _q
+					}
 				} else {
 					query = _sql
 				}
@@ -337,6 +345,10 @@ func (etlx *ETLX) getDynamicQueriesIfAny(conn db.DBInterface, sqlData any, item
 					_sql, _, _, err := etlx.QueryBuilder(nil, name)
 					if err != nil {
 						fmt.Printf("QUERY DOC ERR ON KEY %s: %v\n", name, err)
+						_q, _e := etlx.Config[name].(string)
+						if !_e {
+							query = _q
+						}
 					} else {
 						query = _sql
 					}
@@ -446,6 +458,10 @@ func (etlx *ETLX) ExecuteQuery(conn db.DBInterface, sqlData any, item map[string
 			_sql, _, _, err := etlx.QueryBuilder(nil, queries)
 			if err != nil {
 				fmt.Printf("QUERY DOC ERR ON KEY %s: %v\n", queries, err)
+				_q, _e := etlx.Config[queries].(string)
+				if !_e {
+					query = _q
+				}
 			} else {
 				query = _sql
 			}
@@ -494,6 +510,10 @@ func (etlx *ETLX) ExecuteQuery(conn db.DBInterface, sqlData any, item map[string
 				_sql, _, _, err := etlx.QueryBuilder(nil, queryKey)
 				if err != nil {
 					fmt.Printf("QUERY DOC ERR ON KEY %s: %v\n", queryKey, err)
+					_q, _e := etlx.Config[queryKey].(string)
+					if !_e {
+						query = _q
+					}
 				} else {
 					query = _sql
 				}