Method Execute

backend/helpers/pluginhelper/api/api_extractor.go:73–142

Execute sub-task

()

// Execute sub-task
func (extractor *ApiExtractor) Execute() errors.Error {
	// load data from database
	db := extractor.args.Ctx.GetDal()
	logger := extractor.args.Ctx.GetLogger()
	if !db.HasTable(extractor.table) {
		return nil
	}
	clauses := []dal.Clause{
		dal.From(extractor.table),
		dal.Where("params = ?", extractor.params),
		dal.Orderby("id ASC"),
	}

	count, err := db.Count(clauses...)
	if err != nil {
		return errors.Default.Wrap(err, "error getting count of clauses")
	}
	cursor, err := db.Cursor(clauses...)
	if err != nil {
		return errors.Default.Wrap(err, "error running DB query")
	}
	logger.Info("get data from %s where params=%s and got %d", extractor.table, extractor.params, count)
	defer cursor.Close()
	// batch save divider
	divider := NewBatchSaveDivider(extractor.args.Ctx, extractor.args.BatchSize, extractor.table, extractor.params)

	// progress
	extractor.args.Ctx.SetProgress(0, -1)
	ctx := extractor.args.Ctx.GetContext()
	// iterate all rows
	for cursor.Next() {
		select {
		case <-ctx.Done():
			return errors.Convert(ctx.Err())
		default:
		}
		row := &RawData{}
		err = db.Fetch(cursor, row)
		if err != nil {
			return errors.Default.Wrap(err, "error fetching row")
		}

		results, err := extractor.args.Extract(row)
		if err != nil {
			return errors.Default.Wrap(err, "error calling plugin Extract implementation")
		}
		for _, result := range results {
			// get the batch operator for the specific type
			batch, err := divider.ForType(reflect.TypeOf(result))
			if err != nil {
				return errors.Default.Wrap(err, "error getting batch from result")
			}
			// set raw data origin field
			setRawDataOrigin(result, common.RawDataOrigin{
				RawDataTable:  extractor.table,
				RawDataId:     row.ID,
				RawDataParams: row.Params,
			})