feat: Python return datasource object and allow filter by type (#13616)

related: https://github.com/apache/dolphinscheduler-sdk-python/pull/75

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java

@@ -490,29 +490,37 @@ public class PythonGateway {
     }
 
     /**
-     * Get datasource by given datasource name. It return map contain datasource id, type, name.
-     * Useful in Python API create sql task which need datasource information.
+     * Get single datasource by given datasource name. if type is not null,
+     * it will return the datasource match the type.
      *
-     * @param datasourceName user who create or update schedule
+     * @param datasourceName datasource name of datasource
+     * @param type datasource type
      */
-    public Map<String, Object> getDatasourceInfo(String datasourceName) {
-        Map<String, Object> result = new HashMap<>();
+    public DataSource getDatasource(String datasourceName, String type) {
+
         List<DataSource> dataSourceList = dataSourceMapper.queryDataSourceByName(datasourceName);
         if (dataSourceList == null || dataSourceList.isEmpty()) {
             String msg = String.format("Can not find any datasource by name %s", datasourceName);
             log.error(msg);
             throw new IllegalArgumentException(msg);
-        } else if (dataSourceList.size() > 1) {
+        }
+
+        List<DataSource> dataSourceListMatchType = dataSourceList.stream()
+                .filter(dataSource -> type == null || StringUtils.equalsIgnoreCase(dataSource.getType().name(), type))
+                .collect(Collectors.toList());
+
+        log.info("Get the datasource list match the type are: {}", dataSourceListMatchType);
+        if (dataSourceListMatchType.size() > 1) {
             String msg = String.format("Get more than one datasource by name %s", datasourceName);
             log.error(msg);
             throw new IllegalArgumentException(msg);
-        } else {
-            DataSource dataSource = dataSourceList.get(0);
-            result.put("id", dataSource.getId());
-            result.put("type", dataSource.getType().name());
-            result.put("name", dataSource.getName());
         }
-        return result;
+
+        return dataSourceListMatchType.stream().findFirst().orElseThrow(() -> {
+            String msg = String.format("Can not find any datasource by name %s and type %s", datasourceName, type);
+            log.error(msg);
+            return new IllegalArgumentException(msg);
+        });
     }
 
     /**