diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml
index 91e7b06ecc..2f9a96a0d3 100644
--- a/backend/windmill-api/openapi-deref.yaml
+++ b/backend/windmill-api/openapi-deref.yaml
@@ -1,6 +1,6 @@
 openapi: 3.0.3
 info:
-  version: 1.246.15
+  version: 1.249.0
   title: Windmill API
   contact:
     name: Windmill Team
@@ -7800,6 +7800,8 @@ paths:
                   completed:
                     type: boolean
                   result: {}
+                  success:
+                    type: boolean
                   started:
                     type: boolean
                 required:
@@ -10429,10 +10431,36 @@ paths:
                       - Csv
                       - Parquet
                       - Unknown
-                  download_url:
-                    type: string
                 required: &ref_175
                   - content_type
+  /w/{workspace}/job_helpers/generate_download_url:
+    get:
+      summary: Generate a unique URL to download the file
+      operationId: generateDownloadUrl
+      tags:
+        - helpers
+      parameters:
+        - name: workspace
+          in: path
+          required: true
+          schema: *ref_0
+        - name: file_key
+          in: query
+          required: true
+          schema:
+            type: string
+      responses:
+        '200':
+          description: Download URL
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  download_url:
+                    type: string
+                required:
+                  - download_url
   /w/{workspace}/job_helpers/delete_s3_file:
     delete:
       summary: Permanently delete file from S3
diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml
index ef90b51022..1d4a26267c 100644
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -7009,6 +7009,32 @@ paths:
               schema:
                 $ref: "#/components/schemas/WindmillFilePreview"
 
+  /w/{workspace}/job_helpers/generate_download_url:
+    get:
+      summary: Generate a unique URL to download the file
+      operationId: generateDownloadUrl
+      tags:
+        - helpers
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: file_key
+          in: query
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: Download URL
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  download_url:
+                    type: string
+                required:
+                  - download_url
+
   /w/{workspace}/job_helpers/delete_s3_file:
     delete:
       summary: Permanently delete file from S3
@@ -9281,8 +9307,6 @@ components:
         content_type:
           type: string
           enum: ["RawText", "Csv", "Parquet", "Unknown"]
-        download_url:
-          type: string
       required:
         - content_type
 
diff --git a/backend/windmill-api/src/job_helpers.rs b/backend/windmill-api/src/job_helpers.rs
index 5ceb0cf0b8..656677c520 100644
--- a/backend/windmill-api/src/job_helpers.rs
+++ b/backend/windmill-api/src/job_helpers.rs
@@ -29,6 +29,7 @@ use polars::{
 };
 use serde::{Deserialize, Serialize};
 use tower_http::cors::{Any, CorsLayer};
+use windmill_common::error::JsonResult;
 use windmill_common::{
     db::UserDB,
     error,
@@ -77,6 +78,10 @@ pub fn workspaced_service() -> Router {
             "/load_file_preview",
             get(load_file_preview).layer(cors.clone()),
         )
+        .route(
+            "/generate_download_url",
+            get(generate_download_url).layer(cors.clone()),
+        )
         .route(
             "/delete_s3_file",
             delete(delete_s3_file).layer(cors.clone()),
@@ -450,7 +455,6 @@ struct LoadFilePreviewResponse {
     pub content: Option<String>,
     pub content_type: WindmillContentType,
     pub msg: Option<String>,
-    pub download_url: Option<String>,
 }
 
 #[derive(Serialize)]
@@ -543,19 +547,6 @@ async fn load_file_preview(
             )
         };
 
-    // URL expires 30 minutes after its generation
-    let presigned_config = PresigningConfig::expires_in(Duration::from_secs(60 * 30))
-        .map_err(|err| error::Error::InternalErr(err.to_string()))?;
-    let download_url = s3_client
-        .get_object()
-        .bucket(&s3_bucket)
-        .key(&file_key)
-        .presigned(presigned_config)
-        .await
-        .map_err(|err| error::Error::InternalErr(err.to_string()))?
-        .uri()
-        .to_string();
-
     let file_chunk_length = if s3_object_content_length.is_some() {
         cmp::min(
             query.read_bytes_length,
@@ -645,20 +636,61 @@ async fn load_file_preview(
             content_type: content_type,
             content: Some(content),
             msg: None,
-            download_url: Some(download_url),
         },
 
         Err(err) => LoadFilePreviewResponse {
             content_type: content_type,
             content: None,
             msg: Some(err.to_string()),
-            download_url: Some(download_url),
         },
     };
 
     return Ok(Json(response));
 }
 
+#[derive(Deserialize)]
+struct GenerateDownloadUrlQuery {
+    pub file_key: String,
+}
+
+#[derive(Serialize)]
+struct GenerateDownloadUrlResponse {
+    pub download_url: String,
+}
+
+async fn generate_download_url(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Tokened { token }: Tokened,
+    Path(w_id): Path<String>,
+    Query(query): Query<GenerateDownloadUrlQuery>,
+) -> JsonResult<GenerateDownloadUrlResponse> {
+    let file_key = query.file_key.clone();
+    let s3_resource_opt = get_workspace_s3_resource(&authed, &db, None, &token, &w_id).await?;
+
+    let s3_resource = s3_resource_opt.ok_or(error::Error::InternalErr(
+        "No files storage resource defined at the workspace level".to_string(),
+    ))?;
+
+    let s3_client = build_s3_client(&s3_resource);
+    let s3_bucket = s3_resource.bucket.clone();
+
+    // URL expires 5 minutes after its generation
+    let presigned_config = PresigningConfig::expires_in(Duration::from_secs(60 * 5))
+        .map_err(|err| error::Error::InternalErr(err.to_string()))?;
+    let download_url = s3_client
+        .get_object()
+        .bucket(&s3_bucket)
+        .key(&file_key)
+        .presigned(presigned_config)
+        .await
+        .map_err(|err| error::Error::InternalErr(err.to_string()))?
+        .uri()
+        .to_string();
+
+    return Ok(Json(GenerateDownloadUrlResponse { download_url }));
+}
+
 #[derive(Deserialize)]
 struct DeleteS3FileQuery {
     pub file_key: String,
diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte
index dbcd35736b..d0844b2ac0 100644
--- a/frontend/src/lib/components/DisplayResult.svelte
+++ b/frontend/src/lib/components/DisplayResult.svelte
@@ -2,7 +2,7 @@
 	import { Highlight } from 'svelte-highlight'
 	import { json } from 'svelte-highlight/languages'
 	import TableCustom from './TableCustom.svelte'
-	import { copyToClipboard, roughSizeOfObject, truncate } from '$lib/utils'
+	import { copyToClipboard, emptyString, roughSizeOfObject, truncate } from '$lib/utils'
 	import { Button, Drawer, DrawerContent } from './common'
 	import { ClipboardCopy, Download, Expand, PanelRightOpen, Table2 } from 'lucide-svelte'
 	import Portal from 'svelte-portal'
@@ -10,6 +10,8 @@
 	import S3FilePicker from './S3FilePicker.svelte'
 	import AutoDataTable from './table/AutoDataTable.svelte'
 	import Markdown from 'svelte-exmarkdown'
+	import { HelpersService } from '$lib/gen'
+	import { workspaceStore } from '$lib/stores'
 
 	export let result: any
 	export let requireHtmlApproval = false
@@ -199,6 +201,18 @@
 
 		return result
 	}
+
+	async function downloadS3File(fileKey: string | undefined) {
+		if (emptyString(fileKey)) {
+			return
+		}
+		const downloadUrl = await HelpersService.generateDownloadUrl({
+			workspace: $workspaceStore!,
+			fileKey: fileKey!
+		})
+		console.log('download URL ', downloadUrl.download_url)
+		window.open(downloadUrl.download_url, '_blank')
+	}
 </script>
 
 <div class="inline-highlight relative grow min-h-[200px]">
@@ -347,6 +361,13 @@
 		{:else if !forceJson && resultKind == 's3object'}
 			<div class="absolute top-1 h-full w-full">
 				<Highlight class="" language={json} code={toJsonStr(result).replace(/\\n/g, '\n')} />
+				<button
+					class="text-secondary underline text-2xs whitespace-nowrap"
+					on:click={() => {
+						downloadS3File(result?.s3)
+					}}
+					><span class="flex items-center gap-1"><Download size={12} />download</span>
+				</button>
 				<button
 					class="text-secondary underline text-2xs whitespace-nowrap"
 					on:click={() => {
@@ -359,6 +380,13 @@
 			<div class="absolute top-1 h-full w-full">
 				{#each result as s3object}
 					<Highlight class="" language={json} code={toJsonStr(s3object).replace(/\\n/g, '\n')} />
+					<button
+						class="text-secondary underline text-2xs whitespace-nowrap"
+						on:click={() => {
+							downloadS3File(result?.s3)
+						}}
+						><span class="flex items-center gap-1"><Download size={12} />download</span>
+					</button>
 					<button
 						class="text-secondary text-2xs whitespace-nowrap"
 						on:click={() => {
diff --git a/frontend/src/lib/components/S3FilePicker.svelte b/frontend/src/lib/components/S3FilePicker.svelte
index 27db7026a7..43ae11b699 100644
--- a/frontend/src/lib/components/S3FilePicker.svelte
+++ b/frontend/src/lib/components/S3FilePicker.svelte
@@ -79,7 +79,6 @@
 				fileKey: string
 				contentPreview: string | undefined
 				contentType: string | undefined
-				downloadUrl: string | undefined
 		  }
 		| undefined = undefined
 
@@ -194,8 +193,7 @@
 			filePreview = {
 				fileKey: fileKey,
 				contentPreview: filePreviewContent,
-				contentType: filePreviewRaw.content_type,
-				downloadUrl: filePreviewRaw.download_url
+				contentType: filePreviewRaw.content_type
 			}
 		}
 		filePreviewLoading = false
@@ -260,6 +258,18 @@
 		drawer.openDrawer?.()
 	}
 
+	export async function downloadS3File(fileKey: string | undefined) {
+		if (fileKey === undefined) {
+			return
+		}
+		const downloadUrl = await HelpersService.generateDownloadUrl({
+			workspace: $workspaceStore!,
+			fileKey: fileKey
+		})
+		console.log('download URL ', downloadUrl.download_url)
+		window.open(downloadUrl.download_url, '_blank')
+	}
+
 	async function reloadContent() {
 		if (initialFileKey !== undefined) {
 			initialFileKeyInternalCopy = { ...initialFileKey }
@@ -454,7 +464,9 @@
 											title="Download file from S3"
 											variant="border"
 											color="light"
-											href={filePreview.downloadUrl}
+											on:click={() => {
+												downloadS3File(fileMetadata?.fileKey)
+											}}
 											startIcon={{ icon: Download }}
 											iconOnly={true}
 										/>
diff --git a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte
index ec12fb667f..64206e197c 100644
--- a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte
+++ b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte
@@ -1,5 +1,5 @@
 <script lang="ts">
-	import { copyToClipboard, pluralize, truncate } from '$lib/utils'
+	import { copyToClipboard, emptyString, pluralize, truncate } from '$lib/utils'
 
 	import { createEventDispatcher } from 'svelte'
 	import { Badge } from '../common'
@@ -7,8 +7,10 @@
 	import WarningMessage from './WarningMessage.svelte'
 	import { NEVER_TESTED_THIS_FAR } from '../flows/models'
 	import Portal from 'svelte-portal'
-	import { PanelRightOpen } from 'lucide-svelte'
+	import { Download, PanelRightOpen } from 'lucide-svelte'
 	import S3FilePicker from '../S3FilePicker.svelte'
+	import { HelpersService } from '$lib/gen'
+	import { workspaceStore } from '$lib/stores'
 
 	export let json: any
 	export let level = 0
@@ -55,6 +57,18 @@
 		dispatch('select', rawKey ? key : computeKey(key, isArray, currentPath))
 	}
 
+	async function downloadS3File(fileKey: string | undefined) {
+		if (emptyString(fileKey)) {
+			return
+		}
+		const downloadUrl = await HelpersService.generateDownloadUrl({
+			workspace: $workspaceStore!,
+			fileKey: fileKey!
+		})
+		console.log('download URL ', downloadUrl.download_url)
+		window.open(downloadUrl.download_url, '_blank')
+	}
+
 	$: keyLimit = isArray ? 1 : 100
 
 	$: fullyCollapsed = keys.length > 1 && collapsed
@@ -135,6 +149,13 @@
 			{#if level == 0 && topBrackets}
 				<span class="h-0">{closeBracket}</span>
 				{#if getTypeAsString(json) === 's3object'}
+					<button
+						class="text-secondary underline text-2xs whitespace-nowrap"
+						on:click={() => {
+							downloadS3File(json?.s3)
+						}}
+						><span class="flex items-center gap-1"><Download size={12} />download</span>
+					</button>
 					<button
 						class="text-secondary underline text-2xs whitespace-nowrap ml-1"
 						on:click={() => {
diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py
index 825e9570b1..2da7f7d8eb 100644
--- a/python-client/wmill/wmill/client.py
+++ b/python-client/wmill/wmill/client.py
@@ -13,7 +13,7 @@ from typing import Dict, Any, Union, Literal
 
 import httpx
 
-from .s3_types import Boto3ConnectionSettings, DuckDbConnectionSettings, PolarsConnectionSettings
+from .s3_types import Boto3ConnectionSettings, DuckDbConnectionSettings, PolarsConnectionSettings, S3Object
 
 _client: "Windmill | None" = None
 
@@ -368,20 +368,76 @@ class Windmill:
                 f"/w/{self.workspace}/job_helpers/v2/s3_resource_info",
                 json={} if s3_resource_path == "" else {"s3_resource_path": s3_resource_path},
             ).json()
-            endpoint_url_prefix = "https://" if s3_resource["useSSL"] else "http://"
-            boto3_settings = Boto3ConnectionSettings(
-                {
-                    "endpoint_url": "{}{}".format(endpoint_url_prefix, s3_resource["endPoint"]),
-                    "region_name": s3_resource["region"],
-                    "use_ssl": s3_resource["useSSL"],
-                    "aws_access_key_id": s3_resource["accessKey"],
-                    "aws_secret_access_key": s3_resource["secretKey"],
-                    # no need for path_style here as boto3 is clever enough to determine which one to use
-                }
-            )
-            return boto3_settings
+            return self.__boto3_connection_settings(s3_resource)
         except JSONDecodeError as e:
-            raise Exception("Could not generate Polars S3 connection settings from the provided resource") from e
+            raise Exception("Could not generate Boto3 S3 connection settings from the provided resource") from e
+
+    def load_s3_file(self, s3object: S3Object, s3_resource_path: str = ""):
+        """
+        Load a file from the workspace s3 bucket and returns the bytes stream.
+
+        '''python
+        from wmill import S3Object
+
+        s3_obj = S3Object(s3="/path/to/my_file.txt")
+        my_obj = client.load_s3_file(s3_obj)
+        file_content = my_obj["Body"].read().decode("utf-8")
+        '''
+        """
+        try:
+            s3_resource = self.post(
+                f"/w/{self.workspace}/job_helpers/v2/s3_resource_info",
+                json={} if s3_resource_path == "" else {"s3_resource_path": s3_resource_path},
+            ).json()
+        except JSONDecodeError as e:
+            raise Exception("Could not generate Boto3 S3 connection settings from the provided resource") from e
+
+        import boto3
+
+        args = self.__boto3_connection_settings(s3_resource)
+        s3client = boto3.client("s3", **args)
+        bucket = s3_resource["bucket"]
+        return s3client.get_object(bucket, Key=s3object["s3"])
+
+    def write_s3_file(self, s3object: S3Object, file_content: bytes, s3_resource_path: str = ""):
+        """
+        Write a file to the workspace S3 bucket
+
+        '''python
+        from wmill import S3Object
+
+        s3_obj = S3Object(s3="/path/to/my_file.txt")
+        file_content = b'Hello Windmill!'
+        client.write_s3_file(s3_obj, file_content)
+        '''
+        """
+        try:
+            s3_resource = self.post(
+                f"/w/{self.workspace}/job_helpers/v2/s3_resource_info",
+                json={} if s3_resource_path == "" else {"s3_resource_path": s3_resource_path},
+            ).json()
+        except JSONDecodeError as e:
+            raise Exception("Could not generate Boto3 S3 connection settings from the provided resource") from e
+
+        import boto3
+
+        args = self.__boto3_connection_settings(s3_resource)
+        s3client = boto3.client("s3", **args)
+        bucket = s3_resource["bucket"]
+        s3client.put_object(bucket, Key=s3object["s3"], Body=file_content)
+
+    def __boto3_connection_settings(self, s3_resource) -> Boto3ConnectionSettings:
+        endpoint_url_prefix = "https://" if s3_resource["useSSL"] else "http://"
+        return Boto3ConnectionSettings(
+            {
+                "endpoint_url": "{}{}".format(endpoint_url_prefix, s3_resource["endPoint"]),
+                "region_name": s3_resource["region"],
+                "use_ssl": s3_resource["useSSL"],
+                "aws_access_key_id": s3_resource["accessKey"],
+                "aws_secret_access_key": s3_resource["secretKey"],
+                # no need for path_style here as boto3 is clever enough to determine which one to use
+            }
+        )
 
     def whoami(self) -> dict:
         return self.get("/users/whoami").json()
@@ -582,6 +638,7 @@ def get_id_token(audience: str) -> str:
     """
     return _client.get_id_token(audience)
 
+
 @init_global_client
 def get_job_status(job_id: str) -> JobStatus:
     return _client.get_job_status(job_id)
diff --git a/typescript-client/client.ts b/typescript-client/client.ts
index 4f90a6e2f5..38dfe5753b 100644
--- a/typescript-client/client.ts
+++ b/typescript-client/client.ts
@@ -7,6 +7,7 @@ import {
 } from "./index";
 import { OpenAPI } from "./index";
 import type { DenoS3LightClientSettings } from "./index";
+import { S3Object } from "./s3Types";
 
 export {
   AdminService,
@@ -23,6 +24,7 @@ export {
   UserService,
   WorkspaceService,
 } from "./index";
+import { S3Client } from 'https://deno.land/x/s3_lite_client@0.2.0/mod.ts';
 
 export type Sql = string;
 export type Email = string;
@@ -292,6 +294,25 @@ export async function denoS3LightClientSettings(
   return settings;
 }
 
+export async function loadS3File(
+  s3object: S3Object,
+  s3ResourcePath: string | undefined
+): Promise<Response> {
+  const settings = await denoS3LightClientSettings(s3ResourcePath);
+  const s3 = new S3Client(settings);
+  return await s3.getObject(s3object.s3);
+}
+
+export async function writeS3File(
+  s3object: S3Object,
+  fileContent: ReadableStream<Uint8Array> | Uint8Array | string,
+  s3ResourcePath: string | undefined
+): Promise<Response> {
+  const settings = await denoS3LightClientSettings(s3ResourcePath);
+  const s3 = new S3Client(settings);
+  return await s3.putObject(s3object.s3, fileContent);
+}
+
 /**
  * Get URLs needed for resuming a flow after this step
  * @param approver approver name