[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #3162: [Feature][Connector-V2][My Hours]Add My Hours Source Connector

[GitBox <gi...@apache.org>]

TyrantLucifer commented on code in PR #3162:
URL: https://github.com/apache/incubator-seatunnel/pull/3162#discussion_r1003333146


##########
seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSource.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.myhours.source;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.config.CheckConfigUtil;
+import org.apache.seatunnel.common.config.CheckResult;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
+import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpClientProvider;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse;
+import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSource;
+import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceReader;
+import org.apache.seatunnel.connectors.seatunnel.myhours.source.config.MyHoursSourceConfig;
+import org.apache.seatunnel.connectors.seatunnel.myhours.source.config.MyHoursSourceParameter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.auto.service.AutoService;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+@Slf4j
+@AutoService(SeaTunnelSource.class)
+public class MyHoursSource extends HttpSource {
+    private final MyHoursSourceParameter myHoursSourceParameter = new MyHoursSourceParameter();
+    @Override
+    public String getPluginName() {
+        return "MyHours";
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        CheckResult result = CheckConfigUtil.checkAllExists(pluginConfig, MyHoursSourceConfig.EMAIL, MyHoursSourceConfig.PASSWORD);
+        if (!result.isSuccess()) {
+            throw new PrepareFailException(getPluginName(), PluginType.SOURCE, result.getMsg());
+        }
+        //Login to get accessToken
+        String accessToken = null;
+        try {
+            accessToken = getAccessToken(pluginConfig);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        this.myHoursSourceParameter.buildWithConfig(pluginConfig, accessToken);
+        buildSchemaWithConfig(pluginConfig);
+    }
+
+    @Override
+    public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext readerContext) throws Exception {
+        return new HttpSourceReader(this.myHoursSourceParameter, readerContext, this.deserializationSchema);
+    }
+
+    private String getAccessToken(Config pluginConfig) throws IOException {
+        MyHoursSourceParameter myHoursLoginParameter = new MyHoursSourceParameter();
+        myHoursLoginParameter.buildWithLoginConfig(pluginConfig);
+        HttpClientProvider loginHttpClient = new HttpClientProvider(myHoursLoginParameter);
+        try {
+            HttpResponse response = loginHttpClient.doPost(myHoursLoginParameter.getUrl(), myHoursLoginParameter.getBody());
+            if (HttpResponse.STATUS_OK == response.getCode()) {
+                String content = response.getContent();
+                if (!Strings.isNullOrEmpty(content)) {
+                    ObjectMapper om = new ObjectMapper();
+                    Map<String, String> contentMap = om.readValue(content, Map.class);
+                    return contentMap.get(MyHoursSourceConfig.ACCESSTOKEN);

Review Comment:
   ```suggestion
                       Map<String, String> contentMap = JsonUtils.toMap(content);
                       return contentMap.get(MyHoursSourceConfig.ACCESSTOKEN);
   ```



##########
seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/config/MyHoursSourceConfig.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.myhours.source.config;
+
+import lombok.Data;
+
+@Data

Review Comment:
   Remove useless annotation.



##########
seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/config/MyHoursSourceParameter.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.myhours.source.config;
+
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpConfig;
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Data
+@SuppressWarnings("MagicNumber")

Review Comment:
   Why add this annotation?



##########
seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/config/MyHoursSourceParameter.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.myhours.source.config;
+
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpConfig;
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Data
+@SuppressWarnings("MagicNumber")
+public class MyHoursSourceParameter extends HttpParameter {
+
+    public void buildWithConfig(Config pluginConfig, String accessToken) {
+        // set url
+        if (pluginConfig.hasPath(MyHoursSourceConfig.PROJECTS)) {
+            String projects = pluginConfig.getString(MyHoursSourceConfig.PROJECTS);
+            if (projects.equals(MyHoursSourceConfig.ALL)) {
+                this.setUrl(MyHoursSourceConfig.ALL_PROJECTS_URL);
+            }
+            else {
+                this.setUrl(MyHoursSourceConfig.ACTIVE_PROJECTS_URL);
+            }
+        }
+        else {
+            String users = pluginConfig.getString(MyHoursSourceConfig.USERS);
+            if (users.equals(MyHoursSourceConfig.MEMBER)) {
+                this.setUrl(MyHoursSourceConfig.ALL_MEMBERS_URL);
+            }
+            else {
+                this.setUrl(MyHoursSourceConfig.ALL_CLIENTS_URL);
+            }
+        }
+        // set method
+        this.setMethod(HttpConfig.METHOD_DEFAULT_VALUE);
+        // set headers
+        this.headers = new HashMap<>();
+        this.headers.put(MyHoursSourceConfig.AUTHORIZATION, MyHoursSourceConfig.ACCESSTOKEN_PREFIX + " " + accessToken);
+        this.setHeaders(headers);
+        // set retry
+        if (pluginConfig.hasPath(HttpConfig.RETRY)) {
+            this.setRetry(pluginConfig.getInt(HttpConfig.RETRY));
+            if (pluginConfig.hasPath(HttpConfig.RETRY_BACKOFF_MULTIPLIER_MS)) {
+                this.setRetryBackoffMultiplierMillis(pluginConfig.getInt(HttpConfig.RETRY_BACKOFF_MULTIPLIER_MS));
+            }
+            if (pluginConfig.hasPath(HttpConfig.RETRY_BACKOFF_MAX_MS)) {
+                this.setRetryBackoffMaxMillis(pluginConfig.getInt(HttpConfig.RETRY_BACKOFF_MAX_MS));
+            }
+        }
+    }
+
+    public void buildWithLoginConfig(Config pluginConfig) throws JsonProcessingException {
+        // set url
+        this.setUrl(MyHoursSourceConfig.AUTHORIZATION_URL);
+        // set method
+        this.setMethod(MyHoursSourceConfig.POST);
+        // set body
+        Map bodyParams = new HashMap();
+        String email = pluginConfig.getString(MyHoursSourceConfig.EMAIL);
+        String password = pluginConfig.getString(MyHoursSourceConfig.PASSWORD);
+        bodyParams.put(MyHoursSourceConfig.GRANTTYPE, MyHoursSourceConfig.PASSWORD);
+        bodyParams.put(MyHoursSourceConfig.EMAIL, email);
+        bodyParams.put(MyHoursSourceConfig.PASSWORD, password);
+        bodyParams.put(MyHoursSourceConfig.CLIENTID, MyHoursSourceConfig.API);
+        ObjectMapper om = new ObjectMapper();
+        String body = om.writeValueAsString(bodyParams);

Review Comment:
   ```suggestion
           String body = JsonUtils.toJsonString(bodyParams);
   ```



##########
seatunnel-connectors-v2/connector-http/connector-http-myhours/src/main/java/org/apache/seatunnel/connectors/seatunnel/myhours/source/MyHoursSource.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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 org.apache.seatunnel.connectors.seatunnel.myhours.source;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.config.CheckConfigUtil;
+import org.apache.seatunnel.common.config.CheckResult;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
+import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpClientProvider;
+import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse;
+import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSource;
+import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceReader;
+import org.apache.seatunnel.connectors.seatunnel.myhours.source.config.MyHoursSourceConfig;
+import org.apache.seatunnel.connectors.seatunnel.myhours.source.config.MyHoursSourceParameter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.auto.service.AutoService;
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+@Slf4j
+@AutoService(SeaTunnelSource.class)
+public class MyHoursSource extends HttpSource {
+    private final MyHoursSourceParameter myHoursSourceParameter = new MyHoursSourceParameter();
+    @Override
+    public String getPluginName() {
+        return "MyHours";
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        CheckResult result = CheckConfigUtil.checkAllExists(pluginConfig, MyHoursSourceConfig.EMAIL, MyHoursSourceConfig.PASSWORD);
+        if (!result.isSuccess()) {
+            throw new PrepareFailException(getPluginName(), PluginType.SOURCE, result.getMsg());
+        }
+        //Login to get accessToken
+        String accessToken = null;
+        try {
+            accessToken = getAccessToken(pluginConfig);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        this.myHoursSourceParameter.buildWithConfig(pluginConfig, accessToken);
+        buildSchemaWithConfig(pluginConfig);
+    }
+
+    @Override
+    public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext readerContext) throws Exception {
+        return new HttpSourceReader(this.myHoursSourceParameter, readerContext, this.deserializationSchema);
+    }
+
+    private String getAccessToken(Config pluginConfig) throws IOException {
+        MyHoursSourceParameter myHoursLoginParameter = new MyHoursSourceParameter();
+        myHoursLoginParameter.buildWithLoginConfig(pluginConfig);
+        HttpClientProvider loginHttpClient = new HttpClientProvider(myHoursLoginParameter);
+        try {
+            HttpResponse response = loginHttpClient.doPost(myHoursLoginParameter.getUrl(), myHoursLoginParameter.getBody());
+            if (HttpResponse.STATUS_OK == response.getCode()) {
+                String content = response.getContent();
+                if (!Strings.isNullOrEmpty(content)) {
+                    ObjectMapper om = new ObjectMapper();
+                    Map<String, String> contentMap = om.readValue(content, Map.class);
+                    return contentMap.get(MyHoursSourceConfig.ACCESSTOKEN);
+                }
+            }
+            log.error("login http client execute exception, http response status code:[{}], content:[{}]", response.getCode(), response.getContent());

Review Comment:
   If get token failed, I think should throw exception there not print error message, because without token connector will generate data failed.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@seatunnel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org