You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@devlake.apache.org by ab...@apache.org on 2024/04/28 14:56:00 UTC
(incubator-devlake) branch main updated: feat: jira issue / changelog converter supports incremental mode (#7394)
This is an automated email from the ASF dual-hosted git repository.
abeizn pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git
The following commit(s) were added to refs/heads/main by this push:
new d7cb635c6 feat: jira issue / changelog converter supports incremental mode (#7394)
d7cb635c6 is described below
commit d7cb635c626193d2ff6a72183562cafe1aab6cbb
Author: Klesh Wong <zh...@merico.dev>
AuthorDate: Sun Apr 28 22:55:53 2024 +0800
feat: jira issue / changelog converter supports incremental mode (#7394)
* feat: jira issue / changelog converter supports incremental mode
* fix: state manager unit test
* fix: gitcli clone adopts
* refactor: untify parameter processing for BatchSize
* fix: tidy up
* fix: unit test
---
.../pluginhelper/api/api_extractor_stateful.go | 17 +--
.../pluginhelper/api/data_convertor_stateful.go | 129 +++++++++++++++++++++
.../pluginhelper/api/subtask_state_manager.go | 39 ++++++-
.../pluginhelper/api/subtask_state_manager_test.go | 3 +-
.../plugins/gitextractor/parser/clone_gitcli.go | 9 +-
.../jira/tasks/issue_changelog_convertor.go | 74 ++++++------
.../jira/tasks/issue_changelog_extractor.go | 7 +-
backend/plugins/jira/tasks/issue_convertor.go | 61 +++++-----
backend/plugins/jira/tasks/issue_extractor.go | 9 +-
9 files changed, 251 insertions(+), 97 deletions(-)
diff --git a/backend/helpers/pluginhelper/api/api_extractor_stateful.go b/backend/helpers/pluginhelper/api/api_extractor_stateful.go
index 42fcda261..66ad5281a 100644
--- a/backend/helpers/pluginhelper/api/api_extractor_stateful.go
+++ b/backend/helpers/pluginhelper/api/api_extractor_stateful.go
@@ -18,7 +18,6 @@ limitations under the License.
package api
import (
- "fmt"
"reflect"
"github.com/apache/incubator-devlake/core/dal"
@@ -30,7 +29,6 @@ import (
// StatefulApiExtractorArgs is a struct that contains the arguments for a stateful api extractor
type StatefulApiExtractorArgs struct {
*SubtaskCommonArgs
- Table string // subtask normally reads from a table
Extract func(row *RawData) ([]any, errors.Error)
}
@@ -41,10 +39,6 @@ type StatefulApiExtractor struct {
// NewStatefulApiExtractor creates a new StatefulApiExtractor
func NewStatefulApiExtractor(args *StatefulApiExtractorArgs) (*StatefulApiExtractor, errors.Error) {
- // process args
- if args.BatchSize == 0 {
- args.BatchSize = 500
- }
stateManager, err := NewSubtaskStateManager(args.SubtaskCommonArgs)
if err != nil {
return nil, err
@@ -60,13 +54,14 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
// load data from database
db := extractor.GetDal()
logger := extractor.GetLogger()
- table := fmt.Sprintf("_raw_%s", extractor.Table)
+ table := extractor.GetRawDataTable()
+ params := extractor.GetRawDataParams()
if !db.HasTable(table) {
return nil
}
clauses := []dal.Clause{
dal.From(table),
- dal.Where("params = ?", extractor.Params),
+ dal.Where("params = ?", params),
dal.Orderby("id ASC"),
}
@@ -86,10 +81,10 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
if err != nil {
return errors.Default.Wrap(err, "error running DB query")
}
- logger.Info("get data from %s where params=%s and got %d", table, extractor.Params, count)
+ logger.Info("get data from %s where params=%s and got %d", table, params, count)
defer cursor.Close()
// batch save divider
- divider := NewBatchSaveDivider(extractor.SubTaskContext, extractor.BatchSize, table, extractor.Params)
+ divider := NewBatchSaveDivider(extractor.SubTaskContext, extractor.GetBatchSize(), table, params)
divider.SetIncrementalMode(extractor.IsIncremental())
// progress
@@ -121,8 +116,8 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
// set raw data origin field
setRawDataOrigin(result, common.RawDataOrigin{
RawDataTable: table,
+ RawDataParams: params,
RawDataId: row.ID,
- RawDataParams: row.Params,
})
// records get saved into db when slots were max outed
err = batch.Add(result)
diff --git a/backend/helpers/pluginhelper/api/data_convertor_stateful.go b/backend/helpers/pluginhelper/api/data_convertor_stateful.go
new file mode 100644
index 000000000..ddb710a2f
--- /dev/null
+++ b/backend/helpers/pluginhelper/api/data_convertor_stateful.go
@@ -0,0 +1,129 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package api
+
+import (
+ "reflect"
+
+ "github.com/apache/incubator-devlake/core/dal"
+ "github.com/apache/incubator-devlake/core/errors"
+ plugin "github.com/apache/incubator-devlake/core/plugin"
+)
+
+type StatefulDataConverterArgs[InputType any] struct {
+ *SubtaskCommonArgs
+ Input func(*SubtaskStateManager) (dal.Rows, errors.Error)
+ Convert func(row *InputType) ([]any, errors.Error)
+ BatchSize int
+}
+
+type StatefulDataConverter[InputType any] struct {
+ *StatefulDataConverterArgs[InputType]
+ *SubtaskStateManager
+}
+
+func NewStatefulDataConverter[
+ OptType any,
+ InputType any,
+](
+ args *StatefulDataConverterArgs[InputType],
+) (*StatefulDataConverter[InputType], errors.Error) {
+ // process args
+ if args.BatchSize == 0 {
+ args.BatchSize = 500
+ }
+ stateManager, err := NewSubtaskStateManager(args.SubtaskCommonArgs)
+ if err != nil {
+ return nil, err
+ }
+ return &StatefulDataConverter[InputType]{
+ StatefulDataConverterArgs: args,
+ SubtaskStateManager: stateManager,
+ }, nil
+}
+
+func (converter *StatefulDataConverter[InputType]) Execute() errors.Error {
+ // load data from database
+ db := converter.GetDal()
+
+ table := converter.GetRawDataTable()
+ params := converter.GetRawDataParams()
+
+ // batch save divider
+ RAW_DATA_ORIGIN := "RawDataOrigin"
+ divider := NewBatchSaveDivider(converter, converter.BatchSize, table, params)
+ divider.SetIncrementalMode(converter.IsIncremental())
+
+ // set progress
+ converter.SetProgress(0, -1)
+
+ cursor, err := converter.Input(converter.SubtaskStateManager)
+ if err != nil {
+ return err
+ }
+ defer cursor.Close()
+ ctx := converter.GetContext()
+ // iterate all rows
+ for cursor.Next() {
+ select {
+ case <-ctx.Done():
+ return errors.Convert(ctx.Err())
+ default:
+ }
+ inputRow := new(InputType)
+ err := db.Fetch(cursor, inputRow)
+ if err != nil {
+ return errors.Default.Wrap(err, "error fetching rows")
+ }
+
+ results, err := converter.Convert(inputRow)
+ if err != nil {
+ return errors.Default.Wrap(err, "error calling Converter plugin 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
+ origin := reflect.ValueOf(result).Elem().FieldByName(RAW_DATA_ORIGIN)
+ if origin.IsValid() {
+ origin.Set(reflect.ValueOf(inputRow).Elem().FieldByName(RAW_DATA_ORIGIN))
+ }
+ // records get saved into db when slots were max outed
+ err = batch.Add(result)
+ if err != nil {
+ return errors.Default.Wrap(err, "error adding result to batch")
+ }
+ }
+ converter.IncProgress(1)
+ }
+
+ // save the last batches
+ err = divider.Close()
+ if err != nil {
+ return err
+ }
+ // save the incremantal state
+ return converter.SubtaskStateManager.Close()
+}
+
+// Check if DataConverter implements SubTask interface
+var _ plugin.SubTask = (*StatefulDataConverter[any])(nil)
diff --git a/backend/helpers/pluginhelper/api/subtask_state_manager.go b/backend/helpers/pluginhelper/api/subtask_state_manager.go
index 80d832cb7..22f71453d 100644
--- a/backend/helpers/pluginhelper/api/subtask_state_manager.go
+++ b/backend/helpers/pluginhelper/api/subtask_state_manager.go
@@ -18,22 +18,48 @@ limitations under the License.
package api
import (
+ "fmt"
+ "reflect"
"time"
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/models"
plugin "github.com/apache/incubator-devlake/core/plugin"
+ "github.com/apache/incubator-devlake/core/utils"
)
// SubtaskCommonArgs is a struct that contains the common arguments for a subtask
type SubtaskCommonArgs struct {
plugin.SubTaskContext
- Params string // for filtering rows belonging to the scope (jira board, github repo) of the subtask
- SubtaskConfig string // for determining whether the subtask should run in incremental or full sync mode
+ Table string // raw table name
+ Params any // for filtering rows belonging to the scope (jira board, github repo) of the subtask
+ SubtaskConfig any // for determining whether the subtask should run in incremental or full sync mode
BatchSize int // batch size for saving data
}
+func (args *SubtaskCommonArgs) GetRawDataTable() string {
+ return fmt.Sprintf("_raw_%s", args.Table)
+}
+
+func (args *SubtaskCommonArgs) GetRawDataParams() string {
+ if args.Params == nil || reflect.ValueOf(args.Params).IsZero() {
+ panic(errors.Default.New("Params is nil"))
+ }
+ return utils.ToJsonString(args.Params)
+}
+
+func (args *SubtaskCommonArgs) GetSubtaskConfig() string {
+ return utils.ToJsonString(args.SubtaskConfig)
+}
+
+func (args *SubtaskCommonArgs) GetBatchSize() int {
+ if args.BatchSize == 0 {
+ args.BatchSize = 500
+ }
+ return args.BatchSize
+}
+
// SubtaskStateManager manages the state of a subtask. It is used to determine whether
// the subtask should run in incremental mode or full sync mode and what time range to collect.
type SubtaskStateManager struct {
@@ -56,15 +82,16 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
if syncPolicy == nil {
syncPolicy = &models.SyncPolicy{}
}
+ params := args.GetRawDataParams()
// load the previous state from the database
state := &models.SubtaskState{}
- err = db.First(state, dal.Where(`plugin = ? AND subtask =? AND params = ?`, plugin, subtask, args.Params))
+ err = db.First(state, dal.Where(`plugin = ? AND subtask =? AND params = ?`, plugin, subtask, params))
if err != nil {
if db.IsErrorNotFound(err) {
state = &models.SubtaskState{
Plugin: plugin,
Subtask: subtask,
- Params: args.Params,
+ Params: params,
}
err = nil
} else {
@@ -81,7 +108,7 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
isIncremental: false,
since: syncPolicy.TimeAfter,
until: &now,
- config: args.SubtaskConfig,
+ config: utils.ToJsonString(args.SubtaskConfig),
}
// fallback to the previous timeAfter if no new value
if stateManager.since == nil {
@@ -94,7 +121,7 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
// if timeAfter is not set or NOT before the previous vaule, we are in the incremental mode
if (syncPolicy.TimeAfter == nil || state.TimeAfter == nil || !syncPolicy.TimeAfter.Before(*state.TimeAfter)) &&
// and the previous config is the same as the current config
- (state.PrevConfig == stateManager.config) {
+ (state.PrevConfig == "" || state.PrevConfig == stateManager.config) {
stateManager.isIncremental = true
stateManager.since = state.PrevStartedAt
}
diff --git a/backend/helpers/pluginhelper/api/subtask_state_manager_test.go b/backend/helpers/pluginhelper/api/subtask_state_manager_test.go
index e8645431c..9bbcc4f39 100644
--- a/backend/helpers/pluginhelper/api/subtask_state_manager_test.go
+++ b/backend/helpers/pluginhelper/api/subtask_state_manager_test.go
@@ -101,7 +101,7 @@ func TestSubtaskStateManager(t *testing.T) {
},
{
name: "config no changed",
- state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time2, PrevConfig: "hello"},
+ state: &models.SubtaskState{TimeAfter: &time1, PrevStartedAt: &time2, PrevConfig: `"hello"`},
syncPolicy: &models.SyncPolicy{TimeAfter: &time1},
config: "hello",
expectedIsIncremental: true,
@@ -173,6 +173,7 @@ func TestSubtaskStateManager(t *testing.T) {
stateManager, err := NewSubtaskStateManager(&SubtaskCommonArgs{
SubTaskContext: mockSubtaskCtx,
SubtaskConfig: fmt.Sprintf("%v", tc.config),
+ Params: "whatever",
})
assert.Nil(t, err)
assert.Equal(t, tc.expectedSince, stateManager.since)
diff --git a/backend/plugins/gitextractor/parser/clone_gitcli.go b/backend/plugins/gitextractor/parser/clone_gitcli.go
index 22d4661d0..20f8f200d 100644
--- a/backend/plugins/gitextractor/parser/clone_gitcli.go
+++ b/backend/plugins/gitextractor/parser/clone_gitcli.go
@@ -30,7 +30,6 @@ import (
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/log"
"github.com/apache/incubator-devlake/core/plugin"
- "github.com/apache/incubator-devlake/core/utils"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
)
@@ -52,6 +51,7 @@ func NewGitcliCloner(basicRes context.BasicRes) *GitcliCloner {
// CloneRepoConfig is the configuration for the CloneRepo method
// the subtask should run in Full Sync mode whenever the configuration is changed
type CloneRepoConfig struct {
+ UseGoGit *bool
SkipCommitStat *bool
SkipCommitFiles *bool
NoShallowClone bool
@@ -63,12 +63,13 @@ func (g *GitcliCloner) CloneRepo(ctx plugin.SubTaskContext, localDir string) err
if !taskData.Options.NoShallowClone {
stateManager, err := api.NewSubtaskStateManager(&api.SubtaskCommonArgs{
SubTaskContext: ctx,
- Params: utils.ToJsonString(taskData.Options.GitExtractorApiParams),
- SubtaskConfig: utils.ToJsonString(CloneRepoConfig{
+ Params: taskData.Options.GitExtractorApiParams,
+ SubtaskConfig: CloneRepoConfig{
+ UseGoGit: taskData.Options.UseGoGit,
SkipCommitStat: taskData.Options.SkipCommitStat,
SkipCommitFiles: taskData.Options.SkipCommitFiles,
NoShallowClone: taskData.Options.NoShallowClone,
- }),
+ },
})
if err != nil {
return err
diff --git a/backend/plugins/jira/tasks/issue_changelog_convertor.go b/backend/plugins/jira/tasks/issue_changelog_convertor.go
index cd18d4808..46dc0df2f 100644
--- a/backend/plugins/jira/tasks/issue_changelog_convertor.go
+++ b/backend/plugins/jira/tasks/issue_changelog_convertor.go
@@ -18,7 +18,6 @@ limitations under the License.
package tasks
import (
- "reflect"
"regexp"
"strconv"
"strings"
@@ -52,13 +51,12 @@ type IssueChangelogItemResult struct {
Created time.Time
}
-func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
- data := taskCtx.GetData().(*JiraTaskData)
+func ConvertIssueChangelogs(subtaskCtx plugin.SubTaskContext) errors.Error {
+ data := subtaskCtx.GetData().(*JiraTaskData)
+ db := subtaskCtx.GetDal()
connectionId := data.Options.ConnectionId
boardId := data.Options.BoardId
- logger := taskCtx.GetLogger()
- db := taskCtx.GetDal()
- logger.Info("covert changelog")
+
var allStatus []models.JiraStatus
err := db.All(&allStatus, dal.Where("connection_id = ?", connectionId))
if err != nil {
@@ -68,43 +66,44 @@ func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
for _, v := range allStatus {
statusMap[v.ID] = v
}
- // select all changelogs belongs to the board
- clauses := []dal.Clause{
- dal.Select("_tool_jira_issue_changelog_items.*, _tool_jira_issue_changelogs.issue_id, author_account_id, author_display_name, created"),
- dal.From("_tool_jira_issue_changelog_items"),
- dal.Join(`left join _tool_jira_issue_changelogs on (
- _tool_jira_issue_changelogs.connection_id = _tool_jira_issue_changelog_items.connection_id
- AND _tool_jira_issue_changelogs.changelog_id = _tool_jira_issue_changelog_items.changelog_id
- )`),
- dal.Join(`left join _tool_jira_board_issues on (
- _tool_jira_board_issues.connection_id = _tool_jira_issue_changelogs.connection_id
- AND _tool_jira_board_issues.issue_id = _tool_jira_issue_changelogs.issue_id
- )`),
- dal.Where("_tool_jira_issue_changelog_items.connection_id = ? AND _tool_jira_board_issues.board_id = ?", connectionId, boardId),
- }
- cursor, err := db.Cursor(clauses...)
- if err != nil {
- logger.Error(err, "")
- return err
- }
- defer cursor.Close()
+
issueIdGenerator := didgen.NewDomainIdGenerator(&models.JiraIssue{})
sprintIdGenerator := didgen.NewDomainIdGenerator(&models.JiraSprint{})
changelogIdGenerator := didgen.NewDomainIdGenerator(&models.JiraIssueChangelogItems{})
accountIdGen := didgen.NewDomainIdGenerator(&models.JiraAccount{})
- converter, err := api.NewDataConverter(api.DataConverterArgs{
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
+
+ converter, err := api.NewStatefulDataConverter[IssueChangelogItemResult](&api.StatefulDataConverterArgs[IssueChangelogItemResult]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: subtaskCtx,
+ Table: RAW_ISSUE_TABLE,
Params: JiraApiParams{
- ConnectionId: connectionId,
- BoardId: boardId,
+ ConnectionId: data.Options.ConnectionId,
+ BoardId: data.Options.BoardId,
},
- Table: RAW_CHANGELOG_TABLE,
},
- InputRowType: reflect.TypeOf(IssueChangelogItemResult{}),
- Input: cursor,
- Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
- row := inputRow.(*IssueChangelogItemResult)
+ Input: func(stateManager *api.SubtaskStateManager) (dal.Rows, errors.Error) {
+ clauses := []dal.Clause{
+ dal.Select("_tool_jira_issue_changelog_items.*, _tool_jira_issue_changelogs.issue_id, author_account_id, author_display_name, created"),
+ dal.From("_tool_jira_issue_changelog_items"),
+ dal.Join(`left join _tool_jira_issue_changelogs on (
+ _tool_jira_issue_changelogs.connection_id = _tool_jira_issue_changelog_items.connection_id
+ AND _tool_jira_issue_changelogs.changelog_id = _tool_jira_issue_changelog_items.changelog_id
+ )`),
+ dal.Join(`left join _tool_jira_board_issues on (
+ _tool_jira_board_issues.connection_id = _tool_jira_issue_changelogs.connection_id
+ AND _tool_jira_board_issues.issue_id = _tool_jira_issue_changelogs.issue_id
+ )`),
+ dal.Where("_tool_jira_issue_changelog_items.connection_id = ? AND _tool_jira_board_issues.board_id = ?", connectionId, boardId),
+ }
+ if stateManager.IsIncremental() {
+ since := stateManager.GetSince()
+ if since != nil {
+ clauses = append(clauses, dal.Where("_tool_jira_issue_changelog_items.created_at >= ? ", since))
+ }
+ }
+ return db.Cursor(clauses...)
+ },
+ Convert: func(row *IssueChangelogItemResult) ([]interface{}, errors.Error) {
changelog := &ticket.IssueChangelogs{
DomainEntity: domainlayer.DomainEntity{Id: changelogIdGenerator.Generate(
row.ConnectionId,
@@ -149,10 +148,11 @@ func ConvertIssueChangelogs(taskCtx plugin.SubTaskContext) errors.Error {
}
}
return []interface{}{changelog}, nil
+
},
})
+
if err != nil {
- logger.Info(err.Error())
return err
}
diff --git a/backend/plugins/jira/tasks/issue_changelog_extractor.go b/backend/plugins/jira/tasks/issue_changelog_extractor.go
index 813f57015..2e8b595ba 100644
--- a/backend/plugins/jira/tasks/issue_changelog_extractor.go
+++ b/backend/plugins/jira/tasks/issue_changelog_extractor.go
@@ -22,7 +22,6 @@ import (
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
- "github.com/apache/incubator-devlake/core/utils"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/jira/models"
"github.com/apache/incubator-devlake/plugins/jira/tasks/apiv2models"
@@ -47,12 +46,12 @@ func ExtractIssueChangelogs(subtaskCtx plugin.SubTaskContext) errors.Error {
extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs{
SubtaskCommonArgs: &api.SubtaskCommonArgs{
SubTaskContext: subtaskCtx,
- Params: utils.ToJsonString(JiraApiParams{
+ Table: RAW_CHANGELOG_TABLE,
+ Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
- }),
+ },
},
- Table: RAW_CHANGELOG_TABLE,
Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
// process input
var input apiv2models.Input
diff --git a/backend/plugins/jira/tasks/issue_convertor.go b/backend/plugins/jira/tasks/issue_convertor.go
index e82ae3995..f30037182 100644
--- a/backend/plugins/jira/tasks/issue_convertor.go
+++ b/backend/plugins/jira/tasks/issue_convertor.go
@@ -20,7 +20,6 @@ package tasks
import (
"net/url"
"path/filepath"
- "reflect"
"strings"
"github.com/apache/incubator-devlake/core/dal"
@@ -41,48 +40,51 @@ var ConvertIssuesMeta = plugin.SubTaskMeta{
DomainTypes: []string{plugin.DOMAIN_TYPE_TICKET},
}
-func ConvertIssues(taskCtx plugin.SubTaskContext) errors.Error {
- db := taskCtx.GetDal()
- data := taskCtx.GetData().(*JiraTaskData)
-
- jiraIssue := &models.JiraIssue{}
- // select all issues belongs to the board
- clauses := []dal.Clause{
- dal.Select("_tool_jira_issues.*"),
- dal.From(jiraIssue),
- dal.Join(`left join _tool_jira_board_issues
- on _tool_jira_board_issues.issue_id = _tool_jira_issues.issue_id
- and _tool_jira_board_issues.connection_id = _tool_jira_issues.connection_id`),
- dal.Where(
- "_tool_jira_board_issues.connection_id = ? AND _tool_jira_board_issues.board_id = ?",
- data.Options.ConnectionId,
- data.Options.BoardId,
- ),
- }
- cursor, err := db.Cursor(clauses...)
+func ConvertIssues(subtaskCtx plugin.SubTaskContext) errors.Error {
+ data := subtaskCtx.GetData().(*JiraTaskData)
+ db := subtaskCtx.GetDal()
+ mappings, err := getTypeMappings(data, db)
if err != nil {
return err
}
- defer cursor.Close()
issueIdGen := didgen.NewDomainIdGenerator(&models.JiraIssue{})
accountIdGen := didgen.NewDomainIdGenerator(&models.JiraAccount{})
boardIdGen := didgen.NewDomainIdGenerator(&models.JiraBoard{})
boardId := boardIdGen.Generate(data.Options.ConnectionId, data.Options.BoardId)
- converter, err := api.NewDataConverter(api.DataConverterArgs{
- InputRowType: reflect.TypeOf(models.JiraIssue{}),
- Input: cursor,
- RawDataSubTaskArgs: api.RawDataSubTaskArgs{
- Ctx: taskCtx,
+ converter, err := api.NewStatefulDataConverter[models.JiraIssue](&api.StatefulDataConverterArgs[models.JiraIssue]{
+ SubtaskCommonArgs: &api.SubtaskCommonArgs{
+ SubTaskContext: subtaskCtx,
+ Table: RAW_ISSUE_TABLE,
Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
},
- Table: RAW_ISSUE_TABLE,
+ SubtaskConfig: mappings,
},
- Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
- jiraIssue := inputRow.(*models.JiraIssue)
+ Input: func(stateManager *api.SubtaskStateManager) (dal.Rows, errors.Error) {
+ clauses := []dal.Clause{
+ dal.Select("_tool_jira_issues.*"),
+ dal.From("_tool_jira_issues"),
+ dal.Join(`left join _tool_jira_board_issues
+ on _tool_jira_board_issues.issue_id = _tool_jira_issues.issue_id
+ and _tool_jira_board_issues.connection_id = _tool_jira_issues.connection_id`),
+ dal.Where(
+ "_tool_jira_board_issues.connection_id = ? AND _tool_jira_board_issues.board_id = ?",
+ data.Options.ConnectionId,
+ data.Options.BoardId,
+ ),
+ }
+ if stateManager.IsIncremental() {
+ since := stateManager.GetSince()
+ if since != nil {
+ clauses = append(clauses, dal.Where("_tool_jira_issues.updated_at >= ? ", since))
+ }
+ }
+ return db.Cursor(clauses...)
+ },
+ Convert: func(jiraIssue *models.JiraIssue) ([]interface{}, errors.Error) {
issue := &ticket.Issue{
DomainEntity: domainlayer.DomainEntity{
Id: issueIdGen.Generate(jiraIssue.ConnectionId, jiraIssue.IssueId),
@@ -139,6 +141,7 @@ func ConvertIssues(taskCtx plugin.SubTaskContext) errors.Error {
return result, nil
},
})
+
if err != nil {
return err
}
diff --git a/backend/plugins/jira/tasks/issue_extractor.go b/backend/plugins/jira/tasks/issue_extractor.go
index a76acfabb..bac817402 100644
--- a/backend/plugins/jira/tasks/issue_extractor.go
+++ b/backend/plugins/jira/tasks/issue_extractor.go
@@ -26,7 +26,6 @@ import (
"github.com/apache/incubator-devlake/core/dal"
"github.com/apache/incubator-devlake/core/errors"
"github.com/apache/incubator-devlake/core/plugin"
- "github.com/apache/incubator-devlake/core/utils"
"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
"github.com/apache/incubator-devlake/plugins/jira/models"
"github.com/apache/incubator-devlake/plugins/jira/tasks/apiv2models"
@@ -62,13 +61,13 @@ func ExtractIssues(subtaskCtx plugin.SubTaskContext) errors.Error {
extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs{
SubtaskCommonArgs: &api.SubtaskCommonArgs{
SubTaskContext: subtaskCtx,
- Params: utils.ToJsonString(JiraApiParams{
+ Table: RAW_ISSUE_TABLE,
+ Params: JiraApiParams{
ConnectionId: data.Options.ConnectionId,
BoardId: data.Options.BoardId,
- }),
- SubtaskConfig: utils.ToJsonString(mappings),
+ },
+ SubtaskConfig: mappings,
},
- Table: RAW_ISSUE_TABLE,
Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
return extractIssues(data, mappings, row)
},