DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Add endpoint for bulk record patch

Browse Source
This commit is contained in:
Tomaž Jerman
2023-02-28 11:39:08 +01:00
committed by Kelani Tolulope
parent 850f4ef6b0
commit 2d174620fa
9 changed files with 367 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/compose/automation/records_handler.go
View File

@@ -17,7 +17,7 @@ type (
Create(ctx context.Context, record *types.Record) (*types.Record, *types.RecordValueErrorSet, error)
Update(ctx context.Context, record *types.Record) (*types.Record, *types.RecordValueErrorSet, error)
Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (types.RecordSet, *types.RecordValueErrorSet, error)
Bulk(ctx context.Context, skipFailed bool, oo ...*types.RecordBulkOperation) ([]types.RecordBulkOperationResult, error)
Report(ctx context.Context, namespaceID, moduleID uint64, metrics, dimensions, filter string) (out any, err error)
Validate(ctx context.Context, rec *types.Record) error

8
server/compose/rest.yaml
View File

@@ -804,6 +804,14 @@ endpoints:
- { name: ownedBy, type: "uint64", title: Record Owner }
- { name: meta, type: "map[string]any", title: Record meta-data, parser: payload.ParseMeta }
- { name: records, type: "types.RecordBulkSet", title: Records }
- name: patch
method: PATCH
title: Partially update record values
path: "/"
parameters:
post:
- { name: records, type: "[]string", title: Records to update }
- { name: values, type: "types.RecordValueSet", title: Fields to update and their values }
- name: bulkDelete
method: DELETE
title: Delete record row from module section

19
server/compose/rest/handlers/record.go
View File

@@ -29,6 +29,7 @@ type (
Create(context.Context, *request.RecordCreate) (interface{}, error)
Read(context.Context, *request.RecordRead) (interface{}, error)
Update(context.Context, *request.RecordUpdate) (interface{}, error)
Patch(context.Context, *request.RecordPatch) (interface{}, error)
BulkDelete(context.Context, *request.RecordBulkDelete) (interface{}, error)
Delete(context.Context, *request.RecordDelete) (interface{}, error)
Undelete(context.Context, *request.RecordUndelete) (interface{}, error)
@@ -51,6 +52,7 @@ type (
Create func(http.ResponseWriter, *http.Request)
Read func(http.ResponseWriter, *http.Request)
Update func(http.ResponseWriter, *http.Request)
Patch func(http.ResponseWriter, *http.Request)
BulkDelete func(http.ResponseWriter, *http.Request)
Delete func(http.ResponseWriter, *http.Request)
Undelete func(http.ResponseWriter, *http.Request)
@@ -224,6 +226,22 @@ func NewRecord(h RecordAPI) *Record {
api.Send(w, r, value)
},
Patch: func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
params := request.NewRecordPatch()
if err := params.Fill(r); err != nil {
api.Send(w, r, err)
return
}
value, err := h.Patch(r.Context(), params)
if err != nil {
api.Send(w, r, err)
return
}
api.Send(w, r, value)
},
BulkDelete: func(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
params := request.NewRecordBulkDelete()
@@ -368,6 +386,7 @@ func (h Record) MountRoutes(r chi.Router, middlewares ...func(http.Handler) http
r.Post("/namespace/{namespaceID}/module/{moduleID}/record/", h.Create)
r.Get("/namespace/{namespaceID}/module/{moduleID}/record/{recordID}", h.Read)
r.Post("/namespace/{namespaceID}/module/{moduleID}/record/{recordID}", h.Update)
r.Patch("/namespace/{namespaceID}/module/{moduleID}/record/", h.Patch)
r.Delete("/namespace/{namespaceID}/module/{moduleID}/record/", h.BulkDelete)
r.Delete("/namespace/{namespaceID}/module/{moduleID}/record/{recordID}", h.Delete)
r.Post("/namespace/{namespaceID}/module/{moduleID}/record/{recordID}/undelete", h.Undelete)

80
server/compose/rest/record.go
View File

@@ -12,6 +12,7 @@ import (
"time"
"github.com/cortezaproject/corteza/server/pkg/revisions"
"github.com/spf13/cast"
"github.com/cortezaproject/corteza/server/compose/rest/request"
"github.com/cortezaproject/corteza/server/compose/service"
@@ -30,6 +31,15 @@ import (
)
type (
recordBulkPatchRecord struct {
Record *types.Record `json:"record"`
Error error `json:"error,omitempty"`
ValueErrors *types.RecordValueErrorSet `json:"valueErrors,omitempty"`
}
recordBulkPatchPayload struct {
Records []recordBulkPatchRecord `json:"records"`
}
recordPayload struct {
*types.Record
@@ -203,14 +213,50 @@ func (ctrl *Record) Create(ctx context.Context, r *request.RecordCreate) (interf
}
oo = append(oo, oob...)
rr, dd, err := ctrl.record.Bulk(ctx, oo...)
results, err := ctrl.record.Bulk(ctx, false, oo...)
if rve := types.IsRecordValueErrorSet(err); rve != nil {
return ctrl.handleValidationError(rve), nil
}
var (
rr types.RecordSet
dd = &types.RecordValueErrorSet{}
)
for _, r := range results {
rr = append(rr, r.Record)
dd.Merge(r.DuplicationError)
}
return ctrl.makeBulkPayload(ctx, m, dd, err, rr...)
}
func (ctrl *Record) Patch(ctx context.Context, req *request.RecordPatch) (interface{}, error) {
var (
err error
)
oo := make([]*types.RecordBulkOperation, 0)
for _, r := range req.Records {
oo = append(oo, &types.RecordBulkOperation{
RecordID: cast.ToUint64(r),
NamespaceID: req.NamespaceID,
ModuleID: req.ModuleID,
Record: &types.Record{Values: req.Values},
Operation: types.OperationTypePatch,
})
}
results, err := ctrl.record.Bulk(ctx, true, oo...)
if rve := types.IsRecordValueErrorSet(err); rve != nil {
return ctrl.handleValidationError(rve), nil
}
return ctrl.makeRecordBulkPatchPayload(ctx, results, err)
}
func (ctrl *Record) Update(ctx context.Context, r *request.RecordUpdate) (interface{}, error) {
var (
m *types.Module
@@ -255,11 +301,19 @@ func (ctrl *Record) Update(ctx context.Context, r *request.RecordUpdate) (interf
}
oo = append(oo, oob...)
rr, dd, err := ctrl.record.Bulk(ctx, oo...)
results, err := ctrl.record.Bulk(ctx, false, oo...)
if rve := types.IsRecordValueErrorSet(err); rve != nil {
return ctrl.handleValidationError(rve), nil
}
var rr types.RecordSet
dd := &types.RecordValueErrorSet{}
for _, r := range results {
rr = append(rr, r.Record)
dd.Merge(r.DuplicationError)
}
return ctrl.makeBulkPayload(ctx, m, dd, err, rr...)
}
@@ -637,6 +691,28 @@ func (ctrl Record) makeBulkPayload(ctx context.Context, m *types.Module, dd *typ
}, nil
}
func (ctrl Record) makeRecordBulkPatchPayload(ctx context.Context, rr []types.RecordBulkOperationResult, err error) (*recordBulkPatchPayload, error) {
if err != nil {
return nil, err
}
out := &recordBulkPatchPayload{
Records: make([]recordBulkPatchRecord, 0, len(rr)),
}
for _, r := range rr {
vr := r.ValueError
vr.Merge(r.DuplicationError)
out.Records = append(out.Records, recordBulkPatchRecord{
Record: r.Record,
ValueErrors: vr,
Error: r.Error,
})
}
return out, nil
}
func (ctrl Record) makePayload(ctx context.Context, m *types.Module, r *types.Record, dd *types.RecordValueErrorSet, err error) (*recordPayload, error) {
if err != nil || r == nil {
return nil, err

124
server/compose/rest/request/record.go
View File

@@ -319,6 +319,28 @@ type (
Records types.RecordBulkSet
}
RecordPatch struct {
// NamespaceID PATH parameter
//
// Namespace ID
NamespaceID uint64 `json:",string"`
// ModuleID PATH parameter
//
// Module ID
ModuleID uint64 `json:",string"`
// Records POST parameter
//
// Records to update
Records []string
// Values POST parameter
//
// Fields to update and their values
Values types.RecordValueSet
}
RecordBulkDelete struct {
// NamespaceID PATH parameter
//
@@ -1596,6 +1618,108 @@ func (r *RecordUpdate) Fill(req *http.Request) (err error) {
return err
}
// NewRecordPatch request
func NewRecordPatch() *RecordPatch {
return &RecordPatch{}
}
// Auditable returns all auditable/loggable parameters
func (r RecordPatch) Auditable() map[string]interface{} {
return map[string]interface{}{
"namespaceID": r.NamespaceID,
"moduleID": r.ModuleID,
"records": r.Records,
"values": r.Values,
}
}
// Auditable returns all auditable/loggable parameters
func (r RecordPatch) GetNamespaceID() uint64 {
return r.NamespaceID
}
// Auditable returns all auditable/loggable parameters
func (r RecordPatch) GetModuleID() uint64 {
return r.ModuleID
}
// Auditable returns all auditable/loggable parameters
func (r RecordPatch) GetRecords() []string {
return r.Records
}
// Auditable returns all auditable/loggable parameters
func (r RecordPatch) GetValues() types.RecordValueSet {
return r.Values
}
// Fill processes request and fills internal variables
func (r *RecordPatch) Fill(req *http.Request) (err error) {
if strings.HasPrefix(strings.ToLower(req.Header.Get("content-type")), "application/json") {
err = json.NewDecoder(req.Body).Decode(r)
switch {
case err == io.EOF:
err = nil
case err != nil:
return fmt.Errorf("error parsing http request body: %w", err)
}
}
{
// Caching 32MB to memory, the rest to disk
if err = req.ParseMultipartForm(32 << 20); err != nil && err != http.ErrNotMultipart {
return err
} else if err == nil {
// Multipart params
}
}
{
if err = req.ParseForm(); err != nil {
return err
}
// POST params
//if val, ok := req.Form["records[]"]; ok && len(val) > 0 {
// r.Records, err = val, nil
// if err != nil {
// return err
// }
//}
//if val, ok := req.Form["values[]"]; ok && len(val) > 0 {
// r.Values, err = types.RecordValueSet(val), nil
// if err != nil {
// return err
// }
//}
}
{
var val string
// path params
val = chi.URLParam(req, "namespaceID")
r.NamespaceID, err = payload.ParseUint64(val), nil
if err != nil {
return err
}
val = chi.URLParam(req, "moduleID")
r.ModuleID, err = payload.ParseUint64(val), nil
if err != nil {
return err
}
}
return err
}
// NewRecordBulkDelete request
func NewRecordBulkDelete() *RecordBulkDelete {
return &RecordBulkDelete{}

146
server/compose/service/record.go
View File

@@ -124,7 +124,7 @@ type (
Create(ctx context.Context, record *types.Record) (*types.Record, *types.RecordValueErrorSet, error)
Update(ctx context.Context, record *types.Record) (*types.Record, *types.RecordValueErrorSet, error)
Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (types.RecordSet, *types.RecordValueErrorSet, error)
Bulk(ctx context.Context, skipFailed bool, oo ...*types.RecordBulkOperation) ([]types.RecordBulkOperationResult, error)
Validate(ctx context.Context, rec *types.Record) error
@@ -545,14 +545,18 @@ func (svc record) RecordExport(ctx context.Context, f types.RecordFilter) (err e
// Bulk handles provided set of bulk record operations.
// It's able to create, update or delete records in a single transaction.
func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (rr types.RecordSet, dd *types.RecordValueErrorSet, err error) {
func (svc record) Bulk(ctx context.Context, skipFailed bool, oo ...*types.RecordBulkOperation) (rr []types.RecordBulkOperationResult, err error) {
var pr *types.Record
rr = make([]types.RecordBulkOperationResult, len(oo))
err = func() error {
// pre-verify all
for _, p := range oo {
switch p.Operation {
case types.OperationTypeCreate, types.OperationTypeUpdate, types.OperationTypeDelete:
case types.OperationTypeCreate,
types.OperationTypeUpdate,
types.OperationTypeDelete,
types.OperationTypePatch:
// ok
default:
return RecordErrUnknownBulkOperation(&recordActionProps{bulkOperation: string(p.Operation)})
@@ -560,22 +564,7 @@ func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (r
}
var (
// in case we get record value errors from create or update operations
// we ll merge the errors into one slice and return it all together
//
// this is done under assumption that potential before-record-update/create automation
// scripts are playing by the rules and do not do any changes before any potential
// record value errors are returned
//
// @todo all records/values could and should be pre-validated
// before we start storing any changes
rves = &types.RecordValueErrorSet{}
// duplication errors
ddes = &types.RecordValueErrorSet{}
// merge of record value errors and duplication errors
ee = &types.RecordValueErrorSet{}
dupErrors = &types.RecordValueErrorSet{}
action func(props ...*recordActionProps) *recordAction
r *types.Record
@@ -583,8 +572,34 @@ func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (r
aProp = &recordActionProps{}
)
for _, p := range oo {
for i, p := range oo {
var (
valueErrors *types.RecordValueErrorSet
)
r = p.Record
// Fetchthe requested record; primarily used for ops which don't need a base
if p.RecordID != 0 {
r, valueErrors, err = svc.FindByID(ctx, p.NamespaceID, p.ModuleID, p.RecordID)
// This one can't be recovered
if err != nil {
continue
}
if p.Operation == types.OperationTypePatch {
r.Values = p.Record.Values
}
rr[i] = types.RecordBulkOperationResult{
Record: r,
ValueError: valueErrors,
Error: err,
}
} else {
rr[i] = types.RecordBulkOperationResult{
Record: r,
}
}
aProp.setRecord(r)
@@ -596,8 +611,8 @@ func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (r
Name: p.LinkBy,
}
if pr != nil {
rv.Value = strconv.FormatUint(rr[0].ID, 10)
rv.Ref = rr[0].ID
rv.Value = strconv.FormatUint(rr[0].Record.ID, 10)
rv.Ref = rr[0].Record.ID
}
r.Values = r.Values.Set(rv)
}
@@ -605,64 +620,79 @@ func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (r
switch p.Operation {
case types.OperationTypeCreate:
action = RecordActionCreate
r, ddes, err = svc.create(ctx, r)
r, dupErrors, err = svc.create(ctx, r)
case types.OperationTypeUpdate:
action = RecordActionUpdate
r, ddes, err = svc.update(ctx, r)
r, dupErrors, err = svc.update(ctx, r)
case types.OperationTypeDelete:
action = RecordActionDelete
r, err = svc.delete(ctx, r.NamespaceID, r.ModuleID, r.ID)
case types.OperationTypePatch:
action = RecordActionPatch
r, dupErrors, err = svc.patch(ctx, r, r.Values)
}
aProp.setChanged(r)
// Attach meta ID to each value error for FE identification
if !ddes.HasStrictErrors() && r != nil {
ddes.SetMetaID(r.ID)
}
if !ddes.IsValid() && dd == nil {
dd = ddes
} else {
dd.Merge(ddes)
if !dupErrors.HasStrictErrors() && r != nil {
dupErrors.SetMetaID(r.ID)
}
rr[i].DuplicationError = dupErrors
if rve := types.IsRecordValueErrorSet(err); rve != nil {
if valueErrors == nil {
valueErrors = &types.RecordValueErrorSet{}
}
// Attach additional meta to each value error for FE identification
for _, re := range rve.Set {
re.Meta["id"] = p.ID
if p.ID != "" {
re.Meta["id"] = p.ID
}
rves.Push(re)
valueErrors.Push(re)
}
// log record value error for this record
_ = svc.recordAction(ctx, aProp, action, err)
rr[i].ValueError = valueErrors
// Clear the current error
err = nil
// do not return errors just yet, values on other records from the payload (if any)
// might have errors too
continue
}
_ = svc.recordAction(ctx, aProp, action, err)
if err != nil {
if !skipFailed && err != nil {
return err
} else {
rr[i].Error = err
err = nil
}
rr = append(rr, r)
if pr == nil {
pr = r
}
}
// merge record value errors and strict duplication errors
if dd.HasStrictErrors() {
ee.Merge(rves, dd)
} else {
ee = rves
var ee = &types.RecordValueErrorSet{}
for _, r := range rr {
if !r.ValueError.IsValid() {
ee.Merge(r.ValueError)
}
if r.DuplicationError.HasStrictErrors() {
ee.Merge(r.DuplicationError)
}
}
if !ee.IsValid() {
if !skipFailed && !ee.IsValid() {
// Any errors gathered?
return RecordErrValueInput().Wrap(ee)
}
@@ -673,14 +703,14 @@ func (svc record) Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (r
if len(oo) == 1 {
// was not really a bulk operation, and we already recorded the action
// inside transaction loop
return rr, dd, err
return rr, err
} else {
// when doing bulk op (updating and/or creating more than one record at once),
// we already log action for each operation
//
// to log the fact that the bulk op was done, we do one additional recording
// without any props
return rr, dd, svc.recordAction(ctx, &recordActionProps{}, RecordActionBulk, err)
return rr, svc.recordAction(ctx, &recordActionProps{}, RecordActionBulk, err)
}
}
@@ -1089,6 +1119,34 @@ func (svc record) update(ctx context.Context, upd *types.Record) (rec *types.Rec
return
}
// patch prepares a payload for the update function and utilizes that
func (svc record) patch(ctx context.Context, upd *types.Record, values types.RecordValueSet) (rec *types.Record, dd *types.RecordValueErrorSet, err error) {
var (
old *types.Record
)
if upd.ID == 0 {
return nil, dd, RecordErrInvalidID()
}
_, _, old, err = loadRecordCombo(ctx, svc.store, svc.dal, upd.NamespaceID, upd.ModuleID, upd.ID)
if err != nil {
return
}
// Create an update version from the old
upd = old.Clone()
for _, v := range values {
err = upd.SetValue(v.Name, v.Place, v.Value)
if err != nil {
return
}
}
upd.Values.SetUpdatedFlag(true)
return svc.update(ctx, upd)
}
func (svc record) Create(ctx context.Context, new *types.Record) (rec *types.Record, dd *types.RecordValueErrorSet, err error) {
var (
aProps = &recordActionProps{record: new}

20
server/compose/service/record_actions.gen.go generated
View File

@@ -570,6 +570,26 @@ func RecordActionDelete(props ...*recordActionProps) *recordAction {
return a
}
// RecordActionPatch returns "compose:record.patch" action
//
// This function is auto-generated.
//
func RecordActionPatch(props ...*recordActionProps) *recordAction {
a := &recordAction{
timestamp: time.Now(),
resource: "compose:record",
action: "patch",
log: "patched {{record}}",
severity: actionlog.Notice,
}
if len(props) > 0 {
a.props = props[0]
}
return a
}
// RecordActionUndelete returns "compose:record.undelete" action
//
// This function is auto-generated.

3
server/compose/service/record_actions.yaml
View File

@@ -70,6 +70,9 @@ actions:
- action: delete
log: "deleted {{record}}"
- action: patch
log: "patched {{record}}"
- action: undelete
log: "undeleted {{record}}"

13
server/compose/types/record.go
View File

@@ -19,11 +19,21 @@ type (
OperationType string
RecordBulkOperation struct {
Record *Record
Record *Record
RecordID uint64
NamespaceID uint64
ModuleID uint64
LinkBy string
Operation OperationType
ID string
}
RecordBulkOperationResult struct {
Record *Record
Error error
ValueError *RecordValueErrorSet
DuplicationError *RecordValueErrorSet
}
RecordBulk struct {
RefField string `json:"refField,omitempty"`
@@ -96,6 +106,7 @@ const (
OperationTypeCreate OperationType = "create"
OperationTypeUpdate OperationType = "update"
OperationTypeDelete OperationType = "delete"
OperationTypePatch OperationType = "patch"
)
func (f RecordFilter) ToConstraintedFilter(c map[string][]any) filter.Filter {