jprdonnelly/kestra
1
0
Fork 0
mirror of https://github.com/kestra-io/kestra.git synced 2025-12-19 18:05:41 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat(execution): allow listing the internal storage from the run context

Browse Source 
Part-of: https://github.com/kestra-io/kestra-ee/issues/4228
This commit is contained in:
Loïc Mathieu
2025-12-15 17:32:25 +01:00
parent e302b4be4a
commit 429e7c7945
3 changed files with 10 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
core/src/main/java/io/kestra/core/storages/InternalStorage.java
View File

@@ -1,6 +1,5 @@
package io.kestra.core.storages;
import io.kestra.core.repositories.NamespaceFileMetadataRepositoryInterface;
import io.kestra.core.services.NamespaceService;
import jakarta.annotation.Nullable;
import org.slf4j.Logger;
@@ -272,7 +271,13 @@ public class InternalStorage implements Storage {
return this.storage.put(context.getTenantId(), context.getNamespace(), resolve, new BufferedInputStream(inputStream));
}
@Override
public Optional<StorageContext.Task> getTaskStorageContext() {
return Optional.ofNullable((context instanceof StorageContext.Task task) ? task : null);
}
@Override
public List<FileAttributes> list(URI uri) throws IOException {
return this.storage.list(context.getTenantId(), context.getNamespace(), uri);
}
}

2
core/src/main/java/io/kestra/core/storages/Storage.java
View File

@@ -173,4 +173,6 @@ public interface Storage {
* @return the task storage context
*/
Optional<StorageContext.Task> getTaskStorageContext();
List<FileAttributes> list(URI uri) throws IOException;
}

5
core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
View File

@@ -602,9 +602,8 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
String subflowOutputsBase = (String) taskOutput.get(ExecutableUtils.TASK_VARIABLE_SUBFLOW_OUTPUTS_BASE_URI);
URI subflowOutputsBaseUri = URI.create(StorageContext.KESTRA_PROTOCOL + subflowOutputsBase + "/");
StorageInterface storage = ((DefaultRunContext) runContext).getApplicationContext().getBean(StorageInterface.class);
if (storage.exists(runContext.flowInfo().tenantId(), runContext.flowInfo().namespace(), subflowOutputsBaseUri)) {
List<FileAttributes> list = storage.list(runContext.flowInfo().tenantId(), runContext.flowInfo().namespace(), subflowOutputsBaseUri);
if (runContext.storage().isFileExist(subflowOutputsBaseUri)) {
List<FileAttributes> list = runContext.storage().list(subflowOutputsBaseUri);;
if (!list.isEmpty()) {
// Merge outputs from each sub-flow into a single stored in the internal storage.