feature: Datasource 抽象 (#170)

Author: Stone-afk

.CHANGELOG.md

@@ -16,6 +16,7 @@
 - [eorm: 分库分表: Merger抽象与批量查询实现](https://github.com/ecodeclub/eorm/pull/160)
 - [eorm: 增强的 ShardingAlgorithm 设计与实现](https://github.com/ecodeclub/eorm/pull/161)
 - [eorm: 分库分表: Merger排序实现](https://github.com/ecodeclub/eorm/pull/166)
+- [eorm: Datasource 抽象](https://github.com/ecodeclub/eorm/pull/167)
 - [eorm: 分库分表: Merger分页实现](https://github.com/ecodeclub/eorm/pull/175)
 - [eorm: BasicTypeValue重命名](https://github.com/ecodeclub/eorm/pull/177)
 - [eorm: 分库分表: hash、shadow_hash算法不符合预期](https://github.com/ecodeclub/eorm/pull/174)

builder.go

@@ -18,20 +18,22 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/ecodeclub/eorm/internal/datasource"
+
 	"github.com/ecodeclub/eorm/internal/errs"
 	"github.com/ecodeclub/eorm/internal/model"
+	"github.com/ecodeclub/eorm/internal/query"
 	"github.com/valyala/bytebufferpool"
 )
 
 var _ Executor = &Inserter[any]{}
 var _ Executor = &Updater[any]{}
 var _ Executor = &Deleter[any]{}
 
+var EmptyQuery = Query{}
+
 // Query 代表一个查询
-type Query struct {
-	SQL  string
-	Args []any
-}
+type Query = query.Query
 
 // Querier 查询器，代表最基本的查询
 type Querier[T any] struct {
@@ -48,7 +50,7 @@ func RawQuery[T any](sess Session, sql string, args ...any) Querier[T] {
 		core:    sess.getCore(),
 		Session: sess,
 		qc: &QueryContext{
-			q: &Query{
+			q: Query{
 				SQL:  sql,
 				Args: args,
 			},
@@ -57,7 +59,7 @@ func RawQuery[T any](sess Session, sql string, args ...any) Querier[T] {
 	}
 }
 
-func newQuerier[T any](sess Session, q *Query, meta *model.TableMeta, typ string) Querier[T] {
+func newQuerier[T any](sess Session, q Query, meta *model.TableMeta, typ string) Querier[T] {
 	return Querier[T]{
 		core:    sess.getCore(),
 		Session: sess,
@@ -72,7 +74,7 @@ func newQuerier[T any](sess Session, q *Query, meta *model.TableMeta, typ string
 // Exec 执行 SQL
 func (q Querier[T]) Exec(ctx context.Context) Result {
 	var handler HandleFunc = func(ctx context.Context, qc *QueryContext) *QueryResult {
-		res, err := q.Session.execContext(ctx, qc.q.SQL, qc.q.Args...)
+		res, err := q.Session.execContext(ctx, datasource.Query(qc.q))
 		return &QueryResult{Result: res, Err: err}
 	}
 
@@ -326,16 +328,16 @@ func (b *builder) buildColumn(c Column) error {
 // buildSubquery 構建子查詢 SQL，
 // useAlias 決定是否顯示別名，即使有別名
 func (b *builder) buildSubquery(sub Subquery, useAlias bool) error {
-	query, err := sub.q.Build()
+	q, err := sub.q.Build()
 	if err != nil {
 		return err
 	}
 	b.writeByte('(')
 	// 拿掉最後 ';'
-	b.writeString(query.SQL[:len(query.SQL)-1])
+	b.writeString(q.SQL[:len(q.SQL)-1])
 	// 因為有 build() ，所以理應 args 也需要跟 SQL 一起處理
-	if len(query.Args) > 0 {
-		b.addArgs(query.Args...)
+	if len(q.Args) > 0 {
+		b.addArgs(q.Args...)
 	}
 	b.writeByte(')')
 	if useAlias {

builder_test.go

@@ -21,15 +21,17 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/ecodeclub/eorm/internal/datasource/single"
+
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/ecodeclub/eorm/internal/errs"
 	"github.com/ecodeclub/eorm/internal/valuer"
 	"github.com/stretchr/testify/assert"
 )
 
 func ExampleRawQuery() {
-	orm := memoryDB()
-	q := RawQuery[any](orm, `SELECT * FROM user_tab WHERE id = ?;`, 1)
+	db := memoryDB()
+	q := RawQuery[any](db, `SELECT * FROM user_tab WHERE id = ?;`, 1)
 	fmt.Printf(`
 SQL: %s
 Args: %v
@@ -40,9 +42,9 @@ Args: %v
 }
 
 func ExampleQuerier_Exec() {
-	orm := memoryDB()
+	db := memoryDB()
 	// 在 Exec 的时候，泛型参数可以是任意的
-	q := RawQuery[any](orm, `CREATE TABLE IF NOT EXISTS groups (
+	q := RawQuery[any](db, `CREATE TABLE IF NOT EXISTS groups (
    group_id INTEGER PRIMARY KEY,
    name TEXT NOT NULL
 )`)
@@ -54,10 +56,6 @@ func ExampleQuerier_Exec() {
 	// SUCCESS
 }
 
-func (q Query) string() string {
-	return fmt.Sprintf("SQL: %s\nArgs: %#v\n", q.SQL, q.Args)
-}
-
 func TestQuerier_Get(t *testing.T) {
 	t.Run("unsafe", func(t *testing.T) {
 		testQuerierGet(t, valuer.PrimitiveCreator{Creator: valuer.NewUnsafeValue})
@@ -75,7 +73,7 @@ func testQuerierGet(t *testing.T, creator valuer.PrimitiveCreator) {
 	}
 	defer func() { _ = db.Close() }()
 
-	orm, err := openDB("mysql", db)
+	orm, err := OpenDS("mysql", single.NewDB(db))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -165,7 +163,7 @@ func testQuerier_GetMulti(t *testing.T, creator valuer.PrimitiveCreator) {
 	defer func() {
 		_ = db.Close()
 	}()
-	orm, err := openDB("mysql", db)
+	orm, err := OpenDS("mysql", single.NewDB(db))
 	if err != nil {
 		t.Fatal(err)
 	}

core.go

@@ -32,7 +32,7 @@ type core struct {
 }
 
 func getHandler[T any](ctx context.Context, sess Session, c core, qc *QueryContext) *QueryResult {
-	rows, err := sess.queryContext(ctx, qc.q.SQL, qc.q.Args...)
+	rows, err := sess.queryContext(ctx, qc.q)
 	if err != nil {
 		return &QueryResult{Err: err}
 	}
@@ -68,7 +68,7 @@ func get[T any](ctx context.Context, sess Session, core core, qc *QueryContext)
 }
 
 func getMultiHandler[T any](ctx context.Context, sess Session, c core, qc *QueryContext) *QueryResult {
-	rows, err := sess.queryContext(ctx, qc.q.SQL, qc.q.Args...)
+	rows, err := sess.queryContext(ctx, qc.q)
 	if err != nil {
 		return &QueryResult{Err: err}
 	}

db.go

@@ -17,11 +17,11 @@ package eorm
 import (
 	"context"
 	"database/sql"
-	"database/sql/driver"
-	"log"
-	"time"
 
+	"github.com/ecodeclub/eorm/internal/datasource"
+	"github.com/ecodeclub/eorm/internal/datasource/single"
 	"github.com/ecodeclub/eorm/internal/dialect"
+	"github.com/ecodeclub/eorm/internal/errs"
 	"github.com/ecodeclub/eorm/internal/model"
 	"github.com/ecodeclub/eorm/internal/valuer"
 )
@@ -39,8 +39,8 @@ type DBOption func(db *DB)
 
 // DB represents a database
 type DB struct {
-	db *sql.DB
 	core
+	ds datasource.DataSource
 }
 
 // DBWithMiddlewares 为 db 配置 Middleware
@@ -50,31 +50,37 @@ func DBWithMiddlewares(ms ...Middleware) DBOption {
 	}
 }
 
+func DBOptionWithMetaRegistry(r model.MetaRegistry) DBOption {
+	return func(db *DB) {
+		db.metaRegistry = r
+	}
+}
+
 func UseReflection() DBOption {
 	return func(db *DB) {
 		db.valCreator = valuer.PrimitiveCreator{Creator: valuer.NewUnsafeValue}
 	}
 }
 
-func (db *DB) queryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
-	return db.db.QueryContext(ctx, query, args...)
+func (db *DB) queryContext(ctx context.Context, q datasource.Query) (*sql.Rows, error) {
+	return db.ds.Query(ctx, q)
 }
 
-func (db *DB) execContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
-	return db.db.ExecContext(ctx, query, args...)
+func (db *DB) execContext(ctx context.Context, q datasource.Query) (sql.Result, error) {
+	return db.ds.Exec(ctx, q)
 }
 
 // Open 创建一个 ORM 实例
 // 注意该实例是一个无状态的对象，你应该尽可能复用它
 func Open(driver string, dsn string, opts ...DBOption) (*DB, error) {
-	db, err := sql.Open(driver, dsn)
+	db, err := single.OpenDB(driver, dsn)
 	if err != nil {
 		return nil, err
 	}
-	return openDB(driver, db, opts...)
+	return OpenDS(driver, db, opts...)
 }
 
-func openDB(driver string, db *sql.DB, opts ...DBOption) (*DB, error) {
+func OpenDS(driver string, ds datasource.DataSource, opts ...DBOption) (*DB, error) {
 	dl, err := dialect.Of(driver)
 	if err != nil {
 		return nil, err
@@ -88,37 +94,28 @@ func openDB(driver string, db *sql.DB, opts ...DBOption) (*DB, error) {
 				Creator: valuer.NewUnsafeValue,
 			},
 		},
-		db: db,
+		ds: ds,
 	}
 	for _, o := range opts {
 		o(orm)
 	}
 	return orm, nil
 }
 
-// BeginTx 开启事务
 func (db *DB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*Tx, error) {
-	tx, err := db.db.BeginTx(ctx, opts)
+	inst, ok := db.ds.(datasource.TxBeginner)
+	if !ok {
+		return nil, errs.ErrNotCompleteTxBeginner
+	}
+	tx, err := inst.BeginTx(ctx, opts)
 	if err != nil {
 		return nil, err
 	}
-	return &Tx{tx: tx, db: db.db, core: db.core}, nil
-}
-
-// Wait 会等待数据库连接
-// 注意只能用于测试
-func (db *DB) Wait() error {
-	err := db.db.Ping()
-	for err == driver.ErrBadConn {
-		log.Printf("等待数据库启动...")
-		err = db.db.Ping()
-		time.Sleep(time.Second)
-	}
-	return err
+	return &Tx{tx: tx, core: db.getCore()}, nil
 }
 
 func (db *DB) Close() error {
-	return db.db.Close()
+	return db.ds.Close()
 }
 
 func (db *DB) getCore() core {