pub async fn download_source(
        &self,
        file_source: GcsRemoteDif,
        destination: PathBuf,
    ) -> Result<DownloadStatus, DownloadError> {
        let key = file_source.key();
        let bucket = file_source.source.bucket.clone();
        log::debug!("Fetching from GCS: {} (from {})", &key, bucket);
        let token = self.get_token(&file_source.source.source_key).await?;
        log::debug!("Got valid GCS token");

        let mut url = Url::parse("https://www.googleapis.com/download/storage/v1/b?alt=media")
            .map_err(|_| GcsError::InvalidUrl)?;
        // Append path segments manually for proper encoding
        url.path_segments_mut()
            .map_err(|_| GcsError::InvalidUrl)?
            .extend(&[&bucket, "o", &key]);

        let source = RemoteDif::from(file_source);
        let request = self
            .client
            .get(url.clone())
            .header("authorization", format!("Bearer {}", token.access_token))
            .send();
        let request = tokio::time::timeout(self.connect_timeout, request);
        let request = super::measure_download_time(source.source_metric_key(), request);

        match request.await {
            Ok(Ok(response)) => {
                if response.status().is_success() {
                    log::trace!("Success hitting GCS {} (from {})", &key, bucket);

                    let content_length = response
                        .headers()
                        .get(header::CONTENT_LENGTH)
                        .and_then(|hv| hv.to_str().ok())
                        .and_then(|s| s.parse::<u32>().ok());

                    let timeout =
                        content_length.map(|cl| content_length_timeout(cl, self.streaming_timeout));
                    let stream = response.bytes_stream().map_err(DownloadError::Reqwest);

                    super::download_stream(source, stream, destination, timeout).await
                } else if matches!(
                    response.status(),
                    StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED
                ) {
                    log::debug!(
                        "Insufficient permissions to download from GCS {} (from {})",
                        &key,
                        &bucket,
                    );
                    Err(DownloadError::Permissions)
                // If it's a client error, chances are either it's a 404 or it's permission-related.
                } else if response.status().is_client_error() {
                    log::debug!(
                        "Unexpected client error status code from GCS {} (from {}): {}",
                        &key,
                        &bucket,
                        response.status()
                    );
                    Ok(DownloadStatus::NotFound)
                } else {
                    log::debug!(
                        "Unexpected status code from GCS {} (from {}): {}",
                        &key,
                        &bucket,
                        response.status()
                    );
                    Err(DownloadError::Rejected(response.status()))
                }
            }
            Ok(Err(e)) => {
                log::debug!(
                    "Skipping response from GCS {} (from {}): {}",
                    &key,
                    &bucket,
                    &e
                );
                Err(DownloadError::Reqwest(e))
            }
            Err(_) => {
                // Timeout
                Err(DownloadError::Canceled)
            }
        }
    }
/// Downloads a source hosted on GCS.
///
/// # Directly thrown errors
/// - [`GcsError::InvalidUrl`]
/// - [`DownloadError::Reqwest`]
/// - [`DownloadError::Rejected`]
/// - [`DownloadError::Canceled`]
Downloads a source hosted on GCS.
Downloads a source hosted on GCS .
pub async fn download_source(
        &self,
        file_source: GcsRemoteDif,
        destination: PathBuf,
    ) -> Result<DownloadStatus, DownloadError> {
        let key = file_source.key();
        let bucket = file_source.source.bucket.clone();
        log::debug!("Fetching from GCS: {} (from {})", &key, bucket);
        let token = self.get_token(&file_source.source.source_key).await?;
        log::debug!("Got valid GCS token");
        let mut url = Url::parse("https://www.googleapis.com/download/storage/v1/b?alt=media")
            .map_err(|_| GcsError::InvalidUrl)?;
        // Append path segments manually for proper encoding
        url.path_segments_mut()
            .map_err(|_| GcsError::InvalidUrl)?
            .extend(&[&bucket, "o", &key]);
        let source = RemoteDif::from(file_source);
        let request = self
            .client
            .get(url.clone())
            .header("authorization", format!("Bearer {}", token.access_token))
            .send();
        let request = tokio::time::timeout(self.connect_timeout, request);
        let request = super::measure_download_time(source.source_metric_key(), request);
        match request.await {
            Ok(Ok(response)) => {
                if response.status().is_success() {
                    log::trace!("Success hitting GCS {} (from {})", &key, bucket);
                    let content_length = response
                        .headers()
                        .get(header::CONTENT_LENGTH)
                        .and_then(|hv| hv.to_str().ok())
                        .and_then(|s| s.parse::<u32>().ok());
                    let timeout =
                        content_length.map(|cl| content_length_timeout(cl, self.streaming_timeout));
                    let stream = response.bytes_stream().map_err(DownloadError::Reqwest);
                    super::download_stream(source, stream, destination, timeout).await
                } else if matches!(
                    response.status(),
                    StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED
                ) {
                    log::debug!(
                        "Insufficient permissions to download from GCS {} (from {})",
                        &key,
                        &bucket,
                    );
                    Err(DownloadError::Permissions)
                // If it's a client error, chances are either it's a 404 or it's permission-related.
                } else if response.status().is_client_error() {
                    log::debug!(
                        "Unexpected client error status code from GCS {} (from {}): {}",
                        &key,
                        &bucket,
                        response.status()
                    );
                    Ok(DownloadStatus::NotFound)
                } else {
                    log::debug!(
                        "Unexpected status code from GCS {} (from {}): {}",
                        &key,
                        &bucket,
                        response.status()
                    );
                    Err(DownloadError::Rejected(response.status()))
                }
            }
            Ok(Err(e)) => {
                log::debug!(
                    "Skipping response from GCS {} (from {}): {}",
                    &key,
                    &bucket,
                    &e
                );
                Err(DownloadError::Reqwest(e))
            }
            Err(_) => {
                // Timeout
                Err(DownloadError::Canceled)
            }
        }
    }
pub async fn download_source ( & self , file_source : GcsRemoteDif , destination : PathBuf , ) -> Result < DownloadStatus , DownloadError > { let key = file_source . key ( ) ; let bucket = file_source . source . bucket . clone ( ) ; log :: debug ! ( " " key bucket ) ; let token = self . get_token ( & file_source . source . source_key ) . await ? ; log :: debug ! ( " " ) ; let mut url = Url :: parse ( " " ) . map_err ( | _ | GcsError :: InvalidUrl ) ? ; // Append path segments manually for proper encoding url . path_segments_mut ( ) . map_err ( | _ | GcsError :: InvalidUrl ) ? . extend ( & [ & bucket , " " , & key ] ) ; let source = RemoteDif :: from ( file_source ) ; let request = self . client . get ( url . clone ( ) ) . header ( " " , format ! ( " " token access_token ) ) . send ( ) ; let request = tokio :: time :: timeout ( self . connect_timeout , request ) ; let request = super :: measure_download_time ( source . source_metric_key ( ) , request ) ; match request . await { Ok ( Ok ( response ) ) => { if response . status ( ) . is_success ( ) { log :: trace ! ( " " key bucket ) ; let content_length = response . headers ( ) . get ( header :: CONTENT_LENGTH ) . and_then ( | hv | hv . to_str ( ) . ok ( ) ) . and_then ( | s | s . parse :: < u32 > ( ) . ok ( ) ) ; let timeout = content_length . map ( | cl | content_length_timeout ( cl , self . streaming_timeout ) ) ; let stream = response . bytes_stream ( ) . map_err ( DownloadError :: Reqwest ) ; super :: download_stream ( source , stream , destination , timeout ) . await } else if matches ! ( response status ( ) StatusCode FORBIDDEN StatusCode UNAUTHORIZED ) { log :: debug ! ( " " key bucket ) ; Err ( DownloadError :: Permissions ) // If it's a client error, chances are either it's a 404 or it's permission-related. } else if response . status ( ) . is_client_error ( ) { log :: debug ! ( " " key bucket response status ( ) ) ; Ok ( DownloadStatus :: NotFound ) } else { log :: debug ! ( " " key bucket response status ( ) ) ; Err ( DownloadError :: Rejected ( response . status ( ) ) ) } } Ok ( Err ( e ) ) => { log :: debug ! ( " " key bucket e ) ; Err ( DownloadError :: Reqwest ( e ) ) } Err ( _ ) => { // Timeout Err ( DownloadError :: Canceled ) } } }
Downloads a source hosted on GCS.
Downloads a source hosted on GCS .
&self
file_source
GcsRemoteDif
destination
PathBuf
&self
file_source
GcsRemoteDif
destination
PathBuf
directly thrown errors
None