dbqcore v0.1.0

Author: sholokhov

cmd/check.go

@@ -46,7 +46,7 @@ By automating these checks, you can proactively identify and address data qualit
 			slog.Debug("Reading checks configuration file",
 				"checks_config_path", checksFile)
 
-			checksCfg, err := dbqcore.LoadChecksConfig(checksFile)
+			checksCfg, err := dbqcore.LoadChecksFileConfig(checksFile)
 			if err != nil {
 				return fmt.Errorf("error while loading checks configuration file: %w", err)
 			}
@@ -72,7 +72,7 @@ By automating these checks, you can proactively identify and address data qualit
 						pass, _, err := app.RunCheck(&check, dataSource, dataset, rule.Where)
 
 						fmt.Printf("  check %s ... %s\n", check.Description, getCheckResultLabel(pass))
-						if err == nil {
+						if pass {
 							passedCount += 1
 						} else {
 							failedChecks = append(failedChecks, FailedCheckDetails{
@@ -82,7 +82,7 @@ By automating these checks, you can proactively identify and address data qualit
 							})
 						}
 
-						if !pass && strGetOrDefault(string(check.OnFail), dbqcore.OnFailActionError) == dbqcore.OnFailActionError {
+						if !pass && strGetOrDefault(string(check.OnFail), string(dbqcore.OnFailActionError)) == string(dbqcore.OnFailActionError) {
 							exitCode = 1
 						}
 					}

cmd/version.go

@@ -17,12 +17,12 @@ package cmd
 import (
 	"fmt"
 
-	"github.com/DataBridgeTech/dbqcore"
+	"github.com/DataBridgeTech/dbqcore/dbq"
 	"github.com/spf13/cobra"
 )
 
 const (
-	DbqCtlVersion = "v0.1.1"
+	DbqCtlVersion = "v0.1.2"
 )
 
 func NewVersionCommand() *cobra.Command {
@@ -31,7 +31,7 @@ func NewVersionCommand() *cobra.Command {
 		Short: "Prints dbqctl and core lib version",
 		Run: func(cmd *cobra.Command, args []string) {
 			fmt.Printf("DataBridge Quality CLI: %s\n", DbqCtlVersion)
-			fmt.Printf("DataBridge dbqcore lib version: %s\n", dbqcore.GetDbqCoreLibVersion())
+			fmt.Printf("DataBridge dbqcore lib version: %s\n", dbq.GetDbqCoreLibVersion())
 		},
 	}
 

go.mod

@@ -3,14 +3,14 @@ module github.com/DataBridgeTech/dbqctl
 go 1.24.5
 
 require (
-	github.com/DataBridgeTech/dbqcore v0.0.8
+	github.com/DataBridgeTech/dbqcore v0.1.0
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/pflag v1.0.7
 	github.com/spf13/viper v1.20.1
 	gopkg.in/yaml.v3 v3.0.1
 )
 
-//replace github.com/DataBridgeTech/dbqcore => ../dbqcore
+replace github.com/DataBridgeTech/dbqcore => ../dbqcore
 
 require (
 	github.com/ClickHouse/ch-go v0.67.0 // indirect

go.sum

@@ -12,6 +12,7 @@ github.com/DataBridgeTech/dbqcore v0.0.3/go.mod h1:U2++eavpf8oSKxukSEJpKGfPaOtkA
 github.com/DataBridgeTech/dbqcore v0.0.5 h1:fvGtZNZRyUCAq7IM0bVfpZ/FlsGXMq0LmDf1vOoR1nU=
 github.com/DataBridgeTech/dbqcore v0.0.5/go.mod h1:U2++eavpf8oSKxukSEJpKGfPaOtkApA7AUpzwkrXmVM=
 github.com/DataBridgeTech/dbqcore v0.0.6/go.mod h1:kIYXBoNZyLQ02lCkTHJTl5Lu5Y28R2LMJrpNvTL+ySQ=
+github.com/DataBridgeTech/dbqcore v0.0.8/go.mod h1:kIYXBoNZyLQ02lCkTHJTl5Lu5Y28R2LMJrpNvTL+ySQ=
 github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
 github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=

internal/app.go

@@ -15,11 +15,12 @@
 package internal
 
 import (
-	"fmt"
+	"context"
 	"log/slog"
 	"os"
 
 	"github.com/DataBridgeTech/dbqcore"
+	"github.com/DataBridgeTech/dbqcore/dbq"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -30,7 +31,7 @@ type DbqCliApp interface {
 	PingDataSource(srcId string) (string, error)
 	ImportDatasets(srcId string, filter string) ([]string, error)
 	ProfileDataset(srcId string, dataset string, sample bool, maxConcurrent int) (*dbqcore.TableMetrics, error)
-	RunCheck(check *dbqcore.Check, dataSource *dbqcore.DataSource, dataset string, defaultWhere string) (bool, string, error)
+	RunCheck(check *dbqcore.DataQualityCheck, dataSource *dbqcore.DataSource, dataset string, defaultWhere string) (bool, string, error)
 	GetDbqConfig() *dbqcore.DbqConfig
 	SaveDbqConfig() error
 	SetLogLevel(level slog.Level)
@@ -41,26 +42,29 @@ type DbqAppImpl struct {
 	dbqConfigPath string
 	dbqConfig     *dbqcore.DbqConfig
 	logLevel      slog.Level
+	logger        *slog.Logger
 }
 
 func NewDbqCliApp(dbqConfigPath string) DbqCliApp {
 	dbqConfig, dbqConfigUsedPath := initConfig(dbqConfigPath)
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError}))
 	return &DbqAppImpl{
 		dbqConfigPath: dbqConfigUsedPath,
 		dbqConfig:     dbqConfig,
 		logLevel:      slog.LevelError,
+		logger:        logger, // todo: fix logger init
 	}
 }
 
 func (app *DbqAppImpl) PingDataSource(srcId string) (string, error) {
 	var dataSource = app.FindDataSourceById(srcId)
 
-	cnn, err := getDbqConnector(*dataSource, app.logLevel)
+	cnn, err := dbq.NewDbqConnector(dataSource, app.logger)
 	if err != nil {
 		return "", err
 	}
 
-	info, err := cnn.Ping()
+	info, err := cnn.Ping(context.Background()) // todo: ctx propagation
 	if err != nil {
 		return "", err
 	}
@@ -71,23 +75,23 @@ func (app *DbqAppImpl) PingDataSource(srcId string) (string, error) {
 func (app *DbqAppImpl) ImportDatasets(srcId string, filter string) ([]string, error) {
 	var dataSource = app.FindDataSourceById(srcId)
 
-	cnn, err := getDbqConnector(*dataSource, app.logLevel)
+	cnn, err := dbq.NewDbqConnector(dataSource, app.logger)
 	if err != nil {
 		return []string{}, err
 	}
 
-	return cnn.ImportDatasets(filter)
+	return cnn.ImportDatasets(context.Background(), filter) // todo: ctx propagation
 }
 
 func (app *DbqAppImpl) ProfileDataset(srcId string, dataset string, sample bool, maxConcurrent int) (*dbqcore.TableMetrics, error) {
 	var dataSource = app.FindDataSourceById(srcId)
 
-	cnn, err := getDbqConnector(*dataSource, app.logLevel)
+	dbqProfiler, err := dbq.NewDbqProfiler(dataSource, app.logger)
 	if err != nil {
 		return nil, err
 	}
 
-	return cnn.ProfileDataset(dataset, sample, maxConcurrent)
+	return dbqProfiler.ProfileDataset(context.Background(), dataset, sample, maxConcurrent) // todo: ctx propagation
 }
 
 func (app *DbqAppImpl) GetDbqConfig() *dbqcore.DbqConfig {
@@ -117,12 +121,13 @@ func (app *DbqAppImpl) FindDataSourceById(srcId string) *dbqcore.DataSource {
 	return nil
 }
 
-func (app *DbqAppImpl) RunCheck(check *dbqcore.Check, dataSource *dbqcore.DataSource, dataset string, defaultWhere string) (bool, string, error) {
-	cnn, err := getDbqConnector(*dataSource, app.logLevel)
+func (app *DbqAppImpl) RunCheck(check *dbqcore.DataQualityCheck, dataSource *dbqcore.DataSource, dataset string, defaultWhere string) (bool, string, error) {
+	dbqValidator, err := dbq.NewDbqValidator(dataSource, app.logger)
 	if err != nil {
 		return false, "", err
 	}
-	return cnn.RunCheck(check, dataset, defaultWhere)
+
+	return dbqValidator.RunCheck(context.Background(), check, dataset, defaultWhere) // todo: ctx propagation
 }
 
 func (app *DbqAppImpl) SetLogLevel(logLevel slog.Level) {
@@ -154,13 +159,3 @@ func initConfig(dbqConfigPath string) (*dbqcore.DbqConfig, string) {
 
 	return &dbqConfig, v.ConfigFileUsed()
 }
-
-func getDbqConnector(ds dbqcore.DataSource, logLevel slog.Level) (dbqcore.DbqConnector, error) {
-	logHandler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel})
-	switch ds.Type {
-	case dbqcore.DataSourceTypeClickhouse:
-		return dbqcore.NewClickhouseDbqConnector(ds, slog.New(logHandler))
-	default:
-		return nil, fmt.Errorf("data source type '%s' is not supported", ds.Type)
-	}
-}