server/beego
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add context support for orm

Browse Source 
Signed-off-by: Penghui Liao <liaoishere@gmail.com>
This commit is contained in:
Penghui Liao 2021-01-08 19:04:00 +08:00
parent c603131436
commit 21777d3143
15 changed files with 215 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
client/orm/cmd.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"flag" "flag"
"fmt" "fmt"
"os" "os"
@ -76,6 +77,7 @@ func RunCommand() {
// sync database struct command interface. // sync database struct command interface.
type commandSyncDb struct { type commandSyncDb struct {
ctx context.Context
al *alias al *alias
force bool force bool
verbose bool verbose bool
@ -154,7 +156,7 @@ func (d *commandSyncDb) Run() error {
} }
var fields []*fieldInfo var fields []*fieldInfo
columns, err := d.al.DbBaser.GetColumns(db, mi.table) columns, err := d.al.DbBaser.GetColumns(d.ctx, db, mi.table)
if err != nil { if err != nil {
if d.rtOnError { if d.rtOnError {
return err return err
@ -188,7 +190,7 @@ func (d *commandSyncDb) Run() error {
} }
for _, idx := range indexes[mi.table] { for _, idx := range indexes[mi.table] {
if !d.al.DbBaser.IndexExists(db, idx.Table, idx.Name) { if !d.al.DbBaser.IndexExists(d.ctx, db, idx.Table, idx.Name) {
if !d.noInfo { if !d.noInfo {
fmt.Printf("create index `%s` for table `%s`\n", idx.Name, idx.Table) fmt.Printf("create index `%s` for table `%s`\n", idx.Name, idx.Table)
} }
@ -290,6 +292,7 @@ func RunSyncdb(name string, force bool, verbose bool) error {
al := getDbAlias(name) al := getDbAlias(name)
cmd := new(commandSyncDb) cmd := new(commandSyncDb)
cmd.ctx = context.TODO()
cmd.al = al cmd.al = al
cmd.force = force cmd.force = force
cmd.noInfo = !verbose cmd.noInfo = !verbose

114
client/orm/db.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"database/sql" "database/sql"
"errors" "errors"
"fmt" "fmt"
@ -268,7 +269,7 @@ func (d *dbBase) collectFieldValue(mi *modelInfo, fi *fieldInfo, ind reflect.Val
} }
// create insert sql preparation statement object. // create insert sql preparation statement object.
func (d *dbBase) PrepareInsert(q dbQuerier, mi *modelInfo) (stmtQuerier, string, error) { func (d *dbBase) PrepareInsert(ctx context.Context, q dbQuerier, mi *modelInfo) (stmtQuerier, string, error) {
Q := d.ins.TableQuote() Q := d.ins.TableQuote()
dbcols := make([]string, 0, len(mi.fields.dbcols)) dbcols := make([]string, 0, len(mi.fields.dbcols))
@ -289,12 +290,12 @@ func (d *dbBase) PrepareInsert(q dbQuerier, mi *modelInfo) (stmtQuerier, string,
d.ins.HasReturningID(mi, &query) d.ins.HasReturningID(mi, &query)
stmt, err := q.Prepare(query) stmt, err := q.PrepareContext(ctx, query)
return stmt, query, err return stmt, query, err
} }
// insert struct with prepared statement and given struct reflect value. // insert struct with prepared statement and given struct reflect value.
func (d *dbBase) InsertStmt(stmt stmtQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location) (int64, error) { func (d *dbBase) InsertStmt(ctx context.Context, stmt stmtQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location) (int64, error) {
values, _, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, nil, tz) values, _, err := d.collectValues(mi, ind, mi.fields.dbcols, true, true, nil, tz)
if err != nil { if err != nil {
return 0, err return 0, err
@ -306,7 +307,7 @@ func (d *dbBase) InsertStmt(stmt stmtQuerier, mi *modelInfo, ind reflect.Value,
err := row.Scan(&id) err := row.Scan(&id)
return id, err return id, err
} }
res, err := stmt.Exec(values...) res, err := stmt.ExecContext(ctx, values...)
if err == nil { if err == nil {
return res.LastInsertId() return res.LastInsertId()
} }
@ -314,7 +315,7 @@ func (d *dbBase) InsertStmt(stmt stmtQuerier, mi *modelInfo, ind reflect.Value,
} }
// query sql ,read records and persist in dbBaser. // query sql ,read records and persist in dbBaser.
func (d *dbBase) Read(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string, isForUpdate bool) error { func (d *dbBase) Read(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string, isForUpdate bool) error {
var whereCols []string var whereCols []string
var args []interface{} var args []interface{}
@ -360,7 +361,7 @@ func (d *dbBase) Read(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Lo
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
row := q.QueryRow(query, args...) row := q.QueryRowContext(ctx, query, args...)
if err := row.Scan(refs...); err != nil { if err := row.Scan(refs...); err != nil {
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
return ErrNoRows return ErrNoRows
@ -375,26 +376,26 @@ func (d *dbBase) Read(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Lo
} }
// execute insert sql dbQuerier with given struct reflect.Value. // execute insert sql dbQuerier with given struct reflect.Value.
func (d *dbBase) Insert(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location) (int64, error) { func (d *dbBase) Insert(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location) (int64, error) {
names := make([]string, 0, len(mi.fields.dbcols)) names := make([]string, 0, len(mi.fields.dbcols))
values, autoFields, err := d.collectValues(mi, ind, mi.fields.dbcols, false, true, &names, tz) values, autoFields, err := d.collectValues(mi, ind, mi.fields.dbcols, false, true, &names, tz)
if err != nil { if err != nil {
return 0, err return 0, err
} }
id, err := d.InsertValue(q, mi, false, names, values) id, err := d.InsertValue(ctx, q, mi, false, names, values)
if err != nil { if err != nil {
return 0, err return 0, err
} }
if len(autoFields) > 0 { if len(autoFields) > 0 {
err = d.ins.setval(q, mi, autoFields) err = d.ins.setval(ctx, q, mi, autoFields)
} }
return id, err return id, err
} }
// multi-insert sql with given slice struct reflect.Value. // multi-insert sql with given slice struct reflect.Value.
func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bulk int, tz *time.Location) (int64, error) { func (d *dbBase) InsertMulti(ctx context.Context, q dbQuerier, mi *modelInfo, sind reflect.Value, bulk int, tz *time.Location) (int64, error) {
var ( var (
cnt int64 cnt int64
nums int nums int
@ -440,7 +441,7 @@ func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bul
} }
if i > 1 && i%bulk == 0 || length == i { if i > 1 && i%bulk == 0 || length == i {
num, err := d.InsertValue(q, mi, true, names, values[:nums]) num, err := d.InsertValue(ctx, q, mi, true, names, values[:nums])
if err != nil { if err != nil {
return cnt, err return cnt, err
} }
@ -451,7 +452,7 @@ func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bul
var err error var err error
if len(autoFields) > 0 { if len(autoFields) > 0 {
err = d.ins.setval(q, mi, autoFields) err = d.ins.setval(ctx, q, mi, autoFields)
} }
return cnt, err return cnt, err
@ -459,7 +460,7 @@ func (d *dbBase) InsertMulti(q dbQuerier, mi *modelInfo, sind reflect.Value, bul
// execute insert sql with given struct and given values. // execute insert sql with given struct and given values.
// insert the given values, not the field values in struct. // insert the given values, not the field values in struct.
func (d *dbBase) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, names []string, values []interface{}) (int64, error) { func (d *dbBase) InsertValue(ctx context.Context, q dbQuerier, mi *modelInfo, isMulti bool, names []string, values []interface{}) (int64, error) {
Q := d.ins.TableQuote() Q := d.ins.TableQuote()
marks := make([]string, len(names)) marks := make([]string, len(names))
@ -482,7 +483,7 @@ func (d *dbBase) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, names []s
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
if isMulti || !d.ins.HasReturningID(mi, &query) { if isMulti || !d.ins.HasReturningID(mi, &query) {
res, err := q.Exec(query, values...) res, err := q.ExecContext(ctx, query, values...)
if err == nil { if err == nil {
if isMulti { if isMulti {
return res.RowsAffected() return res.RowsAffected()
@ -498,7 +499,7 @@ func (d *dbBase) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, names []s
} }
return 0, err return 0, err
} }
row := q.QueryRow(query, values...) row := q.QueryRowContext(ctx, query, values...)
var id int64 var id int64
err := row.Scan(&id) err := row.Scan(&id)
return id, err return id, err
@ -507,7 +508,7 @@ func (d *dbBase) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, names []s
// InsertOrUpdate a row // InsertOrUpdate a row
// If your primary key or unique column conflict will update // If your primary key or unique column conflict will update
// If no will insert // If no will insert
func (d *dbBase) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Value, a *alias, args ...string) (int64, error) { func (d *dbBase) InsertOrUpdate(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, a *alias, args ...string) (int64, error) {
args0 := "" args0 := ""
iouStr := "" iouStr := ""
argsMap := map[string]string{} argsMap := map[string]string{}
@ -590,7 +591,7 @@ func (d *dbBase) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Value, a
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
if isMulti || !d.ins.HasReturningID(mi, &query) { if isMulti || !d.ins.HasReturningID(mi, &query) {
res, err := q.Exec(query, values...) res, err := q.ExecContext(ctx, query, values...)
if err == nil { if err == nil {
if isMulti { if isMulti {
return res.RowsAffected() return res.RowsAffected()
@ -607,7 +608,7 @@ func (d *dbBase) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Value, a
return 0, err return 0, err
} }
row := q.QueryRow(query, values...) row := q.QueryRowContext(ctx, query, values...)
var id int64 var id int64
err = row.Scan(&id) err = row.Scan(&id)
if err != nil && err.Error() == `pq: syntax error at or near "ON"` { if err != nil && err.Error() == `pq: syntax error at or near "ON"` {
@ -617,7 +618,7 @@ func (d *dbBase) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Value, a
} }
// execute update sql dbQuerier with given struct reflect.Value. // execute update sql dbQuerier with given struct reflect.Value.
func (d *dbBase) Update(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string) (int64, error) { func (d *dbBase) Update(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string) (int64, error) {
pkName, pkValue, ok := getExistPk(mi, ind) pkName, pkValue, ok := getExistPk(mi, ind)
if !ok { if !ok {
return 0, ErrMissPK return 0, ErrMissPK
@ -674,7 +675,7 @@ func (d *dbBase) Update(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
res, err := q.Exec(query, setValues...) res, err := q.ExecContext(ctx, query, setValues...)
if err == nil { if err == nil {
return res.RowsAffected() return res.RowsAffected()
} }
@ -683,7 +684,7 @@ func (d *dbBase) Update(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.
// execute delete sql dbQuerier with given struct reflect.Value. // execute delete sql dbQuerier with given struct reflect.Value.
// delete index is pk. // delete index is pk.
func (d *dbBase) Delete(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string) (int64, error) { func (d *dbBase) Delete(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string) (int64, error) {
var whereCols []string var whereCols []string
var args []interface{} var args []interface{}
// if specify cols length > 0, then use it for where condition. // if specify cols length > 0, then use it for where condition.
@ -712,7 +713,7 @@ func (d *dbBase) Delete(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.
query := fmt.Sprintf("DELETE FROM %s%s%s WHERE %s%s%s = ?", Q, mi.table, Q, Q, wheres, Q) query := fmt.Sprintf("DELETE FROM %s%s%s WHERE %s%s%s = ?", Q, mi.table, Q, Q, wheres, Q)
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
res, err := q.Exec(query, args...) res, err := q.ExecContext(ctx, query, args...)
if err == nil { if err == nil {
num, err := res.RowsAffected() num, err := res.RowsAffected()
if err != nil { if err != nil {
@ -726,7 +727,7 @@ func (d *dbBase) Delete(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.
ind.FieldByIndex(mi.fields.pk.fieldIndex).SetInt(0) ind.FieldByIndex(mi.fields.pk.fieldIndex).SetInt(0)
} }
} }
err := d.deleteRels(q, mi, args, tz) err := d.deleteRels(ctx, q, mi, args, tz)
if err != nil { if err != nil {
return num, err return num, err
} }
@ -738,7 +739,7 @@ func (d *dbBase) Delete(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.
// update table-related record by querySet. // update table-related record by querySet.
// need querySet not struct reflect.Value to update related records. // need querySet not struct reflect.Value to update related records.
func (d *dbBase) UpdateBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, params Params, tz *time.Location) (int64, error) { func (d *dbBase) UpdateBatch(ctx context.Context, q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, params Params, tz *time.Location) (int64, error) {
columns := make([]string, 0, len(params)) columns := make([]string, 0, len(params))
values := make([]interface{}, 0, len(params)) values := make([]interface{}, 0, len(params))
for col, val := range params { for col, val := range params {
@ -819,13 +820,7 @@ func (d *dbBase) UpdateBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Con
} }
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
var err error res, err := q.ExecContext(ctx, query, values...)
var res sql.Result
if qs != nil && qs.forContext {
res, err = q.ExecContext(qs.ctx, query, values...)
} else {
res, err = q.Exec(query, values...)
}
if err == nil { if err == nil {
return res.RowsAffected() return res.RowsAffected()
} }
@ -834,13 +829,13 @@ func (d *dbBase) UpdateBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Con
// delete related records. // delete related records.
// do UpdateBanch or DeleteBanch by condition of tables' relationship. // do UpdateBanch or DeleteBanch by condition of tables' relationship.
func (d *dbBase) deleteRels(q dbQuerier, mi *modelInfo, args []interface{}, tz *time.Location) error { func (d *dbBase) deleteRels(ctx context.Context, q dbQuerier, mi *modelInfo, args []interface{}, tz *time.Location) error {
for _, fi := range mi.fields.fieldsReverse { for _, fi := range mi.fields.fieldsReverse {
fi = fi.reverseFieldInfo fi = fi.reverseFieldInfo
switch fi.onDelete { switch fi.onDelete {
case odCascade: case odCascade:
cond := NewCondition().And(fmt.Sprintf("%s__in", fi.name), args...) cond := NewCondition().And(fmt.Sprintf("%s__in", fi.name), args...)
_, err := d.DeleteBatch(q, nil, fi.mi, cond, tz) _, err := d.DeleteBatch(ctx, q, nil, fi.mi, cond, tz)
if err != nil { if err != nil {
return err return err
} }
@ -850,7 +845,7 @@ func (d *dbBase) deleteRels(q dbQuerier, mi *modelInfo, args []interface{}, tz *
if fi.onDelete == odSetDefault { if fi.onDelete == odSetDefault {
params[fi.column] = fi.initial.String() params[fi.column] = fi.initial.String()
} }
_, err := d.UpdateBatch(q, nil, fi.mi, cond, params, tz) _, err := d.UpdateBatch(ctx, q, nil, fi.mi, cond, params, tz)
if err != nil { if err != nil {
return err return err
} }
@ -861,7 +856,7 @@ func (d *dbBase) deleteRels(q dbQuerier, mi *modelInfo, args []interface{}, tz *
} }
// delete table-related records. // delete table-related records.
func (d *dbBase) DeleteBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, tz *time.Location) (int64, error) { func (d *dbBase) DeleteBatch(ctx context.Context, q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, tz *time.Location) (int64, error) {
tables := newDbTables(mi, d.ins) tables := newDbTables(mi, d.ins)
tables.skipEnd = true tables.skipEnd = true
@ -886,7 +881,7 @@ func (d *dbBase) DeleteBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Con
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
var rs *sql.Rows var rs *sql.Rows
r, err := q.Query(query, args...) r, err := q.QueryContext(ctx, query, args...)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@ -920,19 +915,14 @@ func (d *dbBase) DeleteBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Con
query = fmt.Sprintf("DELETE FROM %s%s%s WHERE %s%s%s %s", Q, mi.table, Q, Q, mi.fields.pk.column, Q, sqlIn) query = fmt.Sprintf("DELETE FROM %s%s%s WHERE %s%s%s %s", Q, mi.table, Q, Q, mi.fields.pk.column, Q, sqlIn)
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
var res sql.Result res, err := q.ExecContext(ctx, query, args...)
if qs != nil && qs.forContext {
res, err = q.ExecContext(qs.ctx, query, args...)
} else {
res, err = q.Exec(query, args...)
}
if err == nil { if err == nil {
num, err := res.RowsAffected() num, err := res.RowsAffected()
if err != nil { if err != nil {
return 0, err return 0, err
} }
if num > 0 { if num > 0 {
err := d.deleteRels(q, mi, args, tz) err := d.deleteRels(ctx, q, mi, args, tz)
if err != nil { if err != nil {
return num, err return num, err
} }
@ -943,7 +933,7 @@ func (d *dbBase) DeleteBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Con
} }
// read related records. // read related records.
func (d *dbBase) ReadBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, container interface{}, tz *time.Location, cols []string) (int64, error) { func (d *dbBase) ReadBatch(ctx context.Context, q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, container interface{}, tz *time.Location, cols []string) (int64, error) {
val := reflect.ValueOf(container) val := reflect.ValueOf(container)
ind := reflect.Indirect(val) ind := reflect.Indirect(val)
@ -1052,18 +1042,9 @@ func (d *dbBase) ReadBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condi
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
var rs *sql.Rows rs, err := q.QueryContext(ctx, query, args...)
var err error if err != nil {
if qs != nil && qs.forContext { return 0, err
rs, err = q.QueryContext(qs.ctx, query, args...)
if err != nil {
return 0, err
}
} else {
rs, err = q.Query(query, args...)
if err != nil {
return 0, err
}
} }
defer rs.Close() defer rs.Close()
@ -1178,7 +1159,7 @@ func (d *dbBase) ReadBatch(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condi
} }
// excute count sql and return count result int64. // excute count sql and return count result int64.
func (d *dbBase) Count(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, tz *time.Location) (cnt int64, err error) { func (d *dbBase) Count(ctx context.Context, q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, tz *time.Location) (cnt int64, err error) {
tables := newDbTables(mi, d.ins) tables := newDbTables(mi, d.ins)
tables.parseRelated(qs.related, qs.relDepth) tables.parseRelated(qs.related, qs.relDepth)
@ -1200,12 +1181,7 @@ func (d *dbBase) Count(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
var row *sql.Row row := q.QueryRowContext(ctx, query, args...)
if qs != nil && qs.forContext {
row = q.QueryRowContext(qs.ctx, query, args...)
} else {
row = q.QueryRow(query, args...)
}
err = row.Scan(&cnt) err = row.Scan(&cnt)
return return
} }
@ -1655,7 +1631,7 @@ setValue:
} }
// query sql, read values , save to *[]ParamList. // query sql, read values , save to *[]ParamList.
func (d *dbBase) ReadValues(q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, exprs []string, container interface{}, tz *time.Location) (int64, error) { func (d *dbBase) ReadValues(ctx context.Context, q dbQuerier, qs *querySet, mi *modelInfo, cond *Condition, exprs []string, container interface{}, tz *time.Location) (int64, error) {
var ( var (
maps []Params maps []Params
@ -1738,7 +1714,7 @@ func (d *dbBase) ReadValues(q dbQuerier, qs *querySet, mi *modelInfo, cond *Cond
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
rs, err := q.Query(query, args...) rs, err := q.QueryContext(ctx, query, args...)
if err != nil { if err != nil {
return 0, err return 0, err
} }
@ -1853,7 +1829,7 @@ func (d *dbBase) HasReturningID(*modelInfo, *string) bool {
} }
// sync auto key // sync auto key
func (d *dbBase) setval(db dbQuerier, mi *modelInfo, autoFields []string) error { func (d *dbBase) setval(ctx context.Context, db dbQuerier, mi *modelInfo, autoFields []string) error {
return nil return nil
} }
@ -1898,10 +1874,10 @@ func (d *dbBase) GetTables(db dbQuerier) (map[string]bool, error) {
} }
// get all cloumns in table. // get all cloumns in table.
func (d *dbBase) GetColumns(db dbQuerier, table string) (map[string][3]string, error) { func (d *dbBase) GetColumns(ctx context.Context, db dbQuerier, table string) (map[string][3]string, error) {
columns := make(map[string][3]string) columns := make(map[string][3]string)
query := d.ins.ShowColumnsQuery(table) query := d.ins.ShowColumnsQuery(table)
rows, err := db.Query(query) rows, err := db.QueryContext(ctx, query)
if err != nil { if err != nil {
return columns, err return columns, err
} }
@ -1940,7 +1916,7 @@ func (d *dbBase) ShowColumnsQuery(table string) string {
} }
// not implement. // not implement.
func (d *dbBase) IndexExists(dbQuerier, string, string) bool { func (d *dbBase) IndexExists(context.Context, dbQuerier, string, string) bool {
panic(ErrNotImplement) panic(ErrNotImplement)
} }

11
client/orm/db_mysql.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"fmt" "fmt"
"reflect" "reflect"
"strings" "strings"
@ -93,8 +94,8 @@ func (d *dbBaseMysql) ShowColumnsQuery(table string) string {
} }
// execute sql to check index exist. // execute sql to check index exist.
func (d *dbBaseMysql) IndexExists(db dbQuerier, table string, name string) bool { func (d *dbBaseMysql) IndexExists(ctx context.Context, db dbQuerier, table string, name string) bool {
row := db.QueryRow("SELECT count(*) FROM information_schema.statistics "+ row := db.QueryRowContext(ctx, "SELECT count(*) FROM information_schema.statistics "+
"WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?", table, name) "WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?", table, name)
var cnt int var cnt int
row.Scan(&cnt) row.Scan(&cnt)
@ -105,7 +106,7 @@ func (d *dbBaseMysql) IndexExists(db dbQuerier, table string, name string) bool
// If your primary key or unique column conflict will update // If your primary key or unique column conflict will update
// If no will insert // If no will insert
// Add "`" for mysql sql building // Add "`" for mysql sql building
func (d *dbBaseMysql) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Value, a *alias, args ...string) (int64, error) { func (d *dbBaseMysql) InsertOrUpdate(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, a *alias, args ...string) (int64, error) {
var iouStr string var iouStr string
argsMap := map[string]string{} argsMap := map[string]string{}
@ -161,7 +162,7 @@ func (d *dbBaseMysql) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Val
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
if isMulti || !d.ins.HasReturningID(mi, &query) { if isMulti || !d.ins.HasReturningID(mi, &query) {
res, err := q.Exec(query, values...) res, err := q.ExecContext(ctx, query, values...)
if err == nil { if err == nil {
if isMulti { if isMulti {
return res.RowsAffected() return res.RowsAffected()
@ -178,7 +179,7 @@ func (d *dbBaseMysql) InsertOrUpdate(q dbQuerier, mi *modelInfo, ind reflect.Val
return 0, err return 0, err
} }
row := q.QueryRow(query, values...) row := q.QueryRowContext(ctx, query, values...)
var id int64 var id int64
err = row.Scan(&id) err = row.Scan(&id)
return id, err return id, err

11
client/orm/db_oracle.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"fmt" "fmt"
"strings" "strings"
@ -89,8 +90,8 @@ func (d *dbBaseOracle) ShowColumnsQuery(table string) string {
} }
// check index is exist // check index is exist
func (d *dbBaseOracle) IndexExists(db dbQuerier, table string, name string) bool { func (d *dbBaseOracle) IndexExists(ctx context.Context, db dbQuerier, table string, name string) bool {
row := db.QueryRow("SELECT COUNT(*) FROM USER_IND_COLUMNS, USER_INDEXES "+ row := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM USER_IND_COLUMNS, USER_INDEXES "+
"WHERE USER_IND_COLUMNS.INDEX_NAME = USER_INDEXES.INDEX_NAME "+ "WHERE USER_IND_COLUMNS.INDEX_NAME = USER_INDEXES.INDEX_NAME "+
"AND USER_IND_COLUMNS.TABLE_NAME = ? AND USER_IND_COLUMNS.INDEX_NAME = ?", strings.ToUpper(table), strings.ToUpper(name)) "AND USER_IND_COLUMNS.TABLE_NAME = ? AND USER_IND_COLUMNS.INDEX_NAME = ?", strings.ToUpper(table), strings.ToUpper(name))
@ -124,7 +125,7 @@ func (d *dbBaseOracle) GenerateSpecifyIndex(tableName string, useIndex int, inde
// execute insert sql with given struct and given values. // execute insert sql with given struct and given values.
// insert the given values, not the field values in struct. // insert the given values, not the field values in struct.
func (d *dbBaseOracle) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, names []string, values []interface{}) (int64, error) { func (d *dbBaseOracle) InsertValue(ctx context.Context, q dbQuerier, mi *modelInfo, isMulti bool, names []string, values []interface{}) (int64, error) {
Q := d.ins.TableQuote() Q := d.ins.TableQuote()
marks := make([]string, len(names)) marks := make([]string, len(names))
@ -147,7 +148,7 @@ func (d *dbBaseOracle) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, nam
d.ins.ReplaceMarks(&query) d.ins.ReplaceMarks(&query)
if isMulti || !d.ins.HasReturningID(mi, &query) { if isMulti || !d.ins.HasReturningID(mi, &query) {
res, err := q.Exec(query, values...) res, err := q.ExecContext(ctx, query, values...)
if err == nil { if err == nil {
if isMulti { if isMulti {
return res.RowsAffected() return res.RowsAffected()
@ -163,7 +164,7 @@ func (d *dbBaseOracle) InsertValue(q dbQuerier, mi *modelInfo, isMulti bool, nam
} }
return 0, err return 0, err
} }
row := q.QueryRow(query, values...) row := q.QueryRowContext(ctx, query, values...)
var id int64 var id int64
err := row.Scan(&id) err := row.Scan(&id)
return id, err return id, err

9
client/orm/db_postgres.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"fmt" "fmt"
"strconv" "strconv"
) )
@ -140,7 +141,7 @@ func (d *dbBasePostgres) HasReturningID(mi *modelInfo, query *string) bool {
} }
// sync auto key // sync auto key
func (d *dbBasePostgres) setval(db dbQuerier, mi *modelInfo, autoFields []string) error { func (d *dbBasePostgres) setval(ctx context.Context, db dbQuerier, mi *modelInfo, autoFields []string) error {
if len(autoFields) == 0 { if len(autoFields) == 0 {
return nil return nil
} }
@ -151,7 +152,7 @@ func (d *dbBasePostgres) setval(db dbQuerier, mi *modelInfo, autoFields []string
mi.table, name, mi.table, name,
Q, name, Q, Q, name, Q,
Q, mi.table, Q) Q, mi.table, Q)
if _, err := db.Exec(query); err != nil { if _, err := db.ExecContext(ctx, query); err != nil {
return err return err
} }
} }
@ -174,9 +175,9 @@ func (d *dbBasePostgres) DbTypes() map[string]string {
} }
// check index exist in postgresql. // check index exist in postgresql.
func (d *dbBasePostgres) IndexExists(db dbQuerier, table string, name string) bool { func (d *dbBasePostgres) IndexExists(ctx context.Context, db dbQuerier, table string, name string) bool {
query := fmt.Sprintf("SELECT COUNT(*) FROM pg_indexes WHERE tablename = '%s' AND indexname = '%s'", table, name) query := fmt.Sprintf("SELECT COUNT(*) FROM pg_indexes WHERE tablename = '%s' AND indexname = '%s'", table, name)
row := db.QueryRow(query) row := db.QueryRowContext(ctx, query)
var cnt int var cnt int
row.Scan(&cnt) row.Scan(&cnt)
return cnt > 0 return cnt > 0

13
client/orm/db_sqlite.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"database/sql" "database/sql"
"fmt" "fmt"
"reflect" "reflect"
@ -73,11 +74,11 @@ type dbBaseSqlite struct {
var _ dbBaser = new(dbBaseSqlite) var _ dbBaser = new(dbBaseSqlite)
// override base db read for update behavior as SQlite does not support syntax // override base db read for update behavior as SQlite does not support syntax
func (d *dbBaseSqlite) Read(q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string, isForUpdate bool) error { func (d *dbBaseSqlite) Read(ctx context.Context, q dbQuerier, mi *modelInfo, ind reflect.Value, tz *time.Location, cols []string, isForUpdate bool) error {
if isForUpdate { if isForUpdate {
DebugLog.Println("[WARN] SQLite does not support SELECT FOR UPDATE query, isForUpdate param is ignored and always as false to do the work") DebugLog.Println("[WARN] SQLite does not support SELECT FOR UPDATE query, isForUpdate param is ignored and always as false to do the work")
} }
return d.dbBase.Read(q, mi, ind, tz, cols, false) return d.dbBase.Read(ctx, q, mi, ind, tz, cols, false)
} }
// get sqlite operator. // get sqlite operator.
@ -114,9 +115,9 @@ func (d *dbBaseSqlite) ShowTablesQuery() string {
} }
// get columns in sqlite. // get columns in sqlite.
func (d *dbBaseSqlite) GetColumns(db dbQuerier, table string) (map[string][3]string, error) { func (d *dbBaseSqlite) GetColumns(ctx context.Context, db dbQuerier, table string) (map[string][3]string, error) {
query := d.ins.ShowColumnsQuery(table) query := d.ins.ShowColumnsQuery(table)
rows, err := db.Query(query) rows, err := db.QueryContext(ctx, query)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -140,9 +141,9 @@ func (d *dbBaseSqlite) ShowColumnsQuery(table string) string {
} }
// check index exist in sqlite. // check index exist in sqlite.
func (d *dbBaseSqlite) IndexExists(db dbQuerier, table string, name string) bool { func (d *dbBaseSqlite) IndexExists(ctx context.Context, db dbQuerier, table string, name string) bool {
query := fmt.Sprintf("PRAGMA index_list('%s')", table) query := fmt.Sprintf("PRAGMA index_list('%s')", table)
rows, err := db.Query(query) rows, err := db.QueryContext(ctx, query)
if err != nil { if err != nil {
panic(err) panic(err)
} }

5
client/orm/db_tidb.go
View File

@ -15,6 +15,7 @@
package orm package orm
import ( import (
"context"
"fmt" "fmt"
) )
@ -47,8 +48,8 @@ func (d *dbBaseTidb) ShowColumnsQuery(table string) string {
} }
// execute sql to check index exist. // execute sql to check index exist.
func (d *dbBaseTidb) IndexExists(db dbQuerier, table string, name string) bool { func (d *dbBaseTidb) IndexExists(ctx context.Context, db dbQuerier, table string, name string) bool {
row := db.QueryRow("SELECT count(*) FROM information_schema.statistics "+ row := db.QueryRowContext(ctx, "SELECT count(*) FROM information_schema.statistics "+
"WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?", table, name) "WHERE table_schema = DATABASE() AND table_name = ? AND index_name = ?", table, name)
var cnt int var cnt int
row.Scan(&cnt) row.Scan(&cnt)

20
client/orm/models_test.go
View File

@ -492,45 +492,45 @@ var (
helpinfo = `need driver and source! helpinfo = `need driver and source!
Default DB Drivers. Default DB Drivers.
driver: url driver: url
mysql: https://github.com/go-sql-driver/mysql mysql: https://github.com/go-sql-driver/mysql
sqlite3: https://github.com/mattn/go-sqlite3 sqlite3: https://github.com/mattn/go-sqlite3
postgres: https://github.com/lib/pq postgres: https://github.com/lib/pq
tidb: https://github.com/pingcap/tidb tidb: https://github.com/pingcap/tidb
usage: usage:
go get -u github.com/beego/beego/v2/client/orm go get -u github.com/beego/beego/v2/client/orm
go get -u github.com/go-sql-driver/mysql go get -u github.com/go-sql-driver/mysql
go get -u github.com/mattn/go-sqlite3 go get -u github.com/mattn/go-sqlite3
go get -u github.com/lib/pq go get -u github.com/lib/pq
go get -u github.com/pingcap/tidb go get -u github.com/pingcap/tidb
#### MySQL #### MySQL
mysql -u root -e 'create database orm_test;' mysql -u root -e 'create database orm_test;'
export ORM_DRIVER=mysql export ORM_DRIVER=mysql
export ORM_SOURCE="root:@/orm_test?charset=utf8" export ORM_SOURCE="root:@/orm_test?charset=utf8"
go test -v github.com/beego/beego/v2/client/orm go test -v github.com/beego/beego/v2/client/orm
#### Sqlite3 #### Sqlite3
export ORM_DRIVER=sqlite3 export ORM_DRIVER=sqlite3
export ORM_SOURCE='file:memory_test?mode=memory' export ORM_SOURCE='file:memory_test?mode=memory'
go test -v github.com/beego/beego/v2/client/orm go test -v github.com/beego/beego/v2/client/orm
#### PostgreSQL #### PostgreSQL
psql -c 'create database orm_test;' -U postgres psql -c 'create database orm_test;' -U postgres
export ORM_DRIVER=postgres export ORM_DRIVER=postgres
export ORM_SOURCE="user=postgres dbname=orm_test sslmode=disable" export ORM_SOURCE="user=postgres dbname=orm_test sslmode=disable"
go test -v github.com/beego/beego/v2/client/orm go test -v github.com/beego/beego/v2/client/orm
#### TiDB #### TiDB
export ORM_DRIVER=tidb export ORM_DRIVER=tidb
export ORM_SOURCE='memory://test/test' export ORM_SOURCE='memory://test/test'
go test -v github.com/beego/beego/v2/pgk/orm go test -v github.com/beego/beego/v2/pgk/orm
` `
) )

25
client/orm/orm.go
View File

@ -136,7 +136,7 @@ func (o *ormBase) Read(md interface{}, cols ...string) error {
} }
func (o *ormBase) ReadWithCtx(ctx context.Context, md interface{}, cols ...string) error { func (o *ormBase) ReadWithCtx(ctx context.Context, md interface{}, cols ...string) error {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
return o.alias.DbBaser.Read(o.db, mi, ind, o.alias.TZ, cols, false) return o.alias.DbBaser.Read(ctx, o.db, mi, ind, o.alias.TZ, cols, false)
} }
// read data to model, like Read(), but use "SELECT FOR UPDATE" form // read data to model, like Read(), but use "SELECT FOR UPDATE" form
@ -145,7 +145,7 @@ func (o *ormBase) ReadForUpdate(md interface{}, cols ...string) error {
} }
func (o *ormBase) ReadForUpdateWithCtx(ctx context.Context, md interface{}, cols ...string) error { func (o *ormBase) ReadForUpdateWithCtx(ctx context.Context, md interface{}, cols ...string) error {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
return o.alias.DbBaser.Read(o.db, mi, ind, o.alias.TZ, cols, true) return o.alias.DbBaser.Read(ctx, o.db, mi, ind, o.alias.TZ, cols, true)
} }
// Try to read a row from the database, or insert one if it doesn't exist // Try to read a row from the database, or insert one if it doesn't exist
@ -155,7 +155,7 @@ func (o *ormBase) ReadOrCreate(md interface{}, col1 string, cols ...string) (boo
func (o *ormBase) ReadOrCreateWithCtx(ctx context.Context, md interface{}, col1 string, cols ...string) (bool, int64, error) { func (o *ormBase) ReadOrCreateWithCtx(ctx context.Context, md interface{}, col1 string, cols ...string) (bool, int64, error) {
cols = append([]string{col1}, cols...) cols = append([]string{col1}, cols...)
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
err := o.alias.DbBaser.Read(o.db, mi, ind, o.alias.TZ, cols, false) err := o.alias.DbBaser.Read(ctx, o.db, mi, ind, o.alias.TZ, cols, false)
if err == ErrNoRows { if err == ErrNoRows {
// Create // Create
id, err := o.InsertWithCtx(ctx, md) id, err := o.InsertWithCtx(ctx, md)
@ -180,7 +180,7 @@ func (o *ormBase) Insert(md interface{}) (int64, error) {
} }
func (o *ormBase) InsertWithCtx(ctx context.Context, md interface{}) (int64, error) { func (o *ormBase) InsertWithCtx(ctx context.Context, md interface{}) (int64, error) {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
id, err := o.alias.DbBaser.Insert(o.db, mi, ind, o.alias.TZ) id, err := o.alias.DbBaser.Insert(ctx, o.db, mi, ind, o.alias.TZ)
if err != nil { if err != nil {
return id, err return id, err
} }
@ -223,7 +223,7 @@ func (o *ormBase) InsertMultiWithCtx(ctx context.Context, bulk int, mds interfac
for i := 0; i < sind.Len(); i++ { for i := 0; i < sind.Len(); i++ {
ind := reflect.Indirect(sind.Index(i)) ind := reflect.Indirect(sind.Index(i))
mi, _ := o.getMiInd(ind.Interface(), false) mi, _ := o.getMiInd(ind.Interface(), false)
id, err := o.alias.DbBaser.Insert(o.db, mi, ind, o.alias.TZ) id, err := o.alias.DbBaser.Insert(ctx, o.db, mi, ind, o.alias.TZ)
if err != nil { if err != nil {
return cnt, err return cnt, err
} }
@ -234,7 +234,7 @@ func (o *ormBase) InsertMultiWithCtx(ctx context.Context, bulk int, mds interfac
} }
} else { } else {
mi, _ := o.getMiInd(sind.Index(0).Interface(), false) mi, _ := o.getMiInd(sind.Index(0).Interface(), false)
return o.alias.DbBaser.InsertMulti(o.db, mi, sind, bulk, o.alias.TZ) return o.alias.DbBaser.InsertMulti(ctx, o.db, mi, sind, bulk, o.alias.TZ)
} }
return cnt, nil return cnt, nil
} }
@ -245,7 +245,7 @@ func (o *ormBase) InsertOrUpdate(md interface{}, colConflictAndArgs ...string) (
} }
func (o *ormBase) InsertOrUpdateWithCtx(ctx context.Context, md interface{}, colConflitAndArgs ...string) (int64, error) { func (o *ormBase) InsertOrUpdateWithCtx(ctx context.Context, md interface{}, colConflitAndArgs ...string) (int64, error) {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
id, err := o.alias.DbBaser.InsertOrUpdate(o.db, mi, ind, o.alias, colConflitAndArgs...) id, err := o.alias.DbBaser.InsertOrUpdate(ctx, o.db, mi, ind, o.alias, colConflitAndArgs...)
if err != nil { if err != nil {
return id, err return id, err
} }
@ -262,7 +262,7 @@ func (o *ormBase) Update(md interface{}, cols ...string) (int64, error) {
} }
func (o *ormBase) UpdateWithCtx(ctx context.Context, md interface{}, cols ...string) (int64, error) { func (o *ormBase) UpdateWithCtx(ctx context.Context, md interface{}, cols ...string) (int64, error) {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
return o.alias.DbBaser.Update(o.db, mi, ind, o.alias.TZ, cols) return o.alias.DbBaser.Update(ctx, o.db, mi, ind, o.alias.TZ, cols)
} }
// delete model in database // delete model in database
@ -272,7 +272,7 @@ func (o *ormBase) Delete(md interface{}, cols ...string) (int64, error) {
} }
func (o *ormBase) DeleteWithCtx(ctx context.Context, md interface{}, cols ...string) (int64, error) { func (o *ormBase) DeleteWithCtx(ctx context.Context, md interface{}, cols ...string) (int64, error) {
mi, ind := o.getMiInd(md, true) mi, ind := o.getMiInd(md, true)
num, err := o.alias.DbBaser.Delete(o.db, mi, ind, o.alias.TZ, cols) num, err := o.alias.DbBaser.Delete(ctx, o.db, mi, ind, o.alias.TZ, cols)
if err != nil { if err != nil {
return num, err return num, err
} }
@ -297,7 +297,7 @@ func (o *ormBase) QueryM2MWithCtx(ctx context.Context, md interface{}, name stri
panic(fmt.Errorf("<Ormer.QueryM2M> model `%s` . name `%s` is not a m2m field", fi.name, mi.fullName)) panic(fmt.Errorf("<Ormer.QueryM2M> model `%s` . name `%s` is not a m2m field", fi.name, mi.fullName))
} }
return newQueryM2M(md, o, mi, fi, ind) return newQueryM2M(ctx, md, o, mi, fi, ind)
} }
// load related models to md model. // load related models to md model.
@ -470,7 +470,7 @@ func (o *ormBase) QueryTableWithCtx(ctx context.Context, ptrStructOrTableName in
if qs == nil { if qs == nil {
panic(fmt.Errorf("<Ormer.QueryTable> table name: `%s` not exists", name)) panic(fmt.Errorf("<Ormer.QueryTable> table name: `%s` not exists", name))
} }
return return qs.WithContext(ctx)
} }
// return a raw query seter for raw sql string. // return a raw query seter for raw sql string.
@ -596,9 +596,8 @@ func NewOrm() Ormer {
func NewOrmUsingDB(aliasName string) Ormer { func NewOrmUsingDB(aliasName string) Ormer {
if al, ok := dataBaseCache.get(aliasName); ok { if al, ok := dataBaseCache.get(aliasName); ok {
return newDBWithAlias(al) return newDBWithAlias(al)
} else {
panic(fmt.Errorf("<Ormer.Using> unknown db alias name `%s`", aliasName))
} }
panic(fmt.Errorf("<Ormer.Using> unknown db alias name `%s`", aliasName))
} }
// NewOrmWithDB create a new ormer object with specify *sql.DB for query // NewOrmWithDB create a new ormer object with specify *sql.DB for query

17
client/orm/orm_log.go
View File

@ -85,20 +85,31 @@ func (d *stmtQueryLog) Close() error {
} }
func (d *stmtQueryLog) Exec(args ...interface{}) (sql.Result, error) { func (d *stmtQueryLog) Exec(args ...interface{}) (sql.Result, error) {
return d.ExecContext(context.Background(), args...)
}
func (d *stmtQueryLog) ExecContext(ctx context.Context, args ...interface{}) (sql.Result, error) {
a := time.Now() a := time.Now()
res, err := d.stmt.Exec(args...) res, err := d.stmt.ExecContext(ctx, args...)
debugLogQueies(d.alias, "st.Exec", d.query, a, err, args...) debugLogQueies(d.alias, "st.Exec", d.query, a, err, args...)
return res, err return res, err
} }
func (d *stmtQueryLog) Query(args ...interface{}) (*sql.Rows, error) { func (d *stmtQueryLog) Query(args ...interface{}) (*sql.Rows, error) {
return d.QueryContext(context.Background(), args...)
}
func (d *stmtQueryLog) QueryContext(ctx context.Context, args ...interface{}) (*sql.Rows, error) {
a := time.Now() a := time.Now()
res, err := d.stmt.Query(args...) res, err := d.stmt.QueryContext(ctx, args...)
debugLogQueies(d.alias, "st.Query", d.query, a, err, args...) debugLogQueies(d.alias, "st.Query", d.query, a, err, args...)
return res, err return res, err
} }
func (d *stmtQueryLog) QueryRow(args ...interface{}) *sql.Row { func (d *stmtQueryLog) QueryRow(args ...interface{}) *sql.Row {
return d.QueryRowContext(context.Background(), args...)
}
func (d *stmtQueryLog) QueryRowContext(ctx context.Context, args ...interface{}) *sql.Row {
a := time.Now() a := time.Now()
res := d.stmt.QueryRow(args...) res := d.stmt.QueryRow(args...)
debugLogQueies(d.alias, "st.QueryRow", d.query, a, nil, args...) debugLogQueies(d.alias, "st.QueryRow", d.query, a, nil, args...)

9
client/orm/orm_object.go
View File

@ -15,12 +15,14 @@
package orm package orm
import ( import (
"context"
"fmt" "fmt"
"reflect" "reflect"
) )
// an insert queryer struct // an insert queryer struct
type insertSet struct { type insertSet struct {
ctx context.Context
mi *modelInfo mi *modelInfo
orm *ormBase orm *ormBase
stmt stmtQuerier stmt stmtQuerier
@ -44,7 +46,7 @@ func (o *insertSet) Insert(md interface{}) (int64, error) {
if name != o.mi.fullName { if name != o.mi.fullName {
panic(fmt.Errorf("<Inserter.Insert> need model `%s` but found `%s`", o.mi.fullName, name)) panic(fmt.Errorf("<Inserter.Insert> need model `%s` but found `%s`", o.mi.fullName, name))
} }
id, err := o.orm.alias.DbBaser.InsertStmt(o.stmt, o.mi, ind, o.orm.alias.TZ) id, err := o.orm.alias.DbBaser.InsertStmt(o.ctx, o.stmt, o.mi, ind, o.orm.alias.TZ)
if err != nil { if err != nil {
return id, err return id, err
} }
@ -70,11 +72,12 @@ func (o *insertSet) Close() error {
} }
// create new insert queryer. // create new insert queryer.
func newInsertSet(orm *ormBase, mi *modelInfo) (Inserter, error) { func newInsertSet(ctx context.Context, orm *ormBase, mi *modelInfo) (Inserter, error) {
bi := new(insertSet) bi := new(insertSet)
bi.ctx = ctx
bi.orm = orm bi.orm = orm
bi.mi = mi bi.mi = mi
st, query, err := orm.alias.DbBaser.PrepareInsert(orm.db, mi) st, query, err := orm.alias.DbBaser.PrepareInsert(ctx, orm.db, mi)
if err != nil { if err != nil {
return nil, err return nil, err
} }

13
client/orm/orm_querym2m.go
View File

@ -14,10 +14,14 @@
package orm package orm
import "reflect" import (
"context"
"reflect"
)
// model to model struct // model to model struct
type queryM2M struct { type queryM2M struct {
ctx context.Context
md interface{} md interface{}
mi *modelInfo mi *modelInfo
fi *fieldInfo fi *fieldInfo
@ -96,7 +100,7 @@ func (o *queryM2M) Add(mds ...interface{}) (int64, error) {
} }
names = append(names, otherNames...) names = append(names, otherNames...)
values = append(values, otherValues...) values = append(values, otherValues...)
return dbase.InsertValue(orm.db, mi, true, names, values) return dbase.InsertValue(o.ctx, orm.db, mi, true, names, values)
} }
// remove models following the origin model relationship // remove models following the origin model relationship
@ -129,12 +133,13 @@ func (o *queryM2M) Count() (int64, error) {
var _ QueryM2Mer = new(queryM2M) var _ QueryM2Mer = new(queryM2M)
// create new M2M queryer. // create new M2M queryer.
func newQueryM2M(md interface{}, o *ormBase, mi *modelInfo, fi *fieldInfo, ind reflect.Value) QueryM2Mer { func newQueryM2M(ctx context.Context, md interface{}, o *ormBase, mi *modelInfo, fi *fieldInfo, ind reflect.Value) QueryM2Mer {
qm2m := new(queryM2M) qm2m := new(queryM2M)
qm2m.ctx = ctx
qm2m.md = md qm2m.md = md
qm2m.mi = mi qm2m.mi = mi
qm2m.fi = fi qm2m.fi = fi
qm2m.ind = ind qm2m.ind = ind
qm2m.qs = newQuerySet(o, fi.relThroughModelInfo).(*querySet) qm2m.qs = newQuerySet(o, fi.relThroughModelInfo).WithContext(ctx).(*querySet)
return qm2m return qm2m
} }

74
client/orm/orm_queryset.go
View File

@ -17,8 +17,9 @@ package orm
import ( import (
"context" "context"
"fmt" "fmt"
"github.com/beego/beego/v2/client/orm/hints"
"github.com/beego/beego/v2/client/orm/clauses/order_clause" "github.com/beego/beego/v2/client/orm/clauses/order_clause"
"github.com/beego/beego/v2/client/orm/hints"
) )
type colValue struct { type colValue struct {
@ -64,22 +65,21 @@ func ColValue(opt operator, value interface{}) interface{} {
// real query struct // real query struct
type querySet struct { type querySet struct {
mi *modelInfo mi *modelInfo
cond *Condition cond *Condition
related []string related []string
relDepth int relDepth int
limit int64 limit int64
offset int64 offset int64
groups []string groups []string
orders []*order_clause.Order orders []*order_clause.Order
distinct bool distinct bool
forUpdate bool forUpdate bool
useIndex int useIndex int
indexes []string indexes []string
orm *ormBase orm *ormBase
ctx context.Context ctx context.Context
forContext bool aggregate string
aggregate string
} }
var _ QuerySeter = new(querySet) var _ QuerySeter = new(querySet)
@ -221,25 +221,36 @@ func (o querySet) GetCond() *Condition {
return o.cond return o.cond
} }
func (o querySet) getContext() context.Context {
if o.ctx != nil {
return o.ctx
}
return context.Background()
}
// return QuerySeter execution result number // return QuerySeter execution result number
func (o *querySet) Count() (int64, error) { func (o *querySet) Count() (int64, error) {
return o.orm.alias.DbBaser.Count(o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.Count(ctx, o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ)
} }
// check result empty or not after QuerySeter executed // check result empty or not after QuerySeter executed
func (o *querySet) Exist() bool { func (o *querySet) Exist() bool {
cnt, _ := o.orm.alias.DbBaser.Count(o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ) ctx := o.getContext()
cnt, _ := o.orm.alias.DbBaser.Count(ctx, o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ)
return cnt > 0 return cnt > 0
} }
// execute update with parameters // execute update with parameters
func (o *querySet) Update(values Params) (int64, error) { func (o *querySet) Update(values Params) (int64, error) {
return o.orm.alias.DbBaser.UpdateBatch(o.orm.db, o, o.mi, o.cond, values, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.UpdateBatch(ctx, o.orm.db, o, o.mi, o.cond, values, o.orm.alias.TZ)
} }
// execute delete // execute delete
func (o *querySet) Delete() (int64, error) { func (o *querySet) Delete() (int64, error) {
return o.orm.alias.DbBaser.DeleteBatch(o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.DeleteBatch(ctx, o.orm.db, o, o.mi, o.cond, o.orm.alias.TZ)
} }
// return a insert queryer. // return a insert queryer.
@ -248,20 +259,23 @@ func (o *querySet) Delete() (int64, error) {
// i,err := sq.PrepareInsert() // i,err := sq.PrepareInsert()
// i.Add(&user1{},&user2{}) // i.Add(&user1{},&user2{})
func (o *querySet) PrepareInsert() (Inserter, error) { func (o *querySet) PrepareInsert() (Inserter, error) {
return newInsertSet(o.orm, o.mi) ctx := o.getContext()
return newInsertSet(ctx, o.orm, o.mi)
} }
// query all data and map to containers. // query all data and map to containers.
// cols means the columns when querying. // cols means the columns when querying.
func (o *querySet) All(container interface{}, cols ...string) (int64, error) { func (o *querySet) All(container interface{}, cols ...string) (int64, error) {
return o.orm.alias.DbBaser.ReadBatch(o.orm.db, o, o.mi, o.cond, container, o.orm.alias.TZ, cols) ctx := o.getContext()
return o.orm.alias.DbBaser.ReadBatch(ctx, o.orm.db, o, o.mi, o.cond, container, o.orm.alias.TZ, cols)
} }
// query one row data and map to containers. // query one row data and map to containers.
// cols means the columns when querying. // cols means the columns when querying.
func (o *querySet) One(container interface{}, cols ...string) error { func (o *querySet) One(container interface{}, cols ...string) error {
ctx := o.getContext()
o.limit = 1 o.limit = 1
num, err := o.orm.alias.DbBaser.ReadBatch(o.orm.db, o, o.mi, o.cond, container, o.orm.alias.TZ, cols) num, err := o.orm.alias.DbBaser.ReadBatch(ctx, o.orm.db, o, o.mi, o.cond, container, o.orm.alias.TZ, cols)
if err != nil { if err != nil {
return err return err
} }
@ -279,19 +293,22 @@ func (o *querySet) One(container interface{}, cols ...string) error {
// expres means condition expression. // expres means condition expression.
// it converts data to []map[column]value. // it converts data to []map[column]value.
func (o *querySet) Values(results *[]Params, exprs ...string) (int64, error) { func (o *querySet) Values(results *[]Params, exprs ...string) (int64, error) {
return o.orm.alias.DbBaser.ReadValues(o.orm.db, o, o.mi, o.cond, exprs, results, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.ReadValues(ctx, o.orm.db, o, o.mi, o.cond, exprs, results, o.orm.alias.TZ)
} }
// query all data and map to [][]interface // query all data and map to [][]interface
// it converts data to [][column_index]value // it converts data to [][column_index]value
func (o *querySet) ValuesList(results *[]ParamsList, exprs ...string) (int64, error) { func (o *querySet) ValuesList(results *[]ParamsList, exprs ...string) (int64, error) {
return o.orm.alias.DbBaser.ReadValues(o.orm.db, o, o.mi, o.cond, exprs, results, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.ReadValues(ctx, o.orm.db, o, o.mi, o.cond, exprs, results, o.orm.alias.TZ)
} }
// query all data and map to []interface. // query all data and map to []interface.
// it's designed for one row record set, auto change to []value, not [][column]value. // it's designed for one row record set, auto change to []value, not [][column]value.
func (o *querySet) ValuesFlat(result *ParamsList, expr string) (int64, error) { func (o *querySet) ValuesFlat(result *ParamsList, expr string) (int64, error) {
return o.orm.alias.DbBaser.ReadValues(o.orm.db, o, o.mi, o.cond, []string{expr}, result, o.orm.alias.TZ) ctx := o.getContext()
return o.orm.alias.DbBaser.ReadValues(ctx, o.orm.db, o, o.mi, o.cond, []string{expr}, result, o.orm.alias.TZ)
} }
// query all rows into map[string]interface with specify key and value column name. // query all rows into map[string]interface with specify key and value column name.
@ -325,7 +342,6 @@ func (o *querySet) RowsToStruct(ptrStruct interface{}, keyCol, valueCol string)
// set context to QuerySeter. // set context to QuerySeter.
func (o querySet) WithContext(ctx context.Context) QuerySeter { func (o querySet) WithContext(ctx context.Context) QuerySeter {
o.ctx = ctx o.ctx = ctx
o.forContext = true
return &o return &o
} }
@ -341,4 +357,4 @@ func newQuerySet(orm *ormBase, mi *modelInfo) QuerySeter {
func (o querySet) Aggregate(s string) QuerySeter { func (o querySet) Aggregate(s string) QuerySeter {
o.aggregate = s o.aggregate = s
return &o return &o
} }

20
client/orm/orm_test.go
View File

@ -2820,3 +2820,23 @@ func TestCondition(t *testing.T) {
throwFail(t, AssertIs(!cycleFlag, true)) throwFail(t, AssertIs(!cycleFlag, true))
return return
} }
func TestContextCanceled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
user := User{UserName: "slene"}
err := dORM.ReadWithCtx(ctx, &user, "UserName")
throwFail(t, err)
cancel()
err = dORM.ReadWithCtx(ctx, &user, "UserName")
throwFail(t, AssertIs(err, context.Canceled))
ctx, cancel = context.WithCancel(context.Background())
cancel()
qs := dORM.QueryTable(user).WithContext(ctx)
_, err = qs.Filter("UserName", "slene").Count()
throwFail(t, AssertIs(err, context.Canceled))
}

42
client/orm/types.go
View File

@ -236,6 +236,8 @@ type Inserter interface {
// QuerySeter query seter // QuerySeter query seter
type QuerySeter interface { type QuerySeter interface {
// add query context for querySeter
WithContext(context.Context) QuerySeter
// add condition expression to QuerySeter. // add condition expression to QuerySeter.
// for example: // for example:
// filter by UserName == 'slene' // filter by UserName == 'slene'
@ -539,11 +541,11 @@ type RawSeter interface {
type stmtQuerier interface { type stmtQuerier interface {
Close() error Close() error
Exec(args ...interface{}) (sql.Result, error) Exec(args ...interface{}) (sql.Result, error)
// ExecContext(ctx context.Context, args ...interface{}) (sql.Result, error) ExecContext(ctx context.Context, args ...interface{}) (sql.Result, error)
Query(args ...interface{}) (*sql.Rows, error) Query(args ...interface{}) (*sql.Rows, error)
// QueryContext(args ...interface{}) (*sql.Rows, error) QueryContext(ctx context.Context, args ...interface{}) (*sql.Rows, error)
QueryRow(args ...interface{}) *sql.Row QueryRow(args ...interface{}) *sql.Row
// QueryRowContext(ctx context.Context, args ...interface{}) *sql.Row QueryRowContext(ctx context.Context, args ...interface{}) *sql.Row
} }
// db querier // db querier
@ -580,28 +582,28 @@ type txEnder interface {
// base database struct // base database struct
type dbBaser interface { type dbBaser interface {
Read(dbQuerier, *modelInfo, reflect.Value, *time.Location, []string, bool) error Read(context.Context, dbQuerier, *modelInfo, reflect.Value, *time.Location, []string, bool) error
ReadBatch(dbQuerier, *querySet, *modelInfo, *Condition, interface{}, *time.Location, []string) (int64, error) ReadBatch(context.Context, dbQuerier, *querySet, *modelInfo, *Condition, interface{}, *time.Location, []string) (int64, error)
Count(dbQuerier, *querySet, *modelInfo, *Condition, *time.Location) (int64, error) Count(context.Context, dbQuerier, *querySet, *modelInfo, *Condition, *time.Location) (int64, error)
ReadValues(dbQuerier, *querySet, *modelInfo, *Condition, []string, interface{}, *time.Location) (int64, error) ReadValues(context.Context, dbQuerier, *querySet, *modelInfo, *Condition, []string, interface{}, *time.Location) (int64, error)
Insert(dbQuerier, *modelInfo, reflect.Value, *time.Location) (int64, error) Insert(context.Context, dbQuerier, *modelInfo, reflect.Value, *time.Location) (int64, error)
InsertOrUpdate(dbQuerier, *modelInfo, reflect.Value, *alias, ...string) (int64, error) InsertOrUpdate(context.Context, dbQuerier, *modelInfo, reflect.Value, *alias, ...string) (int64, error)
InsertMulti(dbQuerier, *modelInfo, reflect.Value, int, *time.Location) (int64, error) InsertMulti(context.Context, dbQuerier, *modelInfo, reflect.Value, int, *time.Location) (int64, error)
InsertValue(dbQuerier, *modelInfo, bool, []string, []interface{}) (int64, error) InsertValue(context.Context, dbQuerier, *modelInfo, bool, []string, []interface{}) (int64, error)
InsertStmt(stmtQuerier, *modelInfo, reflect.Value, *time.Location) (int64, error) InsertStmt(context.Context, stmtQuerier, *modelInfo, reflect.Value, *time.Location) (int64, error)
Update(dbQuerier, *modelInfo, reflect.Value, *time.Location, []string) (int64, error) Update(context.Context, dbQuerier, *modelInfo, reflect.Value, *time.Location, []string) (int64, error)
UpdateBatch(dbQuerier, *querySet, *modelInfo, *Condition, Params, *time.Location) (int64, error) UpdateBatch(context.Context, dbQuerier, *querySet, *modelInfo, *Condition, Params, *time.Location) (int64, error)
Delete(dbQuerier, *modelInfo, reflect.Value, *time.Location, []string) (int64, error) Delete(context.Context, dbQuerier, *modelInfo, reflect.Value, *time.Location, []string) (int64, error)
DeleteBatch(dbQuerier, *querySet, *modelInfo, *Condition, *time.Location) (int64, error) DeleteBatch(context.Context, dbQuerier, *querySet, *modelInfo, *Condition, *time.Location) (int64, error)
SupportUpdateJoin() bool SupportUpdateJoin() bool
OperatorSQL(string) string OperatorSQL(string) string
GenerateOperatorSQL(*modelInfo, *fieldInfo, string, []interface{}, *time.Location) (string, []interface{}) GenerateOperatorSQL(*modelInfo, *fieldInfo, string, []interface{}, *time.Location) (string, []interface{})
GenerateOperatorLeftCol(*fieldInfo, string, *string) GenerateOperatorLeftCol(*fieldInfo, string, *string)
PrepareInsert(dbQuerier, *modelInfo) (stmtQuerier, string, error) PrepareInsert(context.Context, dbQuerier, *modelInfo) (stmtQuerier, string, error)
MaxLimit() uint64 MaxLimit() uint64
TableQuote() string TableQuote() string
ReplaceMarks(*string) ReplaceMarks(*string)
@ -610,12 +612,12 @@ type dbBaser interface {
TimeToDB(*time.Time, *time.Location) TimeToDB(*time.Time, *time.Location)
DbTypes() map[string]string DbTypes() map[string]string
GetTables(dbQuerier) (map[string]bool, error) GetTables(dbQuerier) (map[string]bool, error)
GetColumns(dbQuerier, string) (map[string][3]string, error) GetColumns(context.Context, dbQuerier, string) (map[string][3]string, error)
ShowTablesQuery() string ShowTablesQuery() string
ShowColumnsQuery(string) string ShowColumnsQuery(string) string
IndexExists(dbQuerier, string, string) bool IndexExists(context.Context, dbQuerier, string, string) bool
collectFieldValue(*modelInfo, *fieldInfo, reflect.Value, bool, *time.Location) (interface{}, error) collectFieldValue(*modelInfo, *fieldInfo, reflect.Value, bool, *time.Location) (interface{}, error)
setval(dbQuerier, *modelInfo, []string) error setval(context.Context, dbQuerier, *modelInfo, []string) error
GenerateSpecifyIndex(tableName string, useIndex int, indexes []string) string GenerateSpecifyIndex(tableName string, useIndex int, indexes []string) string
} }