TBD whether we can support uploading an entire folder: https://stackoverflow.com/questions/49655967/browser-folder-uploading

Otherwise, both Import and Export will be based on a zip file
Add import and export buttons next to Execute in the VS Code Editor
## Import and Export buttons in the Namespace Files Editor

Next to the “Execute” button, we should provide an Import and Export functionality. This will make it easy to:

- Import files developed locally, incl. files users are currently managing via Git
- Build workflows iteratively from Kestra UI. Then export/download and commit to your Git feature branch and open a Pull Request.
- If the directory structure includes flows, the `_flows` will be imported as well.

Export will be zip.

Import may support both:
- a zip file
- an entire directory (preferrable)

(EXTRA: optionally, we may **later** add “Upload directory…” right-click action in VS Code. Atm, it support Upload files only)

## Current design

+ icon for import and the zip folder for Download/Export: https://www.figma.com/file/4tLhsoCsh9yVzYRY1N70kS/App-Kestra?type=design&node-id=3695%3A14501&mode=design&t=Va8ZGcs3qEqtENOW-1

![image](https://github.com/kestra-io/kestra/assets/86264395/57c55dd2-8860-4a48-b15c-c017c84f5080)


core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java core/src/main/java/io/kestra/core/services/FlowService.java jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java ui/src/components/namespace/Editor.vue ui/src/stores/namespaces.js ui/src/translations.json ui/src/utils/utils.js webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java
core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java core/src/main/java/io/kestra/core/services/FlowService.java jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java ui/src/components/namespace/Editor.vue ui/src/stores/namespaces.js ui/src/translations.json ui/src/utils/utils.js webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java
core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java webserver/src/test/java/io/kestra/webserver/controllers/NamespaceFileControllerTest.java
diff --git a/core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java b/core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java
index af53068c6e..d93a31cada 100644
--- a/core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java
+++ b/core/src/main/java/io/kestra/core/repositories/FlowRepositoryInterface.java
@@ -60,6 +60,8 @@ default Optional<FlowWithSource> findByIdWithSource(String tenantId, String name
 
     List<Flow> findByNamespace(String tenantId, String namespace);
 
+    List<FlowWithSource> findByNamespaceWithSource(String tenantId, String namespace);
+
     ArrayListTotal<Flow> find(
         Pageable pageable,
         @Nullable String query,
diff --git a/core/src/main/java/io/kestra/core/services/FlowService.java b/core/src/main/java/io/kestra/core/services/FlowService.java
index 9ed8c6dddb..c8bf70b2c9 100644
--- a/core/src/main/java/io/kestra/core/services/FlowService.java
+++ b/core/src/main/java/io/kestra/core/services/FlowService.java
@@ -4,9 +4,13 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.RunContextFactory;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.serializers.YamlFlowParser;
+import io.kestra.core.tenant.TenantService;
 import io.kestra.core.utils.ListUtils;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.core.annotation.Nullable;
@@ -38,9 +42,36 @@ public class FlowService {
     @Inject
     ConditionService conditionService;
 
+    @Inject
+    FlowRepositoryInterface flowRepository;
+
+    @Inject
+    TenantService tenantService;
+
+    @Inject
+    TaskDefaultService taskDefaultService;
+
+    @Inject
+    YamlFlowParser yamlFlowParser;
+
     @Inject
     ApplicationContext applicationContext;
 
+
+    public void importFlow(String source) {
+        Flow parsed = yamlFlowParser.parse(source, Flow.class);
+        flowRepository
+            .findById(tenantService.resolveTenant(), parsed.getNamespace(), parsed.getId())
+            .ifPresentOrElse(
+                previous -> flowRepository.update(parsed, previous, source, taskDefaultService.injectDefaults(parsed)),
+                () -> flowRepository.create(parsed, source, taskDefaultService.injectDefaults(parsed))
+            );
+    }
+
+    public List<FlowWithSource> findByNamespaceWithSource(String namespace) {
+        return flowRepository.findByNamespaceWithSource(tenantService.resolveTenant(), namespace);
+    }
+
     public Stream<Flow> keepLastVersion(Stream<Flow> stream) {
         return keepLastVersionCollector(stream);
     }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
index 41146419df..90c82f9b19 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
@@ -227,6 +227,28 @@ public List<Flow> findByNamespace(String tenantId, String namespace) {
             });
     }
 
+    @Override
+    public List<FlowWithSource> findByNamespaceWithSource(String tenantId, String namespace) {
+        return this.jdbcRepository
+            .getDslContextWrapper()
+            .transactionResult(configuration -> {
+                SelectConditionStep<Record2<String, String>> select = DSL
+                    .using(configuration)
+                    .select(
+                        field("source_code", String.class),
+                        field("value", String.class)
+                    )
+                    .from(fromLastRevision(true))
+                    .where(field("namespace").eq(namespace))
+                    .and(this.defaultFilter(tenantId));
+
+                return select.fetch().map(record -> FlowWithSource.of(
+                    jdbcRepository.map(record),
+                    record.get("source_code", String.class)
+                ));
+            });
+    }
+
     @SuppressWarnings("unchecked")
     private <R extends Record, E> SelectConditionStep<R> fullTextSelect(String tenantId, DSLContext context, List<Field<Object>> field) {
         ArrayList<Field<Object>> fields = new ArrayList<>(Collections.singletonList(field("value")));
diff --git a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
index e2adc7d51f..ff736b1260 100644
--- a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
+++ b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
@@ -1,13 +1,5 @@
 package io.kestra.repository.memory;
 
-import io.kestra.core.models.SearchResult;
-import io.kestra.core.models.flows.FlowWithSource;
-import io.kestra.core.models.validations.ManualConstraintViolation;
-import io.kestra.core.utils.IdUtils;
-import io.kestra.core.utils.ListUtils;
-import io.micronaut.context.event.ApplicationEventPublisher;
-import io.micronaut.core.value.ValueException;
-import io.micronaut.data.model.Pageable;
 import io.kestra.core.events.CrudEvent;
 import io.kestra.core.events.CrudEventType;
 import io.kestra.core.models.SearchResult;
@@ -21,6 +13,7 @@
 import io.kestra.core.repositories.ArrayListTotal;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.services.FlowService;
+import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
 import io.micronaut.context.event.ApplicationEventPublisher;
 import io.micronaut.core.value.ValueException;
@@ -28,8 +21,6 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
-import org.checkerframework.checker.units.qual.N;
-
 import org.apache.commons.lang3.NotImplementedException;
 
 import javax.annotation.Nullable;
@@ -137,6 +128,17 @@ public List<Flow> findByNamespace(String tenantId, String namespace) {
             .collect(Collectors.toList());
     }
 
+    @Override
+    public List<FlowWithSource> findByNamespaceWithSource(String tenantId, String namespace) {
+        return flows.values()
+            .stream()
+            .filter(flow -> flow.getNamespace().equals(namespace))
+            .filter(flow -> (tenantId == null && flow.getTenantId() == null) || (tenantId != null && tenantId.equals(flow.getTenantId())))
+            .sorted(Comparator.comparingInt(Flow::getRevision))
+            .map(flow -> FlowWithSource.of(flow, FlowService.cleanupSource(findSourceById(tenantId, namespace, flow.getId()).get())))
+            .collect(Collectors.toList());
+    }
+
     @Override
     public ArrayListTotal<Flow> find(
         Pageable pageable,
@@ -169,7 +171,7 @@ public List<FlowWithSource> findWithSource(
         //TODO Non used query, just returns all flow and filter by namespace if set
         return flows.values()
             .stream()
-            .filter(flow ->  namespace == null || flow.getNamespace().equals(namespace) || flow.getNamespace().startsWith(namespace + "."))
+            .filter(flow -> namespace == null || flow.getNamespace().equals(namespace) || flow.getNamespace().startsWith(namespace + "."))
             .filter(flow -> (tenantId == null && flow.getTenantId() == null) || (tenantId != null && tenantId.equals(flow.getTenantId())))
             .filter(flow -> labels == null || labels.isEmpty() || (flow.getLabels() != null && flow.getLabels().stream().anyMatch(label -> labels.containsKey(label.key()) && labels.get(label.key()).equals(label.value()))))
             .sorted(Comparator.comparingInt(Flow::getRevision))
@@ -207,7 +209,7 @@ public FlowWithSource update(Flow flow, Flow previous, String flowSource, Flow f
 
         // control if update is valid
         Optional<ConstraintViolationException> checkUpdate = previous.validateUpdate(flowWithDefaults);
-        if(checkUpdate.isPresent()){
+        if (checkUpdate.isPresent()) {
             throw checkUpdate.get();
         }
 
diff --git a/ui/src/components/namespace/Editor.vue b/ui/src/components/namespace/Editor.vue
index da8399f629..1738e53fe7 100644
--- a/ui/src/components/namespace/Editor.vue
+++ b/ui/src/components/namespace/Editor.vue
@@ -9,6 +9,31 @@
                 allow-create
                 :is-filter="false"
             />
+
+            <el-dropdown>
+                <el-button :icon="Plus" class="p-2 m-0" />
+                <template #dropdown>
+                    <el-dropdown-menu>
+                        <el-dropdown-item :icon="FilePlus" @click="pickFile">
+                            <input ref="filePicker" type="file"
+                                   multiple
+                                   style="display: none"
+                                   @change="importNsFiles" />
+                            {{ $t("namespace files.import.file") }}
+                        </el-dropdown-item>
+                        <el-dropdown-item :icon="FolderPlus" @click="pickFolder">
+                            <input ref="folderPicker" type="file"
+                                   webkitdirectory mozdirectory msdirectory odirectory directory
+                                   style="display: none"
+                                   @change="importNsFiles" />
+                            {{ $t("namespace files.import.folder") }}
+                        </el-dropdown-item>
+                    </el-dropdown-menu>
+                </template>
+            </el-dropdown>
+            <el-tooltip hide-after="50" :content="$t('namespace files.export')">
+                <el-button :icon="FolderZip" class="p-2 m-0" @click="exportNsFiles" />
+            </el-tooltip>
             <trigger-flow
                 :disabled="!flow"
                 :flow-id="flow"
@@ -35,6 +60,10 @@
     import NamespaceSelect from "./NamespaceSelect.vue";
     import TopNavBar from "../layout/TopNavBar.vue";
     import TriggerFlow from "../flows/TriggerFlow.vue";
+    import Plus from "vue-material-design-icons/Plus.vue";
+    import FolderPlus from "vue-material-design-icons/FolderPlus.vue";
+    import FilePlus from "vue-material-design-icons/FilePlus.vue";
+    import FolderZip from "vue-material-design-icons/FolderZip.vue";
 </script>
 
 <script>
@@ -47,6 +76,40 @@
     export default {
         mixins: [RouteContext, RestoreUrl],
         methods: {
+            importNsFiles(event) {
+                const files = [...event.target.files];
+                Promise.all(files.map(file => {
+                    return this.$store
+                        .dispatch("namespace/importFile", {
+                            namespace: this.namespace,
+                            file: file
+                        })
+                })).then(() => {
+                    this.$message({
+                        message: this.$t("namespace files.import.success"),
+                        type: "success"
+                    });
+                }).catch(() => {
+                    this.$message({
+                        message: this.$t("namespace files.import.error"),
+                        type: "error"
+                    });
+                }).finally(() => {
+                    this.$refs.vscodeIde.contentDocument.location.reload(true);
+                    event.target.value = "";
+                });
+            },
+            exportNsFiles() {
+                this.$store.dispatch("namespace/exportFiles", {
+                    namespace: this.namespace
+                });
+            },
+            pickFile() {
+                this.$refs.filePicker.click();
+            },
+            pickFolder() {
+                this.$refs.folderPicker.click();
+            },
             namespaceUpdate(namespace) {
                 localStorage.setItem(storageKeys.LATEST_NAMESPACE, namespace);
                 this.$router.push({
@@ -66,7 +129,8 @@
         data() {
             return {
                 flow: null,
-                tabsNotSaved: []
+                tabsNotSaved: [],
+                uploadFileName: undefined
             }
         },
         mounted() {
diff --git a/ui/src/stores/namespaces.js b/ui/src/stores/namespaces.js
index 5597ff3f9e..8e2ecdd1d5 100644
--- a/ui/src/stores/namespaces.js
+++ b/ui/src/stores/namespaces.js
@@ -1,4 +1,5 @@
 import {apiUrl} from "override/utils/route";
+import Utils from "../utils/utils";
 
 export default {
     namespaced: true,
@@ -12,6 +13,29 @@ export default {
                 commit("setNamespaces", response.data)
             })
         },
+        importFile({commit}, options) {
+            const file = options.file;
+            const formData = new FormData();
+            formData.append("fileContent", file);
+
+            let path;
+            if((file.webkitRelativePath ?? "") === "") {
+                path = file.name;
+            } else {
+                path = file.webkitRelativePath.split("/").slice(1).join("/");
+            }
+            path = path.replace(" ", "_");
+
+            return this.$http.post(
+                `${apiUrl(this)}/namespaces/${options.namespace}/files?path=/${path}`,
+                formData,
+                {headers: {"Content-Type": "multipart/form-data"}}
+            );
+        },
+        exportFiles({commit}, options) {
+            this.$http.get(`${apiUrl(this)}/namespaces/${options.namespace}/files/export`)
+                .then(response => Utils.downloadUrl(response.request.responseURL, options.namespace + "_files.zip"));
+        }
     },
     mutations: {
         setNamespaces(state, namespaces) {
diff --git a/ui/src/translations.json b/ui/src/translations.json
index 989073a7a4..0cee0c3106 100644
--- a/ui/src/translations.json
+++ b/ui/src/translations.json
@@ -523,7 +523,16 @@
     "show": "Show",
     "advanced configuration": "Advanced configuration",
     "all executions": "All executions",
-    "trigger execution id": "Parent Id"
+    "trigger execution id": "Parent Id",
+    "namespace files": {
+      "import": {
+        "file": "Import from ZIP or add file(s)",
+        "folder": "Import from folder",
+        "success": "File(s) successfully imported",
+        "error": "Error(s) occurred while importing file(s)"
+      },
+      "export": "Export as ZIP file"
+    }
   },
   "fr": {
     "id": "Identifiant",
@@ -1039,7 +1048,16 @@
     "row count": "Nombre de lignes",
     "encoding": "Encodage",
     "all executions": "Toutes les exécutions",
-    "trigger execution id": "Id parent"
+    "trigger execution id": "Id parent",
+    "namespace files": {
+      "import": {
+        "file": "Importer depuis un ZIP ou ajouter des fichiers",
+        "folder": "Importer depuis un dossier",
+        "success": "Fichier(s) importé(s) avec succès",
+        "error": "Une erreur est survenue lors de l'importation"
+      },
+      "export": "Export as ZIP file"
+    }
   }
 }
 
diff --git a/ui/src/utils/utils.js b/ui/src/utils/utils.js
index 9fae611ee3..a9a75a4bc3 100644
--- a/ui/src/utils/utils.js
+++ b/ui/src/utils/utils.js
@@ -152,8 +152,10 @@ export default class Utils {
         const link = document.createElement("a");
         link.href = url;
         link.setAttribute("download", filename);
+        link.setAttribute("target", "_blank");
         document.body.appendChild(link);
         link.click();
+        document.body.removeChild(link);
     }
 
     static switchTheme(theme) {
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java b/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
index 0d08f8d6ea..ef61bc23b4 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
@@ -713,8 +713,7 @@ public HttpResponse<Void> importFlows(
         if (fileName.endsWith(".yaml") || fileName.endsWith(".yml")) {
             List<String> sources = List.of(new String(fileUpload.getBytes()).split("---"));
             for (String source : sources) {
-                Flow parsed = yamlFlowParser.parse(source, Flow.class);
-                importFlow(source, parsed);
+                flowService.importFlow(source);
             }
         } else if (fileName.endsWith(".zip")) {
             try (ZipInputStream archive = new ZipInputStream(fileUpload.getInputStream())) {
@@ -725,8 +724,7 @@ public HttpResponse<Void> importFlows(
                     }
 
                     String source = new String(archive.readAllBytes());
-                    Flow parsed = yamlFlowParser.parse(source, Flow.class);
-                    importFlow(source, parsed);
+                    flowService.importFlow(source);
                 }
             }
         } else {
@@ -736,15 +734,6 @@ public HttpResponse<Void> importFlows(
         return HttpResponse.status(HttpStatus.NO_CONTENT);
     }
 
-    protected void importFlow(String source, Flow parsed) {
-        flowRepository
-            .findById(tenantService.resolveTenant(), parsed.getNamespace(), parsed.getId())
-            .ifPresentOrElse(
-                previous -> flowRepository.update(parsed, previous, source, taskDefaultService.injectDefaults(parsed)),
-                () -> flowRepository.create(parsed, source, taskDefaultService.injectDefaults(parsed))
-            );
-    }
-
     protected List<FlowWithSource> setFlowsDisableByIds(List<IdWithNamespace> ids, boolean disable) {
         return ids
             .stream()
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java b/webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java
index 35acca43f8..3801345788 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/NamespaceFileController.java
@@ -1,11 +1,13 @@
 package io.kestra.webserver.controllers;
 
+import io.kestra.core.services.FlowService;
 import io.kestra.core.storages.FileAttributes;
 import io.kestra.core.storages.ImmutableFileAttributes;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.tenant.TenantService;
 import io.kestra.core.utils.Rethrow;
 import io.micronaut.core.annotation.Nullable;
+import io.micronaut.http.HttpResponse;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.annotation.*;
 import io.micronaut.http.multipart.CompletedFileUpload;
@@ -25,15 +27,21 @@
 import java.util.*;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
 
 @Slf4j
 @Validated
 @Controller("/api/v1/namespaces")
 public class NamespaceFileController {
+    public static final String FLOWS_FOLDER = "_flows";
     @Inject
     private StorageInterface storageInterface;
     @Inject
     private TenantService tenantService;
+    @Inject
+    private FlowService flowService;
 
     private final List<StaticFile> staticFiles;
     {
@@ -50,7 +58,7 @@ public class NamespaceFileController {
     }
 
     private final List<Pattern> forbiddenPathPatterns = List.of(
-        Pattern.compile("/_flows.*")
+        Pattern.compile("/" + FLOWS_FOLDER + ".*")
     );
 
 
@@ -151,10 +159,73 @@ public void createFile(
         @Parameter(description = "The internal storage uri") @QueryValue URI path,
         @Part CompletedFileUpload fileContent
     ) throws IOException, URISyntaxException {
-        ensureWritableFile(path);
+        ensureNonReadOnly(path);
+
+        if(fileContent.getFilename().toLowerCase().endsWith(".zip")) {
+            try (ZipInputStream archive = new ZipInputStream(fileContent.getInputStream())) {
+                ZipEntry entry;
+                while ((entry = archive.getNextEntry()) != null) {
+                    if (entry.isDirectory()) {
+                        continue;
+                    }
+
+                    putNamespaceFile(namespace, URI.create("/" + entry.getName()), new BufferedInputStream(new ByteArrayInputStream(archive.readAllBytes())));
+                }
+            }
+        } else {
+            try(BufferedInputStream inputStream = new BufferedInputStream(fileContent.getInputStream())) {
+                putNamespaceFile(namespace, path, inputStream);
+            }
+        }
+    }
+
+    private void putNamespaceFile(String namespace, URI path, BufferedInputStream inputStream) throws IOException {
+        String filePath = path.getPath();
+        if(filePath.matches("/" + FLOWS_FOLDER + "/.*")) {
+            if(filePath.split("/").length != 3) {
+                throw new IllegalArgumentException("Invalid flow file path: " + filePath);
+            }
 
-        try(BufferedInputStream inputStream = new BufferedInputStream(fileContent.getInputStream())) {
-            storageInterface.put(tenantService.resolveTenant(), toNamespacedStorageUri(namespace, path), inputStream);
+            String flowSource = new String(inputStream.readAllBytes());
+            flowSource = flowSource.replaceFirst("(?m)^namespace: .*$", "namespace: " + namespace);
+            flowService.importFlow(flowSource);
+            return;
+        }
+
+        storageInterface.put(tenantService.resolveTenant(), toNamespacedStorageUri(namespace, path), inputStream);
+    }
+
+    @ExecuteOn(TaskExecutors.IO)
+    @Get(uri = "{namespace}/files/export", produces = MediaType.APPLICATION_OCTET_STREAM)
+    @Operation(tags = {"Files"}, summary = "Export namespace files as a ZIP")
+    public HttpResponse<byte[]> export(
+        @Parameter(description = "The namespace id") @PathVariable String namespace
+        ) throws IOException {
+        try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+             ZipOutputStream archive = new ZipOutputStream(bos)) {
+
+            URI baseNamespaceFilesUri = toNamespacedStorageUri(namespace, null);
+            storageInterface.filesByPrefix(tenantService.resolveTenant(), baseNamespaceFilesUri).forEach(Rethrow.throwConsumer(uri -> {
+                try (InputStream inputStream = storageInterface.get(tenantService.resolveTenant(), uri)) {
+                    archive.putNextEntry(new ZipEntry(baseNamespaceFilesUri.relativize(uri).getPath()));
+                    archive.write(inputStream.readAllBytes());
+                    archive.closeEntry();
+                }
+            }));
+
+            flowService.findByNamespaceWithSource(namespace).forEach(Rethrow.throwConsumer(flowWithSource -> {
+                try {
+                    archive.putNextEntry(new ZipEntry(FLOWS_FOLDER + "/" + flowWithSource.getId() + ".yml"));
+                    archive.write(flowWithSource.getSource().getBytes());
+                    archive.closeEntry();
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }));
+
+            archive.finish();
+
+            return HttpResponse.ok(bos.toByteArray()).header("Content-Disposition", "attachment; filename=\"" + namespace + "_files.zip\"");
         }
     }
 
@@ -166,8 +237,8 @@ public void move(
         @Parameter(description = "The internal storage uri to move from") @QueryValue URI from,
         @Parameter(description = "The internal storage uri to move to") @QueryValue URI to
     ) throws IOException, URISyntaxException {
-        ensureWritableFile(from);
-        ensureWritableFile(to);
+        ensureWritableNamespaceFile(from);
+        ensureWritableNamespaceFile(to);
 
         storageInterface.move(tenantService.resolveTenant(), toNamespacedStorageUri(namespace, from), toNamespacedStorageUri(namespace, to));
     }
@@ -179,7 +250,7 @@ public void delete(
         @Parameter(description = "The namespace id") @PathVariable String namespace,
         @Parameter(description = "The internal storage uri of the file / directory to delete") @QueryValue URI path
     ) throws IOException, URISyntaxException {
-        ensureWritableFile(path);
+        ensureWritableNamespaceFile(path);
 
         storageInterface.delete(tenantService.resolveTenant(), toNamespacedStorageUri(namespace, path));
     }
@@ -198,9 +269,13 @@ private URI toNamespacedStorageUri(String namespace, @Nullable URI relativePath)
         return URI.create("kestra://" + storageInterface.namespaceFilePrefix(namespace) + Optional.ofNullable(relativePath).map(URI::getPath).orElse("/"));
     }
 
-    private void ensureWritableFile(URI path) {
+    private void ensureWritableNamespaceFile(URI path) {
         forbiddenPathsGuard(path);
 
+        ensureNonReadOnly(path);
+    }
+
+    private void ensureNonReadOnly(URI path) {
         Optional<StaticFile> maybeStaticFile = staticFiles.stream().filter(staticFile -> path.getPath().equals(staticFile.getServedPath())).findFirst();
         if(maybeStaticFile.isPresent()) {
             throw new IllegalArgumentException("'" + maybeStaticFile.get().getServedPath().replaceFirst("^/", "") + "' file is read-only");

diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
index e1b55c09a7..8dabd12951 100644
--- a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
+++ b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
@@ -13,9 +13,9 @@
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.runners.FlowListeners;
 import io.kestra.core.schedulers.AbstractSchedulerTest;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.services.FlowService;
 import io.kestra.core.services.TaskDefaultService;
 import io.kestra.core.tasks.debugs.Return;
 import io.kestra.core.tasks.flows.Template;
@@ -25,7 +25,6 @@
 import io.kestra.core.utils.TestsUtils;
 import io.micronaut.context.event.ApplicationEventListener;
 import io.micronaut.data.model.Pageable;
-import io.micronaut.data.model.Sort;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
@@ -239,6 +238,20 @@ void findByNamespace() {
         assertThat((long) save.size(), is(1L));
     }
 
+    @Test
+    void findByNamespaceWithSource() {
+        Flow flow = builder()
+            .revision(3)
+            .build();
+        String flowSource = "# comment\n" + flow.generateSource();
+        flowRepository.create(flow, flowSource, taskDefaultService.injectDefaults(flow));
+
+        List<FlowWithSource> save = flowRepository.findByNamespaceWithSource(null, flow.getNamespace());
+        assertThat((long) save.size(), is(1L));
+
+        assertThat(save.get(0).getSource(), is(FlowService.cleanupSource(flowSource)));
+    }
+
     @Test
     void find() {
         List<Flow> save = flowRepository.find(Pageable.from(1, 10),null, null, "io.kestra.tests", Collections.emptyMap());
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/NamespaceFileControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/NamespaceFileControllerTest.java
index cbd411488c..402e225f7e 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/NamespaceFileControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/NamespaceFileControllerTest.java
@@ -1,9 +1,13 @@
 package io.kestra.webserver.controllers;
 
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.storages.FileAttributes;
 import io.kestra.core.storages.StorageInterface;
+import io.kestra.webserver.controllers.h2.JdbcH2ControllerTest;
 import io.micronaut.core.annotation.Nullable;
+import io.micronaut.core.type.Argument;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.client.annotation.Client;
@@ -21,6 +25,7 @@
 import org.junit.jupiter.api.function.Executable;
 
 import java.io.ByteArrayInputStream;
+import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
@@ -37,7 +42,7 @@
 import static org.hamcrest.core.Is.is;
 
 @MicronautTest
-class NamespaceFileControllerTest {
+class NamespaceFileControllerTest extends JdbcH2ControllerTest {
     private static final String NAMESPACE = "io.namespace";
     private static final String GETTING_STARTED_CONTENT;
 
@@ -57,9 +62,17 @@ class NamespaceFileControllerTest {
     @Inject
     private StorageInterface storageInterface;
 
+    @Inject
+    private FlowRepositoryInterface flowRepository;
+
     @BeforeEach
     public void init() throws IOException {
         storageInterface.delete(null, toNamespacedStorageUri(NAMESPACE, null));
+
+        flowRepository.findAll(null)
+            .forEach(flowRepository::delete);
+
+        super.setup();
     }
 
     @SuppressWarnings("unchecked")
@@ -156,9 +169,73 @@ void createFile() throws IOException {
             HttpRequest.POST("/api/v1/namespaces/" + NAMESPACE + "/files?path=/test.txt", body)
                 .contentType(MediaType.MULTIPART_FORM_DATA_TYPE)
         );
-        InputStream inputStream = storageInterface.get(null, toNamespacedStorageUri(NAMESPACE, URI.create("/test.txt")));
+        assertNamespaceFileContent(URI.create("/test.txt"), "Hello");
+    }
+
+    @Test
+    void createFile_AddFlow() throws IOException {
+        String flowSource = flowRepository.findByIdWithSource(null, "io.kestra.tests", "task-flow").get().getSource();
+        File temp = File.createTempFile("task-flow", ".yml");
+        Files.write(temp.toPath(), flowSource.getBytes());
+
+        assertThat(flowRepository.findByIdWithSource(null, NAMESPACE, "task-flow").isEmpty(), is(true));
+
+        MultipartBody body = MultipartBody.builder()
+            .addPart("fileContent", "task-flow.yml", temp)
+            .build();
+        client.toBlocking().exchange(
+            HttpRequest.POST("/api/v1/namespaces/" + NAMESPACE + "/files?path=/_flows/task-flow.yml", body)
+                .contentType(MediaType.MULTIPART_FORM_DATA_TYPE)
+        );
+
+        assertThat(
+            flowRepository.findByIdWithSource(null, NAMESPACE, "task-flow").get().getSource(),
+            is(flowSource.replaceFirst("(?m)^namespace: .*$", "namespace: " + NAMESPACE))
+        );
+
+        assertThat(storageInterface.exists(null, toNamespacedStorageUri(NAMESPACE, URI.create("/_flows/task-flow.yml"))), is(false));
+    }
+
+    @Test
+    void createFile_ExtractZip() throws IOException {
+        String namespaceToExport = "io.kestra.tests";
+
+        storageInterface.put(null, toNamespacedStorageUri(namespaceToExport, URI.create("/file.txt")), new ByteArrayInputStream("file".getBytes()));
+        storageInterface.put(null, toNamespacedStorageUri(namespaceToExport, URI.create("/another_file.txt")), new ByteArrayInputStream("another_file".getBytes()));
+        storageInterface.put(null, toNamespacedStorageUri(namespaceToExport, URI.create("/folder/file.txt")), new ByteArrayInputStream("folder_file".getBytes()));
+        storageInterface.createDirectory(null, toNamespacedStorageUri(namespaceToExport, URI.create("/empty_folder")));
+
+        byte[] zip = client.toBlocking().retrieve(HttpRequest.GET("/api/v1/namespaces/" + namespaceToExport + "/files/export"),
+            Argument.of(byte[].class));
+        File temp = File.createTempFile("files", ".zip");
+        Files.write(temp.toPath(), zip);
+
+        assertThat(flowRepository.findById(null, NAMESPACE, "task-flow").isEmpty(), is(true));
+
+        MultipartBody body = MultipartBody.builder()
+            .addPart("fileContent", "files.zip", temp)
+            .build();
+        client.toBlocking().exchange(
+            HttpRequest.POST("/api/v1/namespaces/" + NAMESPACE + "/files?path=/files.zip", body)
+                .contentType(MediaType.MULTIPART_FORM_DATA_TYPE)
+        );
+
+        assertNamespaceFileContent(URI.create("/file.txt"), "file");
+        assertNamespaceFileContent(URI.create("/another_file.txt"), "another_file");
+        assertThat(storageInterface.exists(null, toNamespacedStorageUri(NAMESPACE, URI.create("/folder"))), is(true));
+        assertNamespaceFileContent(URI.create("/folder/file.txt"), "folder_file");
+        // Highlights the fact that we currently don't export / import empty folders (would require adding a method to storages to also retrieve folders)
+        assertThat(storageInterface.exists(null, toNamespacedStorageUri(NAMESPACE, URI.create("/empty_folder"))), is(false));
+
+        Flow retrievedFlow = flowRepository.findById(null, NAMESPACE, "task-flow").get();
+        assertThat(retrievedFlow.getNamespace(), is(NAMESPACE));
+        assertThat(((io.kestra.core.tasks.flows.Flow) retrievedFlow.getTasks().get(0)).getNamespace(), is(namespaceToExport));
+    }
+
+    private void assertNamespaceFileContent(URI fileUri, String expectedContent) throws IOException {
+        InputStream inputStream = storageInterface.get(null, toNamespacedStorageUri(NAMESPACE, fileUri));
         String content = new String(inputStream.readAllBytes());
-        assertThat(content, is("Hello"));
+        assertThat(content, is(expectedContent));
     }
 
     @Test
@@ -222,13 +299,6 @@ void forbiddenPaths() {
         assertForbiddenErrorThrown(() -> client.toBlocking().retrieve(HttpRequest.GET("/api/v1/namespaces/" + NAMESPACE + "/files?path=/_flows/test.yml")));
         assertForbiddenErrorThrown(() -> client.toBlocking().retrieve(HttpRequest.GET("/api/v1/namespaces/" + NAMESPACE + "/files/stats?path=/_flows/test.yml"), TestFileAttributes.class));
         assertForbiddenErrorThrown(() -> client.toBlocking().retrieve(HttpRequest.GET("/api/v1/namespaces/" + NAMESPACE + "/files/directory?path=/_flows"), TestFileAttributes[].class));
-        assertForbiddenErrorThrown(() -> client.toBlocking().exchange(HttpRequest.POST("/api/v1/namespaces/" + NAMESPACE + "/files/directory?path=/_flows/test", null)));
-        assertForbiddenErrorThrown(() -> {
-            MultipartBody body = MultipartBody.builder()
-                .addPart("fileContent", "test.txt", "Hello".getBytes())
-                .build();
-            client.toBlocking().exchange(HttpRequest.POST("/api/v1/namespaces/" + NAMESPACE + "/files?path=/_flows/test.txt", body).contentType(MediaType.MULTIPART_FORM_DATA_TYPE));
-        });
         assertForbiddenErrorThrown(() -> client.toBlocking().exchange(HttpRequest.PUT("/api/v1/namespaces/" + NAMESPACE + "/files?from=/_flows/test&to=/foo", null)));
         assertForbiddenErrorThrown(() -> client.toBlocking().exchange(HttpRequest.PUT("/api/v1/namespaces/" + NAMESPACE + "/files?from=/foo&to=/_flows/test", null)));
         assertForbiddenErrorThrown(() -> client.toBlocking().exchange(HttpRequest.DELETE("/api/v1/namespaces/" + NAMESPACE + "/files?path=/_flows/test.txt", null)));
