初始提交: db模块

db.yaml

@@ -0,0 +1,10 @@
+database:
+  type: "oracle"  # mysql, postgres, oracle, sqlserver
+  host: "161.189.7.134"
+  port: 1521
+  username: "x6_stock_dev"
+  password: "mosdev"
+  database: "ORCL"
+  max_open_conns: 100
+  max_idle_conns: 10
+  conn_max_lifetime: 300

db_pg.yaml

@@ -0,0 +1,10 @@
+database:
+  type: "postgres"  # mysql, postgres, oracle, sqlserver
+  host: "69.235.172.218"
+  port: 5432
+  username: "x3stock"
+  password: "mos8555"
+  database: "x3stock"
+  max_open_conns: 100
+  max_idle_conns: 10
+  conn_max_lifetime: 300

drivers/driver.go

@@ -0,0 +1,76 @@
+package drivers
+
+import (
+	"database/sql"
+	"fmt"
+	"time" // 添加 time 包导入
+)
+
+// DBConfig 数据库配置
+type DBConfig struct {
+	Type            string `yaml:"type"`
+	Host            string `yaml:"host"`
+	Port            int    `yaml:"port"`
+	Username        string `yaml:"username"`
+	Password        string `yaml:"password"`
+	Database        string `yaml:"database"`
+	MaxOpenConns    int    `yaml:"max_open_conns"`
+	MaxIdleConns    int    `yaml:"max_idle_conns"`
+	ConnMaxLifetime int    `yaml:"conn_max_lifetime"` // 单位：秒
+}
+
+// DBDriver 数据库驱动接口
+type DBDriver interface {
+	// Name 返回驱动名称
+	Name() string
+	// Open 打开数据库连接
+	Open(config DBConfig) (*sql.DB, error)
+	// BuildDSN 构建连接字符串
+	BuildDSN(config DBConfig) string
+}
+
+var driverss = make(map[string]DBDriver)
+
+// Register 注册数据库驱动
+func Register(driver DBDriver) {
+	if driver == nil {
+		panic("db driver: Register driver is nil")
+	}
+	name := driver.Name()
+	if _, dup := driverss[name]; dup {
+		panic("db driver: Register called twice for driver " + name)
+	}
+	driverss[name] = driver
+}
+
+// Get 获取数据库驱动
+func Get(name string) (DBDriver, error) {
+	driver, exists := driverss[name]
+	if !exists {
+		return nil, fmt.Errorf("db driver: unknown driver %q", name)
+	}
+	return driver, nil
+}
+
+// GetAllDrivers 获取所有已注册的驱动名称
+func GetAllDrivers() []string {
+	names := make([]string, 0, len(driverss))
+	for name := range driverss {
+		names = append(names, name)
+	}
+	return names
+}
+
+// configureConnectionPool 配置连接池的公共函数
+func configureConnectionPool(db *sql.DB, config DBConfig) {
+	if config.MaxOpenConns > 0 {
+		db.SetMaxOpenConns(config.MaxOpenConns)
+	}
+	if config.MaxIdleConns > 0 {
+		db.SetMaxIdleConns(config.MaxIdleConns)
+	}
+	if config.ConnMaxLifetime > 0 {
+		// 将秒转换为 time.Duration
+		db.SetConnMaxLifetime(time.Duration(config.ConnMaxLifetime) * time.Second)
+	}
+}

drivers/mysql.go

@@ -0,0 +1,45 @@
+package drivers
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+type MySQLDriver struct{}
+
+func (d *MySQLDriver) Name() string {
+	return "mysql"
+}
+
+func (d *MySQLDriver) BuildDSN(config DBConfig) string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database)
+}
+
+func (d *MySQLDriver) Open(config DBConfig) (*sql.DB, error) {
+	dsn := d.BuildDSN(config)
+	db, err := sql.Open("mysql", dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	// 使用公共的连接池配置函数
+	configureConnectionPool(db, config)
+
+	// 测试连接
+	if err = db.Ping(); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func init() {
+	Register(&MySQLDriver{})
+}

drivers/oracle.go

@@ -0,0 +1,45 @@
+package drivers
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/sijms/go-ora/v2"
+)
+
+type OracleDriver struct{}
+
+func (d *OracleDriver) Name() string {
+	return "oracle"
+}
+
+func (d *OracleDriver) BuildDSN(config DBConfig) string {
+	return fmt.Sprintf("oracle://%s:%s@%s:%d/%s",
+		config.Username,
+		config.Password,
+		config.Host,
+		config.Port,
+		config.Database)
+}
+
+func (d *OracleDriver) Open(config DBConfig) (*sql.DB, error) {
+	dsn := d.BuildDSN(config)
+	db, err := sql.Open("oracle", dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	// 使用公共的连接池配置函数
+	configureConnectionPool(db, config)
+
+	// 测试连接
+	if err = db.Ping(); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func init() {
+	Register(&OracleDriver{})
+}

drivers/postgres.go

@@ -0,0 +1,45 @@
+package drivers
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/lib/pq"
+)
+
+type PostgresDriver struct{}
+
+func (d *PostgresDriver) Name() string {
+	return "postgres"
+}
+
+func (d *PostgresDriver) BuildDSN(config DBConfig) string {
+	return fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
+		config.Host,
+		config.Port,
+		config.Username,
+		config.Password,
+		config.Database)
+}
+
+func (d *PostgresDriver) Open(config DBConfig) (*sql.DB, error) {
+	dsn := d.BuildDSN(config)
+	db, err := sql.Open("postgres", dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	// 使用公共的连接池配置函数
+	configureConnectionPool(db, config)
+
+	// 测试连接
+	if err = db.Ping(); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func init() {
+	Register(&PostgresDriver{})
+}

drivers/sqlserver.go

@@ -0,0 +1,45 @@
+package drivers
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/microsoft/go-mssqldb"
+)
+
+type SQLServerDriver struct{}
+
+func (d *SQLServerDriver) Name() string {
+	return "sqlserver"
+}
+
+func (d *SQLServerDriver) BuildDSN(config DBConfig) string {
+	return fmt.Sprintf("server=%s;port=%d;user id=%s;password=%s;database=%s;encrypt=disable",
+		config.Host,
+		config.Port,
+		config.Username,
+		config.Password,
+		config.Database)
+}
+
+func (d *SQLServerDriver) Open(config DBConfig) (*sql.DB, error) {
+	dsn := d.BuildDSN(config)
+	db, err := sql.Open("sqlserver", dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	// 使用公共的连接池配置函数
+	configureConnectionPool(db, config)
+
+	// 测试连接
+	if err = db.Ping(); err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func init() {
+	Register(&SQLServerDriver{})
+}

factory/factory.go

@@ -0,0 +1,160 @@
+package factory
+
+import (
+	"database/sql"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"git.x2erp.com/qdy/go-base/types"
+	"git.x2erp.com/qdy/go-db/drivers"
+	"gopkg.in/yaml.v2"
+)
+
+// Config 总配置
+type Config struct {
+	Database drivers.DBConfig `yaml:"database"`
+}
+
+// DBFactory 数据库工厂
+type DBFactory struct {
+	config *Config
+}
+
+// NewDBFactory 创建数据库工厂
+func NewDBFactory() (*DBFactory, error) {
+	configFile, err := findConfigFile()
+	if err != nil {
+		return nil, err
+	}
+
+	fmt.Printf("✅ Using config file: %s\n", configFile)
+
+	// 读取配置文件
+	data, err := os.ReadFile(configFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read config file %s: %v", configFile, err)
+	}
+
+	var config Config
+	err = yaml.Unmarshal(data, &config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse config file: %v", err)
+	}
+
+	return &DBFactory{config: &config}, nil
+}
+
+// findConfigFile 查找配置文件
+func findConfigFile() (string, error) {
+	// 1. 首先尝试可执行文件同目录
+	exePath, err := os.Executable()
+	if err == nil {
+		exeDir := filepath.Dir(exePath)
+		configFile := filepath.Join(exeDir, "db.yaml")
+		if _, err := os.Stat(configFile); err == nil {
+			return configFile, nil
+		}
+	}
+
+	// 2. 尝试环境变量指定的路径
+	envConfigPath := os.Getenv("DB_CONFIG_PATH")
+	if envConfigPath != "" {
+		if _, err := os.Stat(envConfigPath); err == nil {
+			return envConfigPath, nil
+		}
+		return "", fmt.Errorf("DB_CONFIG_PATH file not found: %s", envConfigPath)
+	}
+
+	// 3. 如果都没有找到，返回错误
+	exeDir := "unknown"
+	if exePath, err := os.Executable(); err == nil {
+		exeDir = filepath.Dir(exePath)
+	}
+
+	return "", fmt.Errorf(`No configuration file found!
+
+Tried locations:
+1. Executable directory: %s/db.yaml
+2. Environment variable: DB_CONFIG_PATH
+
+Solutions:
+- Place db.yaml in the same directory as the executable
+- Or set DB_CONFIG_PATH environment variable to config file path
+
+Example:
+  export DB_CONFIG_PATH=/path/to/your/db.yaml`, exeDir)
+}
+
+// CreateDB 创建数据库连接
+func (f *DBFactory) CreateDB() (*sql.DB, error) {
+	dbType := f.config.Database.Type
+
+	// 获取对应的驱动
+	dbDriver, err := drivers.Get(dbType)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get database driver: %v", err)
+	}
+
+	// 创建数据库连接
+	db, err := dbDriver.Open(f.config.Database)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open database connection: %v", err)
+	}
+
+	return db, nil
+}
+
+// GetConfig 获取配置信息
+func (f *DBFactory) GetConfig() *Config {
+	return f.config
+}
+
+// GetAvailableDrivers 获取可用的数据库驱动
+func (f *DBFactory) GetAvailableDrivers() []string {
+	return drivers.GetAllDrivers()
+}
+
+// CreateQueryExecutor 创建查询执行器（新增方法）
+func (f *DBFactory) CreateQueryExecutor(db *sql.DB) *QueryExecutor {
+	return NewQueryExecutor(db)
+}
+
+// -------------- 对外暴露的初始化方法（核心入口）--------------
+// NewDBQuery 初始化查询实例（对外提供唯一初始化入口）
+// db: 已初始化的数据库连接（由调用方传入，解耦数据库配置）
+func NewDBQuery(db *sql.DB) *QueryExecutor {
+	// 直接复用原文件的构造函数，对外隐藏实现细节
+	return NewQueryExecutor(db)
+}
+
+// QuickQueryToJSON 快捷查询，直接返回 JSON 字节流
+func QuickQueryToJSON(db *sql.DB, sql string) *types.QueryResult {
+	return NewDBQuery(db).QueryToJSON(sql)
+}
+
+// QuickQueryToCSV 快捷查询，直接返回 CSV 字符串（包含表头）
+func QuickQueryToCSV(db *sql.DB, sql string) *types.QueryResult {
+	return NewDBQuery(db).QueryToCSV(sql)
+}
+
+// QuickExecuteWithColumns 快捷查询，返回完整结果（含列信息）
+func QuickExecuteWithColumns(db *sql.DB, sql string) *types.QueryResult {
+	return NewDBQuery(db).ExecuteQueryWithColumns(sql)
+}
+
+// QuickExecuteDataOnly 快捷查询，返回纯数据（性能优先）
+func QuickExecuteDataOnly(db *sql.DB, sql string) *types.QueryResult {
+	return NewDBQuery(db).ExecuteQueryDataOnly(sql)
+}
+
+// QuickExecuteCSV 快捷查询，返回 CSV 格式结果（支持自定义是否包含表头）
+func QuickExecuteCSV(db *sql.DB, sql string, includeHeader bool) *types.QueryResult {
+	return NewDBQuery(db).ExecuteQueryCSV(sql, includeHeader)
+}
+
+// QuickExecuteCSVStream 快捷流式输出 CSV（直接写入 io.Writer，适合大文件）
+func QuickExecuteCSVStream(db *sql.DB, sql string, w io.Writer, includeHeader bool) (int, error) {
+	return NewDBQuery(db).ExecuteQueryCSVStream(sql, w, includeHeader)
+}

factory/query.go

@@ -0,0 +1,554 @@
+package factory
+
+import (
+	"database/sql"
+	"encoding/csv"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"git.x2erp.com/qdy/go-base/types"
+)
+
+func typesFunction() *types.QueryResult { // 确保使用了该包导出的类型，例如 QueryResult
+	return &types.QueryResult{}
+}
+
+// QueryExecutor 查询执行器
+type QueryExecutor struct {
+	db *sql.DB
+}
+
+// NewQueryExecutor 创建查询执行器
+func NewQueryExecutor(db *sql.DB) *QueryExecutor {
+	return &QueryExecutor{db: db}
+}
+
+// QueryToJSON 执行查询并返回JSON格式数据（统一返回QueryResult）
+func (e *QueryExecutor) QueryToJSON(sql string) *types.QueryResult {
+	startTime := time.Now()
+	result := &types.QueryResult{}
+
+	if sql == "" {
+		result.Success = false
+		result.Error = "SQL query cannot be empty"
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Query execution failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to get columns: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	var results []map[string]interface{}
+	count := 0
+
+	for rows.Next() {
+		count++
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to scan row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+
+		resultMap := make(map[string]interface{})
+		for i, col := range columns {
+			// 完全不处理类型，直接赋值，让 json.Marshal 自己处理
+			resultMap[col] = values[i]
+		}
+		results = append(results, resultMap)
+	}
+
+	if err := rows.Err(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Row iteration error: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	jsonData, err := json.Marshal(results)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("JSON marshal failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	result.Success = true
+	result.Data = map[string]interface{}{
+		"json":  string(jsonData),
+		"rows":  results,
+		"count": count,
+	}
+	result.Count = count
+	result.Time = time.Since(startTime).String()
+	return result
+}
+
+// QueryToCSV 查询并返回 CSV 字符串（包含表头，统一返回QueryResult）
+func (e *QueryExecutor) QueryToCSV(sql string) *types.QueryResult {
+	startTime := time.Now()
+	result := &types.QueryResult{}
+
+	if sql == "" {
+		result.Success = false
+		result.Error = "SQL query cannot be empty"
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Query execution failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to get columns: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	var builder strings.Builder
+	writer := csv.NewWriter(&builder)
+	count := 0
+
+	// 写入表头
+	if err := writer.Write(columns); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to write CSV header: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	for rows.Next() {
+		count++
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]any, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to scan row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+
+		// 所有值转为字符串
+		row := make([]string, len(columns))
+		for i, val := range values {
+			if val == nil {
+				row[i] = ""
+			} else {
+				row[i] = fmt.Sprintf("%v", val)
+			}
+		}
+
+		if err := writer.Write(row); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to write CSV row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+	}
+
+	writer.Flush()
+	if err := writer.Error(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to flush CSV: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	if err := rows.Err(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Row iteration error: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	result.Success = true
+	result.Data = map[string]interface{}{
+		"csv":           builder.String(),
+		"count":         count,
+		"includeHeader": true,
+	}
+	result.Count = count
+	result.Time = time.Since(startTime).String()
+	return result
+}
+
+// ExecuteQueryWithColumns 执行查询并返回完整结果（包含列信息）
+func (e *QueryExecutor) ExecuteQueryWithColumns(sql string) *types.QueryResult {
+	startTime := time.Now()
+	result := &types.QueryResult{}
+
+	if sql == "" {
+		result.Success = false
+		result.Error = "SQL query cannot be empty"
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Query execution failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to get columns: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	var results []map[string]interface{}
+	count := 0
+
+	for rows.Next() {
+		count++
+
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to scan row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+
+		resultRow := make(map[string]interface{})
+		for i, col := range columns {
+			val := values[i]
+			switch v := val.(type) {
+			case []byte:
+				resultRow[col] = string(v)
+			case time.Time:
+				resultRow[col] = v.Format(time.RFC3339)
+			default:
+				resultRow[col] = v
+			}
+		}
+
+		results = append(results, resultRow)
+	}
+
+	if err := rows.Err(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Row iteration error: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	result.Success = true
+	result.Data = results
+	result.Count = count
+	result.Time = time.Since(startTime).String()
+
+	return result
+}
+
+// ExecuteQueryDataOnly 执行查询并返回纯数据（不包含列信息，性能更高）
+func (e *QueryExecutor) ExecuteQueryDataOnly(sql string) *types.QueryResult {
+	startTime := time.Now()
+	result := &types.QueryResult{}
+
+	if sql == "" {
+		result.Success = false
+		result.Error = "SQL query cannot be empty"
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Query execution failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to get columns: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	var results []interface{}
+	count := 0
+
+	for rows.Next() {
+		count++
+
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to scan row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+
+		resultRow := make([]interface{}, len(columns))
+		for i, val := range values {
+			switch v := val.(type) {
+			case []byte:
+				resultRow[i] = string(v)
+			case time.Time:
+				resultRow[i] = v.Format(time.RFC3339)
+			default:
+				resultRow[i] = v
+			}
+		}
+
+		results = append(results, resultRow)
+	}
+
+	if err := rows.Err(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Row iteration error: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	result.Success = true
+	result.Data = map[string]interface{}{
+		"rows":  results,
+		"count": count,
+	}
+	result.Count = count
+	result.Time = time.Since(startTime).String()
+
+	return result
+}
+
+// ExecuteQueryCSV 执行查询并返回CSV格式数据
+func (e *QueryExecutor) ExecuteQueryCSV(sql string, includeHeader bool) *types.QueryResult {
+	startTime := time.Now()
+	result := &types.QueryResult{}
+
+	if sql == "" {
+		result.Success = false
+		result.Error = "SQL query cannot be empty"
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Query execution failed: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to get columns: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	var csvBuilder strings.Builder
+	writer := csv.NewWriter(&csvBuilder)
+
+	if includeHeader {
+		if err := writer.Write(columns); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to write CSV header: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+	}
+
+	count := 0
+
+	for rows.Next() {
+		count++
+
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to scan row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+
+		rowData := make([]string, len(columns))
+		for i, val := range values {
+			if val == nil {
+				rowData[i] = ""
+				continue
+			}
+
+			switch v := val.(type) {
+			case []byte:
+				rowData[i] = string(v)
+			case string:
+				rowData[i] = v
+			case int, int8, int16, int32, int64:
+				rowData[i] = fmt.Sprintf("%d", v)
+			case uint, uint8, uint16, uint32, uint64:
+				rowData[i] = fmt.Sprintf("%d", v)
+			case float32, float64:
+				rowData[i] = fmt.Sprintf("%f", v)
+			case bool:
+				if v {
+					rowData[i] = "true"
+				} else {
+					rowData[i] = "false"
+				}
+			case time.Time:
+				rowData[i] = v.Format(time.RFC3339)
+			default:
+				rowData[i] = fmt.Sprintf("%v", v)
+			}
+		}
+
+		if err := writer.Write(rowData); err != nil {
+			result.Success = false
+			result.Error = fmt.Sprintf("Failed to write CSV row: %v", err)
+			result.Time = time.Since(startTime).String()
+			return result
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Row iteration error: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	writer.Flush()
+	if err := writer.Error(); err != nil {
+		result.Success = false
+		result.Error = fmt.Sprintf("Failed to flush CSV: %v", err)
+		result.Time = time.Since(startTime).String()
+		return result
+	}
+
+	result.Success = true
+	result.Data = map[string]interface{}{
+		"csv":           csvBuilder.String(),
+		"count":         count,
+		"includeHeader": includeHeader,
+	}
+	result.Count = count
+	result.Time = time.Since(startTime).String()
+
+	return result
+}
+
+// ExecuteQueryCSVStream 流式返回CSV数据
+func (e *QueryExecutor) ExecuteQueryCSVStream(sql string, w io.Writer, includeHeader bool) (int, error) {
+	rows, err := e.db.Query(sql)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+
+	columns, err := rows.Columns()
+	if err != nil {
+		return 0, err
+	}
+
+	writer := csv.NewWriter(w)
+	count := 0
+
+	if includeHeader {
+		if err := writer.Write(columns); err != nil {
+			return 0, err
+		}
+	}
+
+	for rows.Next() {
+		count++
+
+		values := make([]interface{}, len(columns))
+		valuePtrs := make([]interface{}, len(columns))
+		for i := range columns {
+			valuePtrs[i] = &values[i]
+		}
+
+		if err := rows.Scan(valuePtrs...); err != nil {
+			return count, err
+		}
+
+		rowData := make([]string, len(columns))
+		for i, val := range values {
+			if val == nil {
+				rowData[i] = ""
+				continue
+			}
+			rowData[i] = fmt.Sprintf("%v", val)
+		}
+
+		if err := writer.Write(rowData); err != nil {
+			return count, err
+		}
+	}
+
+	writer.Flush()
+	if err := writer.Error(); err != nil {
+		return count, err
+	}
+
+	if err := rows.Err(); err != nil {
+		return count, err
+	}
+
+	return count, nil
+}

go.mod

@@ -0,0 +1,21 @@
+module git.x2erp.com/qdy/go-db
+
+go 1.25.4
+
+require (
+	github.com/go-sql-driver/mysql v1.9.3
+	github.com/lib/pq v1.10.9
+	github.com/microsoft/go-mssqldb v1.9.4
+	github.com/sijms/go-ora/v2 v2.9.0
+	gopkg.in/yaml.v2 v2.4.0
+)
+
+require (
+	filippo.io/edwards25519 v1.1.0 // indirect
+	git.x2erp.com/qdy/go-base v0.1.1
+	github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 // indirect
+	github.com/golang-sql/sqlexp v0.1.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	golang.org/x/crypto v0.38.0 // indirect
+	golang.org/x/text v0.25.0 // indirect
+)

go.sum

@@ -0,0 +1,56 @@
+filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
+filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
+git.x2erp.com/qdy/go-base v0.1.1 h1:Yj6rFTCL9CShqdQeE6mUYQMugUpI25nWNL0i09SFtXo=
+git.x2erp.com/qdy/go-base v0.1.1/go.mod h1:+NdHouWcxqex8sJUwDTGSl/OIh5fKOT6tJteefD1MMA=
+github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0 h1:Gt0j3wceWMwPmiazCa8MzMA0MfhmPIz0Qp0FJ6qcM0U=
+github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0/go.mod h1:Ot/6aikWnKWi4l9QB7qVSwa8iMphQNqkWALMoNT3rzM=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.10.1 h1:B+blDbyVIG3WaikNxPnhPiJ1MThR03b3vKGtER95TP4=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.10.1/go.mod h1:JdM5psgjfBf5fo2uWOZhflPWyDBZ/O/CNAH9CtsuZE4=
+github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1 h1:FPKJS1T+clwv+OLGt13a8UjqeRuh0O4SJ3lUriThc+4=
+github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1/go.mod h1:j2chePtV91HrC22tGoRX3sGY42uF13WzmmV80/OdVAA=
+github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azkeys v1.3.1 h1:Wgf5rZba3YZqeTNJPtvqZoBu1sBN/L4sry+u2U3Y75w=
+github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azkeys v1.3.1/go.mod h1:xxCBG/f/4Vbmh2XQJBsOmNdxWUY5j/s27jujKPbQf14=
+github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/internal v1.1.1 h1:bFWuoEKg+gImo7pvkiQEFAc8ocibADgXeiLAxWhWmkI=
+github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/internal v1.1.1/go.mod h1:Vih/3yc6yac2JzU4hzpaDupBJP0Flaia9rXXrU8xyww=
+github.com/AzureAD/microsoft-authentication-library-for-go v1.4.2 h1:oygO0locgZJe7PpYPXT5A29ZkwJaPqcva7BVeemZOZs=
+github.com/AzureAD/microsoft-authentication-library-for-go v1.4.2/go.mod h1:wP83P5OoQ5p6ip3ScPr0BAq0BvuPAvacpEuSzyouqAI=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo=
+github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU=
+github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8=
+github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
+github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 h1:au07oEsX2xN0ktxqI+Sida1w446QrXBRJ0nee3SNZlA=
+github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0=
+github.com/golang-sql/sqlexp v0.1.0 h1:ZCD6MBpcuOVfGVqsEmY5/4FtYiKz6tSyUv9LPEDei6A=
+github.com/golang-sql/sqlexp v0.1.0/go.mod h1:J4ad9Vo8ZCWQ2GMrC4UCQy1JpCbwU9m3EOqtpKwwwHI=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/microsoft/go-mssqldb v1.9.4 h1:sHrj3GcdgkxytZ09aZ3+ys72pMeyEXJowT44j74pNgs=
+github.com/microsoft/go-mssqldb v1.9.4/go.mod h1:GBbW9ASTiDC+mpgWDGKdm3FnFLTUsLYN3iFL90lQ+PA=
+github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
+github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sijms/go-ora/v2 v2.9.0 h1:+iQbUeTeCOFMb5BsOMgUhV8KWyrv9yjKpcK4x7+MFrg=
+github.com/sijms/go-ora/v2 v2.9.0/go.mod h1:QgFInVi3ZWyqAiJwzBQA+nbKYKH77tdp1PYoCqhR2dU=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8=
+golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw=
+golang.org/x/net v0.40.0 h1:79Xs7wF06Gbdcg4kdCCIQArK11Z1hr5POQ6+fIYHNuY=
+golang.org/x/net v0.40.0/go.mod h1:y0hY0exeL2Pku80/zKK7tpntoX23cqL3Oa6njdgRtds=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
+golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

test.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+
+	"git.x2erp.com/qdy/go-db/factory"
+)
+
+func main() {
+	// 创建数据库工厂
+	dbFactory, err := factory.NewDBFactory()
+	if err != nil {
+		log.Fatalf("Failed to create DB factory: %v", err)
+	}
+
+	// 显示可用的数据库驱动
+	drivers := dbFactory.GetAvailableDrivers()
+	fmt.Printf("Available database drivers: %v\n", drivers)
+
+	// 显示当前使用的数据库配置
+	config := dbFactory.GetConfig()
+	fmt.Printf("Using database type: %s\n", config.Database.Type)
+	fmt.Printf("Database host: %s:%d\n", config.Database.Host, config.Database.Port) // 修正这里
+	fmt.Printf("Database name: %s\n", config.Database.Database)                      // 修正这里
+
+	// 创建数据库连接
+	db, err := dbFactory.CreateDB()
+	if err != nil {
+		log.Fatalf("Failed to create database connection: %v", err)
+	}
+	defer db.Close()
+
+	fmt.Println("Successfully connected to database!")
+
+	// 测试连接
+	if err := testConnection(db, config.Database.Type); err != nil { // 修正这里
+		log.Printf("Query test failed: %v", err)
+	} else {
+		fmt.Println("Database connection test passed!")
+	}
+}
+
+func testConnection(db *sql.DB, dbType string) error {
+	var query string
+	switch dbType {
+	case "mysql", "postgres", "sqlserver":
+		query = "SELECT 1"
+	case "oracle":
+		query = "SELECT 1 FROM DUAL"
+	default:
+		query = "SELECT 1"
+	}
+
+	var result int
+	err := db.QueryRow(query).Scan(&result)
+	if err != nil {
+		return fmt.Errorf("test query failed: %v", err)
+	}
+
+	if result != 1 {
+		return fmt.Errorf("unexpected test result: %d", result)
+	}
+
+	return nil
+}