s3/
put_object_request.rs

1//! Builder pattern for S3 PUT operations with customizable options
2//!
3//! This module provides a builder pattern for constructing PUT requests with
4//! various options including custom headers, content type, and other metadata.
5
6use crate::error::S3Error;
7use crate::request::{Request as _, ResponseData};
8use crate::{Bucket, command::Command};
9use http::{HeaderMap, HeaderName, HeaderValue};
10
11#[cfg(feature = "with-tokio")]
12use tokio::io::AsyncRead;
13
14#[cfg(feature = "with-async-std")]
15use async_std::io::Read as AsyncRead;
16
17#[cfg(feature = "with-async-std")]
18use crate::request::async_std_backend::SurfRequest as RequestImpl;
19#[cfg(feature = "sync")]
20use crate::request::blocking::AttoRequest as RequestImpl;
21#[cfg(feature = "with-tokio")]
22use crate::request::tokio_backend::ReqwestRequest as RequestImpl;
23
24/// Builder for constructing S3 PUT object requests with custom options
25///
26/// # Example
27/// ```no_run
28/// use s3::bucket::Bucket;
29/// use s3::creds::Credentials;
30/// use anyhow::Result;
31///
32/// # #[tokio::main]
33/// # async fn main() -> Result<()> {
34/// let bucket = Bucket::new("my-bucket", "us-east-1".parse()?, Credentials::default()?)?;
35///
36/// // Upload with custom headers using builder pattern
37/// #[cfg(not(feature = "sync"))]
38/// let response = bucket.put_object_builder("/my-file.txt", b"Hello, World!")
39///     .with_content_type("text/plain")
40///     .with_cache_control("public, max-age=3600")?
41///     .with_content_encoding("gzip")?
42///     .execute()
43///     .await?;
44/// #[cfg(feature = "sync")]
45/// let response = bucket.put_object_builder("/my-file.txt", b"Hello, World!")
46///     .with_content_type("text/plain")
47///     .with_cache_control("public, max-age=3600")?
48///     .with_content_encoding("gzip")?
49///     .execute()?;
50/// # Ok(())
51/// # }
52/// ```
53#[derive(Debug, Clone)]
54pub struct PutObjectRequest<'a> {
55    bucket: &'a Bucket,
56    path: String,
57    content: Vec<u8>,
58    content_type: String,
59    custom_headers: HeaderMap,
60}
61
62impl<'a> PutObjectRequest<'a> {
63    /// Create a new PUT object request builder
64    pub(crate) fn new<S: AsRef<str>>(bucket: &'a Bucket, path: S, content: &[u8]) -> Self {
65        Self {
66            bucket,
67            path: path.as_ref().to_string(),
68            content: content.to_vec(),
69            content_type: "application/octet-stream".to_string(),
70            custom_headers: HeaderMap::new(),
71        }
72    }
73
74    /// Set the Content-Type header
75    pub fn with_content_type<S: AsRef<str>>(mut self, content_type: S) -> Self {
76        self.content_type = content_type.as_ref().to_string();
77        self
78    }
79
80    /// Set the Cache-Control header
81    pub fn with_cache_control<S: AsRef<str>>(mut self, cache_control: S) -> Result<Self, S3Error> {
82        let value = cache_control
83            .as_ref()
84            .parse::<HeaderValue>()
85            .map_err(S3Error::InvalidHeaderValue)?;
86        self.custom_headers
87            .insert(http::header::CACHE_CONTROL, value);
88        Ok(self)
89    }
90
91    /// Set the Content-Encoding header
92    pub fn with_content_encoding<S: AsRef<str>>(mut self, encoding: S) -> Result<Self, S3Error> {
93        let value = encoding
94            .as_ref()
95            .parse::<HeaderValue>()
96            .map_err(S3Error::InvalidHeaderValue)?;
97        self.custom_headers
98            .insert(http::header::CONTENT_ENCODING, value);
99        Ok(self)
100    }
101
102    /// Set the Content-Disposition header
103    pub fn with_content_disposition<S: AsRef<str>>(
104        mut self,
105        disposition: S,
106    ) -> Result<Self, S3Error> {
107        let value = disposition
108            .as_ref()
109            .parse::<HeaderValue>()
110            .map_err(S3Error::InvalidHeaderValue)?;
111        self.custom_headers
112            .insert(http::header::CONTENT_DISPOSITION, value);
113        Ok(self)
114    }
115
116    /// Set the Expires header
117    pub fn with_expires<S: AsRef<str>>(mut self, expires: S) -> Result<Self, S3Error> {
118        let value = expires
119            .as_ref()
120            .parse::<HeaderValue>()
121            .map_err(S3Error::InvalidHeaderValue)?;
122        self.custom_headers.insert(http::header::EXPIRES, value);
123        Ok(self)
124    }
125
126    /// Add a custom header
127    pub fn with_header<V>(mut self, key: &str, value: V) -> Result<Self, S3Error>
128    where
129        V: AsRef<str>,
130    {
131        let header_name = HeaderName::from_bytes(key.as_bytes())?;
132        let header_value = value
133            .as_ref()
134            .parse::<HeaderValue>()
135            .map_err(S3Error::InvalidHeaderValue)?;
136        self.custom_headers.insert(header_name, header_value);
137        Ok(self)
138    }
139
140    /// Add multiple custom headers (already validated HeaderMap)
141    pub fn with_headers(mut self, headers: HeaderMap) -> Self {
142        self.custom_headers.extend(headers);
143        self
144    }
145
146    /// Add S3 metadata header (x-amz-meta-*)
147    pub fn with_metadata<K: AsRef<str>, V: AsRef<str>>(
148        mut self,
149        key: K,
150        value: V,
151    ) -> Result<Self, S3Error> {
152        let header_name = format!("x-amz-meta-{}", key.as_ref());
153        let name = header_name.parse::<http::HeaderName>()?;
154        let value = value
155            .as_ref()
156            .parse::<HeaderValue>()
157            .map_err(S3Error::InvalidHeaderValue)?;
158        self.custom_headers.insert(name, value);
159        Ok(self)
160    }
161
162    /// Add x-amz-storage-class header
163    pub fn with_storage_class<S: AsRef<str>>(mut self, storage_class: S) -> Result<Self, S3Error> {
164        let header_value = storage_class
165            .as_ref()
166            .parse::<HeaderValue>()
167            .map_err(S3Error::InvalidHeaderValue)?;
168        self.custom_headers.insert(
169            http::HeaderName::from_static("x-amz-storage-class"),
170            header_value,
171        );
172        Ok(self)
173    }
174
175    /// Add x-amz-server-side-encryption header
176    pub fn with_server_side_encryption<S: AsRef<str>>(
177        mut self,
178        encryption: S,
179    ) -> Result<Self, S3Error> {
180        let header_value = encryption
181            .as_ref()
182            .parse::<HeaderValue>()
183            .map_err(S3Error::InvalidHeaderValue)?;
184        self.custom_headers.insert(
185            http::HeaderName::from_static("x-amz-server-side-encryption"),
186            header_value,
187        );
188        Ok(self)
189    }
190
191    /// Execute the PUT request
192    #[maybe_async::maybe_async]
193    pub async fn execute(self) -> Result<ResponseData, S3Error> {
194        let command = Command::PutObject {
195            content: &self.content,
196            content_type: &self.content_type,
197            custom_headers: if self.custom_headers.is_empty() {
198                None
199            } else {
200                Some(self.custom_headers)
201            },
202            multipart: None,
203        };
204
205        let request = RequestImpl::new(self.bucket, &self.path, command).await?;
206        request.response_data(true).await
207    }
208}
209
210/// Builder for streaming PUT operations.
211///
212/// If an error is returned after multipart initiation, the upload is aborted on
213/// a best-effort basis. Cancelling the future cannot run an asynchronous abort,
214/// and in-flight parts or a lost completion response can leave remote state
215/// uncertain.
216///
217/// On multipart uploads, headers supplied to this builder are routed by request:
218/// object metadata and other custom headers are sent at initiation, SSE-C headers
219/// are repeated for initiation, parts, and completion, requester-pays and expected
220/// bucket-owner headers are sent on each request, and `If-Match`/`If-None-Match`
221/// are sent at completion. `Content-MD5`, checksum headers, `Content-Length`,
222/// `Transfer-Encoding`, and `x-amz-content-sha256` are rejected because one
223/// builder value cannot describe the streamed part bodies. `with_content_type`
224/// controls the generated content type for initiation and parts. Bucket-level
225/// extra headers keep their existing raw behavior and must be valid for every
226/// request made by the multipart upload.
227#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
228#[derive(Debug, Clone)]
229pub struct PutObjectStreamRequest<'a> {
230    bucket: &'a Bucket,
231    path: String,
232    content_type: String,
233    custom_headers: HeaderMap,
234    max_concurrent_chunks: Option<std::num::NonZeroUsize>,
235}
236
237#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
238impl<'a> PutObjectStreamRequest<'a> {
239    /// Create a new streaming PUT request builder
240    pub(crate) fn new<S: AsRef<str>>(bucket: &'a Bucket, path: S) -> Self {
241        Self {
242            bucket,
243            path: path.as_ref().to_string(),
244            content_type: "application/octet-stream".to_string(),
245            custom_headers: HeaderMap::new(),
246            max_concurrent_chunks: None,
247        }
248    }
249
250    /// Set the Content-Type header
251    pub fn with_content_type<S: AsRef<str>>(mut self, content_type: S) -> Self {
252        self.content_type = content_type.as_ref().to_string();
253        self
254    }
255
256    /// Set the Cache-Control header
257    pub fn with_cache_control<S: AsRef<str>>(mut self, cache_control: S) -> Result<Self, S3Error> {
258        let value = cache_control
259            .as_ref()
260            .parse::<HeaderValue>()
261            .map_err(S3Error::InvalidHeaderValue)?;
262        self.custom_headers
263            .insert(http::header::CACHE_CONTROL, value);
264        Ok(self)
265    }
266
267    /// Set the Content-Encoding header
268    pub fn with_content_encoding<S: AsRef<str>>(mut self, encoding: S) -> Result<Self, S3Error> {
269        let value = encoding
270            .as_ref()
271            .parse::<HeaderValue>()
272            .map_err(S3Error::InvalidHeaderValue)?;
273        self.custom_headers
274            .insert(http::header::CONTENT_ENCODING, value);
275        Ok(self)
276    }
277
278    /// Add a custom header
279    pub fn with_header<K, V>(mut self, key: K, value: V) -> Result<Self, S3Error>
280    where
281        K: Into<http::HeaderName>,
282        V: AsRef<str>,
283    {
284        let header_value = value
285            .as_ref()
286            .parse::<HeaderValue>()
287            .map_err(S3Error::InvalidHeaderValue)?;
288        self.custom_headers.insert(key.into(), header_value);
289        Ok(self)
290    }
291
292    /// Add multiple custom headers (already validated HeaderMap)
293    pub fn with_headers(mut self, headers: HeaderMap) -> Self {
294        self.custom_headers.extend(headers);
295        self
296    }
297
298    /// Add S3 metadata header (x-amz-meta-*)
299    pub fn with_metadata<K: AsRef<str>, V: AsRef<str>>(
300        mut self,
301        key: K,
302        value: V,
303    ) -> Result<Self, S3Error> {
304        let header_name = format!("x-amz-meta-{}", key.as_ref());
305        let name = header_name.parse::<http::HeaderName>()?;
306        let value = value
307            .as_ref()
308            .parse::<HeaderValue>()
309            .map_err(S3Error::InvalidHeaderValue)?;
310        self.custom_headers.insert(name, value);
311        Ok(self)
312    }
313
314    /// Set the maximum number of multipart parts this upload may have in flight.
315    ///
316    /// A value of `1` uploads parts sequentially through the same streaming
317    /// path. Values greater than `1` set the exact per-upload concurrency cap,
318    /// including values above the automatic default's maximum of 10. A value
319    /// of `0` restores the automatic setting, currently 2–10 parts based on
320    /// available memory, or 3 if memory detection fails. This is a per-upload
321    /// concurrency limit, not a process-wide memory or RSS limit. It applies
322    /// only to multipart uploads; smaller objects use a regular PUT.
323    pub fn with_max_concurrent_chunks(mut self, max: usize) -> Self {
324        self.max_concurrent_chunks = std::num::NonZeroUsize::new(max);
325        self
326    }
327
328    /// Execute the streaming PUT request
329    #[cfg(feature = "with-tokio")]
330    pub async fn execute_stream<R: AsyncRead + Unpin + ?Sized>(
331        self,
332        reader: &mut R,
333    ) -> Result<crate::utils::PutStreamResponse, S3Error> {
334        // AsyncReadExt trait is not used here
335
336        self.bucket
337            ._put_object_stream_with_content_type_and_headers(
338                reader,
339                &self.path,
340                &self.content_type,
341                if self.custom_headers.is_empty() {
342                    None
343                } else {
344                    Some(self.custom_headers)
345                },
346                self.max_concurrent_chunks,
347            )
348            .await
349    }
350
351    #[cfg(feature = "with-async-std")]
352    pub async fn execute_stream<R: AsyncRead + Unpin + ?Sized>(
353        self,
354        reader: &mut R,
355    ) -> Result<crate::utils::PutStreamResponse, S3Error> {
356        self.bucket
357            ._put_object_stream_with_content_type_and_headers(
358                reader,
359                &self.path,
360                &self.content_type,
361                if self.custom_headers.is_empty() {
362                    None
363                } else {
364                    Some(self.custom_headers)
365                },
366                self.max_concurrent_chunks,
367            )
368            .await
369    }
370}
371
372#[cfg(test)]
373mod tests {
374    use super::*;
375    use crate::Region;
376    use crate::creds::Credentials;
377
378    #[test]
379    fn test_builder_chain() {
380        let bucket =
381            Bucket::new("test", Region::UsEast1, Credentials::anonymous().unwrap()).unwrap();
382
383        let content = b"test content";
384        let request = PutObjectRequest::new(&bucket, "/test.txt", content)
385            .with_content_type("text/plain")
386            .with_cache_control("max-age=3600")
387            .unwrap()
388            .with_content_encoding("gzip")
389            .unwrap()
390            .with_metadata("author", "test-user")
391            .unwrap()
392            .with_header("x-custom", "value")
393            .unwrap()
394            .with_storage_class("STANDARD_IA")
395            .unwrap();
396
397        assert_eq!(request.content_type, "text/plain");
398        assert!(
399            request
400                .custom_headers
401                .contains_key(http::header::CACHE_CONTROL)
402        );
403        assert!(
404            request
405                .custom_headers
406                .contains_key(http::header::CONTENT_ENCODING)
407        );
408        assert!(request.custom_headers.contains_key("x-amz-meta-author"));
409        assert!(request.custom_headers.contains_key("x-custom"));
410        assert!(request.custom_headers.contains_key("x-amz-storage-class"));
411    }
412
413    #[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
414    #[test]
415    fn stream_builder_max_concurrency_defaults_and_resets_to_automatic() {
416        let bucket =
417            Bucket::new("test", Region::UsEast1, Credentials::anonymous().unwrap()).unwrap();
418
419        assert_eq!(
420            bucket
421                .put_object_stream_builder("/stream")
422                .max_concurrent_chunks,
423            None
424        );
425        assert_eq!(
426            bucket
427                .put_object_stream_builder("/stream")
428                .with_max_concurrent_chunks(1)
429                .max_concurrent_chunks
430                .unwrap()
431                .get(),
432            1
433        );
434        assert_eq!(
435            bucket
436                .put_object_stream_builder("/stream")
437                .with_max_concurrent_chunks(2)
438                .max_concurrent_chunks
439                .unwrap()
440                .get(),
441            2
442        );
443        assert_eq!(
444            bucket
445                .put_object_stream_builder("/stream")
446                .with_max_concurrent_chunks(32)
447                .max_concurrent_chunks
448                .unwrap()
449                .get(),
450            32
451        );
452        assert_eq!(
453            bucket
454                .put_object_stream_builder("/stream")
455                .with_max_concurrent_chunks(32)
456                .with_max_concurrent_chunks(0)
457                .max_concurrent_chunks,
458            None
459        );
460    }
461
462    #[test]
463    fn test_metadata_headers() {
464        let bucket =
465            Bucket::new("test", Region::UsEast1, Credentials::anonymous().unwrap()).unwrap();
466
467        let request = PutObjectRequest::new(&bucket, "/test.txt", b"test")
468            .with_metadata("key1", "value1")
469            .unwrap()
470            .with_metadata("key2", "value2")
471            .unwrap();
472
473        assert!(request.custom_headers.contains_key("x-amz-meta-key1"));
474        assert!(request.custom_headers.contains_key("x-amz-meta-key2"));
475    }
476}
s3/put_object_request.rs

s3/
put_object_request.rs
