diff --git a/Cargo.lock b/Cargo.lock index fe9695bfd9c..2bb4fedda04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -28,6 +28,20 @@ dependencies = [ "version_check", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "const-random", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -46,6 +60,15 @@ dependencies = [ "cc", ] +[[package]] +name = "android_system_properties" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" +dependencies = [ + "libc", +] + [[package]] name = "anes" version = "0.1.6" @@ -129,6 +152,180 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "arrow" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61d285d16bce7d0be61912f7928342b673067b6b7d7ef6cc179258ba7de1fecf" +dependencies = [ + "arrow-arith", + "arrow-array", + "arrow-buffer", + "arrow-cast", + "arrow-data", + "arrow-ipc", + "arrow-ord", + "arrow-row", + "arrow-schema", + "arrow-select", + "arrow-string", +] + +[[package]] +name = "arrow-arith" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "757ef1836251e88222542a7da2623bc1c9cb9e20afefa6db2c41e79991cd91d4" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "num-traits", +] + +[[package]] +name = "arrow-array" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc9a4a4b2b5ecd0e04df03471661cb61f28bed3c7fd50994715129b01b2edb97" +dependencies = [ + "ahash 0.8.12", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "chrono", + "half", + "hashbrown 0.17.1", + "num-complex", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-buffer" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c12b576ef18c1deb80925a248b25ad84f419198d791b8e293fc6aaa60441fe90" +dependencies = [ + "bytes", + "half", + "num-bigint", + "num-traits", +] + +[[package]] +name = "arrow-cast" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68338a9096a5dc9bc11927c58c43a8526d96bf6abd2012ef6c0c9f505991cc79" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-ord", + "arrow-schema", + "arrow-select", + "atoi", + "base64 0.23.1", + "chrono", + "half", + "lexical-core", + "num-traits", + "ryu", +] + +[[package]] +name = "arrow-data" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "723fe4aeed7604e00b9883a465af4ff0a0e6c44c03e41a68c3d1cbc403e0e44d" +dependencies = [ + "arrow-buffer", + "arrow-schema", + "half", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-ipc" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "149437b14371f5b9ec60f5ddc751483ae99d7a7072653c0075e5e469156eea7b" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", + "flatbuffers", +] + +[[package]] +name = "arrow-ord" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c08dff0686cf23ca4f562803f191ccbeb726dbae6309cd4b4aaf65e0f2c979" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", +] + +[[package]] +name = "arrow-row" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbec439386df71ad570e6758a946111322b9e9dc8db83b5527321f0b4c9119c2" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "half", +] + +[[package]] +name = "arrow-schema" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6fed2ca0d1eade57e811cbe73b98ad50cc08a1183e13b2d2aa43a7df593f40e" + +[[package]] +name = "arrow-select" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "466b19cf75130b891dc1b23a84b343c714c62c64c9c62e365c76aa0ff90a53fb" +dependencies = [ + "ahash 0.8.12", + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "num-traits", +] + +[[package]] +name = "arrow-string" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c838a25bb3691e919e0f617616ac51a4ff8517a952e29ca133cf0c22b2ce65b1" +dependencies = [ + "arrow-array", + "arrow-buffer", + "arrow-data", + "arrow-schema", + "arrow-select", + "memchr", + "num-traits", + "regex", + "regex-syntax", +] + [[package]] name = "async-compression" version = "0.4.42" @@ -185,6 +382,15 @@ dependencies = [ "syn 3.0.2", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -453,7 +659,7 @@ version = "0.1.0" dependencies = [ "async-trait", "azure_core 1.2.0-beta.1", - "base64", + "base64 0.22.1", "futures", "serde", "serde_json", @@ -595,7 +801,7 @@ dependencies = [ "azure_core_test 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", "azure_data_cosmos_driver", "azure_identity 1.0.0", - "base64", + "base64 0.22.1", "clap", "futures", "json-canon", @@ -640,7 +846,7 @@ dependencies = [ "azure_data_cosmos_macros 0.2.0", "azure_identity 1.0.0", "backtrace", - "base64", + "base64 0.22.1", "bytes", "crossbeam-epoch", "futures", @@ -815,7 +1021,7 @@ dependencies = [ "azure_messaging_eventhubs", "azure_messaging_eventhubs_checkpointstore_blob", "azure_storage_blob 1.0.0", - "base64", + "base64 0.22.1", "criterion", "fe2o3-amqp", "futures", @@ -977,6 +1183,7 @@ dependencies = [ name = "azure_storage_blob" version = "1.1.0-beta.3" dependencies = [ + "arrow", "async-stream", "async-trait", "azure_core 1.2.0-beta.1", @@ -1032,7 +1239,7 @@ version = "0.2.0" dependencies = [ "azure_core 1.2.0-beta.1", "azure_storage_common", - "base64", + "base64 0.22.1", "hmac", "include-file", "percent-encoding", @@ -1061,6 +1268,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64ct" version = "1.8.3" @@ -1224,7 +1437,9 @@ version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ + "iana-time-zone", "num-traits", + "windows-link", ] [[package]] @@ -1375,6 +1590,26 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "tiny-keccak", +] + [[package]] name = "convert_case" version = "0.6.0" @@ -1745,6 +1980,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flatbuffers" +version = "25.12.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3" +dependencies = [ + "bitflags", + "rustc_version", +] + [[package]] name = "flate2" version = "1.1.9" @@ -1980,6 +2225,7 @@ checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" dependencies = [ "cfg-if", "crunchy", + "num-traits", "zerocopy", ] @@ -1989,7 +2235,7 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" dependencies = [ - "ahash", + "ahash 0.7.8", ] [[package]] @@ -2004,7 +2250,7 @@ version = "7.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f49d1053f4708f0af3cf9fc5bffc7e68a914a3c45becb231c80068c9c3f78bea" dependencies = [ - "base64", + "base64 0.22.1", "byteorder", "crossbeam-channel", "flate2", @@ -2161,7 +2407,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -2178,6 +2424,30 @@ dependencies = [ "tracing", ] +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core 0.62.2", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "icu_collections" version = "2.2.0" @@ -2465,12 +2735,75 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "lexical-core" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d8d125a277f807e55a77304455eb7b1cb52f2b18c143b60e766c120bd64a594" +dependencies = [ + "lexical-parse-float", + "lexical-parse-integer", + "lexical-util", + "lexical-write-float", + "lexical-write-integer", +] + +[[package]] +name = "lexical-parse-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52a9f232fbd6f550bc0137dcb5f99ab674071ac2d690ac69704593cb4abbea56" +dependencies = [ + "lexical-parse-integer", + "lexical-util", +] + +[[package]] +name = "lexical-parse-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a7a039f8fb9c19c996cd7b2fcce303c1b2874fe1aca544edc85c4a5f8489b34" +dependencies = [ + "lexical-util", +] + +[[package]] +name = "lexical-util" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2604dd126bb14f13fb5d1bd6a66155079cb9fa655b37f875b3a742c705dbed17" + +[[package]] +name = "lexical-write-float" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50c438c87c013188d415fbabbb1dceb44249ab81664efbd31b14ae55dabb6361" +dependencies = [ + "lexical-util", + "lexical-write-integer", +] + +[[package]] +name = "lexical-write-integer" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "409851a618475d2d5796377cad353802345cba92c867d9fbcde9cf4eac4e14df" +dependencies = [ + "lexical-util", +] + [[package]] name = "libc" version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libm" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -2602,12 +2935,40 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + +[[package]] +name = "num-complex" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73f88a1307638156682bada9d7604135552957b7818057dcef22705b4d509495" +dependencies = [ + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -2615,6 +2976,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" dependencies = [ "autocfg", + "libm", ] [[package]] @@ -3327,7 +3689,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -4032,6 +4394,15 @@ dependencies = [ "time-core", ] +[[package]] +name = "tiny-keccak" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237" +dependencies = [ + "crunchy", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -4220,7 +4591,7 @@ dependencies = [ "async-stream", "async-trait", "axum 0.7.9", - "base64", + "base64 0.22.1", "bytes", "h2", "http", @@ -4248,7 +4619,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "http", "http-body", @@ -4462,7 +4833,7 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "753a2fe021e407d4fc9ee6f4f0a33403cc306d5c54c4e4ebe1b8cbde0ca052b9" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures", "quick-xml", @@ -4475,7 +4846,7 @@ dependencies = [ name = "typespec" version = "1.2.0-beta.1" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures", "quick-xml", @@ -4492,7 +4863,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0373af0f9d4f580b3a1a9d9639cedaabe015ed262b35bfbe13941bfb14fe1ea6" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "dyn-clone", "futures", @@ -4515,7 +4886,7 @@ name = "typespec_client_core" version = "1.2.0-beta.1" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "dyn-clone", "futures", @@ -4586,7 +4957,7 @@ version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0" dependencies = [ - "base64", + "base64 0.22.1", "der", "flate2", "log", @@ -4604,7 +4975,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c" dependencies = [ - "base64", + "base64 0.22.1", "http", "httparse", "log", diff --git a/sdk/storage/.cspell.json b/sdk/storage/.cspell.json index 291d53ac6a6..d02f90fa9ba 100644 --- a/sdk/storage/.cspell.json +++ b/sdk/storage/.cspell.json @@ -7,12 +7,14 @@ "ADLS", "appendblock", "appendpos", + "blobprefix", "blockid", "blocklist", "blocklisttype", "Btext", "Chttp", "copyid", + "datatypes", "deletedwithversions", "deletetype", "devstoreaccount", diff --git a/sdk/storage/azure_storage_blob/CHANGELOG.md b/sdk/storage/azure_storage_blob/CHANGELOG.md index 54ac707761e..0454ac10dd6 100644 --- a/sdk/storage/azure_storage_blob/CHANGELOG.md +++ b/sdk/storage/azure_storage_blob/CHANGELOG.md @@ -5,10 +5,13 @@ ### Features Added - Updated the default service version to `2026-10-06`. +- Re-added `BlobContainerClient::list_blobs_hierarchical()`. +- Added `end_before` to `BlobContainerClientListBlobsOptions` and `BlobContainerClientListBlobsHierarchicalOptions`. `end_before` is only supported with the Apache Arrow response format. - The service-calculated CRC64 is now surfaced as `content_crc64` on upload responses, alongside `content_md5`, when a content MD5 is provided with the request. This applies to `stage_block`, `stage_block_from_url`, `upload_pages`, `upload_pages_from_url`, `append_block`, `append_block_from_url`, `upload` and `upload_blob_from_url`. ### Breaking Changes +- `BlobContainerClient::list_blobs()` now requests Apache Arrow by default and returns `Result>` instead of `Result>`. Use `BlobContainerClientListBlobsOptions::response_format` to request a different response format. - Added the `AccessTier::Smart` and `ArchiveStatus::RehydratePendingToSmart` enum variants. - Added `access_tier`, `access_tier_changed_on`, `access_tier_inferred`, and `smart_access_tier` to `BlobDownloadProperties` and marked the struct as non-exhaustive. - Added `BlobClient::start_copy_from_url()` and `BlobClient::abort_copy()` for asynchronous blob copy operations. diff --git a/sdk/storage/azure_storage_blob/Cargo.toml b/sdk/storage/azure_storage_blob/Cargo.toml index b2d86742994..6792bf05db3 100644 --- a/sdk/storage/azure_storage_blob/Cargo.toml +++ b/sdk/storage/azure_storage_blob/Cargo.toml @@ -18,6 +18,7 @@ default = ["tokio", "azure_core/default"] tokio = ["dep:tokio", "azure_core/tokio"] [dependencies] +arrow = { workspace = true, features = ["ipc"] } async-stream.workspace = true async-trait.workspace = true azure_core = { path = "../../core/azure_core", version = "1.2.0-beta.1", features = ["xml"] } diff --git a/sdk/storage/azure_storage_blob/assets.json b/sdk/storage/azure_storage_blob/assets.json index f85f9ed1b0b..a72056fc3bc 100644 --- a/sdk/storage/azure_storage_blob/assets.json +++ b/sdk/storage/azure_storage_blob/assets.json @@ -1,6 +1,6 @@ { "AssetsRepo": "Azure/azure-sdk-assets", "AssetsRepoPrefixPath": "rust", - "Tag": "rust/azure_storage_blob_d5db5e0143", + "Tag": "rust/azure_storage_blob_55bea387aa", "TagPrefix": "rust/azure_storage_blob" } diff --git a/sdk/storage/azure_storage_blob/src/arrow_decode.rs b/sdk/storage/azure_storage_blob/src/arrow_decode.rs new file mode 100644 index 00000000000..6192d9ab2d9 --- /dev/null +++ b/sdk/storage/azure_storage_blob/src/arrow_decode.rs @@ -0,0 +1,904 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +use crate::models::{ + BlobHierarchyList, BlobItem, BlobMetadata, BlobPrefix, BlobProperties, BlobTag, BlobTags, + ListBlobsHierarchicalResponse, ListBlobsResponse, ObjectReplicationMetadata, +}; +use arrow::array::{ + Array, BooleanArray, Int32Array, Int64Array, MapArray, RecordBatch, StringArray, + TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray, + TimestampSecondArray, UInt32Array, UInt64Array, +}; +use arrow::datatypes::{DataType, TimeUnit}; +use arrow::error::ArrowError; +use arrow::ipc::reader::StreamReader; +use azure_core::{ + base64, + error::{Error, ErrorKind}, + http::{ + headers::{self, Headers}, + response::ResponseBody, + DeserializeWith, Etag, Format, RawResponse, + }, + time::OffsetDateTime, + Result, +}; +use serde::{de::DeserializeOwned, Deserialize}; + +const NEXT_MARKER_KEY: &str = "NextMarker"; +const ARROW_CONTENT_TYPE: &str = "application/vnd.apache.arrow.stream"; +const XML_CONTENT_TYPE: &str = "application/xml"; + +#[derive(Clone, Copy)] +enum WireFormat { + Arrow, + Xml, +} + +/// Selects the list blobs deserializer from the response `Content-Type`. +#[derive(Debug, Clone)] +pub struct AutoFormat; + +impl Format for AutoFormat { + fn deserialize>(body: S) -> Result { + azure_core::xml::from_xml(body.as_ref()) + } +} + +impl DeserializeWith for ListBlobsResponse { + fn deserialize_with(body: ResponseBody) -> Result { + body.xml() + } + + fn deserialize_from(response: RawResponse) -> Result { + match wire_format(response.headers())? { + WireFormat::Arrow => decode_arrow_list_blobs(response.body()), + WireFormat::Xml => azure_core::xml::from_xml(response.body()), + } + } +} + +impl DeserializeWith for ListBlobsHierarchicalResponse { + fn deserialize_with(body: ResponseBody) -> Result { + body.xml() + } + + fn deserialize_from(response: RawResponse) -> Result { + match wire_format(response.headers())? { + WireFormat::Arrow => decode_arrow_list_blobs_hierarchy(response.body()), + WireFormat::Xml => azure_core::xml::from_xml(response.body()), + } + } +} + +pub(crate) fn decode_next_marker(headers: &Headers, bytes: &[u8]) -> Result> { + match wire_format(headers)? { + WireFormat::Arrow => arrow_next_marker(bytes), + WireFormat::Xml => { + #[derive(Deserialize)] + struct ListBlobsPage { + #[serde(rename = "NextMarker")] + next_marker: Option, + } + + let page: ListBlobsPage = azure_core::xml::from_xml(bytes)?; + Ok(page.next_marker.filter(|marker| !marker.is_empty())) + } + } +} + +fn wire_format(headers: &Headers) -> Result { + let Some(content_type) = headers.get_optional_str(&headers::CONTENT_TYPE) else { + return Ok(WireFormat::Xml); + }; + let media_type = content_type.split(';').next().unwrap_or_default().trim(); + if media_type.eq_ignore_ascii_case(ARROW_CONTENT_TYPE) { + Ok(WireFormat::Arrow) + } else if media_type.eq_ignore_ascii_case(XML_CONTENT_TYPE) { + Ok(WireFormat::Xml) + } else { + Err(Error::with_message( + ErrorKind::DataConversion, + format!("unsupported list blobs Content-Type: {content_type}"), + )) + } +} + +fn arrow_next_marker(bytes: &[u8]) -> Result> { + let reader = StreamReader::try_new(bytes, None).map_err(to_error)?; + Ok(reader + .schema() + .metadata() + .get(NEXT_MARKER_KEY) + .filter(|marker| !marker.is_empty()) + .cloned()) +} + +fn read_arrow_rows(bytes: &[u8], mut per_row: F) -> Result> +where + F: FnMut(&RecordBatch, usize), +{ + let reader = StreamReader::try_new(bytes, None).map_err(to_error)?; + let next_marker = reader + .schema() + .metadata() + .get(NEXT_MARKER_KEY) + .filter(|marker| !marker.is_empty()) + .cloned(); + for batch in reader { + let batch = batch.map_err(to_error)?; + for row in 0..batch.num_rows() { + per_row(&batch, row); + } + } + Ok(next_marker) +} + +fn decode_arrow_list_blobs(bytes: &[u8]) -> Result { + let mut blob_items = Vec::new(); + let next_marker = read_arrow_rows(bytes, |batch, row| { + blob_items.push(row_to_blob_item(batch, row)); + })?; + Ok(ListBlobsResponse { + blob_items, + next_marker, + ..Default::default() + }) +} + +fn decode_arrow_list_blobs_hierarchy(bytes: &[u8]) -> Result { + let mut blob_items = Vec::new(); + let mut blob_prefixes = Vec::new(); + // Virtual-directory rows are marked with ResourceType == "blobprefix" and carry only a Name. + let next_marker = read_arrow_rows(bytes, |batch, row| { + if is_blob_prefix_row(batch, row) { + blob_prefixes.push(BlobPrefix { + name: string_at(batch, "Name", row), + }); + } else { + blob_items.push(row_to_blob_item(batch, row)); + } + })?; + Ok(ListBlobsHierarchicalResponse { + hierarchical_list: BlobHierarchyList { + blob_items, + blob_prefixes: (!blob_prefixes.is_empty()).then_some(blob_prefixes), + }, + next_marker, + ..Default::default() + }) +} + +fn is_blob_prefix_row(batch: &RecordBatch, row: usize) -> bool { + string_at(batch, "ResourceType", row) + .is_some_and(|value| value.eq_ignore_ascii_case("blobprefix")) +} + +fn row_to_blob_item(batch: &RecordBatch, row: usize) -> BlobItem { + let properties = BlobProperties { + access_tier: string_at(batch, "AccessTier", row).and_then(|value| value.parse().ok()), + access_tier_change_time: timestamp_at(batch, "AccessTierChangeTime", row), + access_tier_inferred: bool_at(batch, "AccessTierInferred", row), + archive_status: string_at(batch, "ArchiveStatus", row).and_then(|value| value.parse().ok()), + blob_sequence_number: i64_at(batch, "x-ms-blob-sequence-number", row), + blob_type: string_at(batch, "BlobType", row).and_then(|value| value.parse().ok()), + cache_control: string_at(batch, "Cache-Control", row), + content_disposition: string_at(batch, "Content-Disposition", row), + content_encoding: string_at(batch, "Content-Encoding", row), + content_language: string_at(batch, "Content-Language", row), + content_length: u64_at(batch, "Content-Length", row), + content_md5: string_at(batch, "Content-MD5", row) + .and_then(|value| base64::decode(value).ok()), + content_type: string_at(batch, "Content-Type", row), + copy_completion_time: timestamp_at(batch, "CopyCompletionTime", row), + copy_id: string_at(batch, "CopyId", row), + copy_progress: string_at(batch, "CopyProgress", row), + copy_source: string_at(batch, "CopySource", row), + copy_status: string_at(batch, "CopyStatus", row).and_then(|value| value.parse().ok()), + copy_status_description: string_at(batch, "CopyStatusDescription", row), + creation_time: timestamp_at(batch, "Creation-Time", row), + deleted_time: timestamp_at(batch, "DeletedTime", row), + destination_snapshot: string_at(batch, "CopyDestinationSnapshot", row), + encryption_key_sha256: string_at(batch, "CustomerProvidedKeySha256", row), + encryption_scope: string_at(batch, "EncryptionScope", row), + etag: string_at(batch, "Etag", row).map(Etag::from), + expires_on: timestamp_at(batch, "Expiry-Time", row), + immutability_policy_expires_on: timestamp_at(batch, "ImmutabilityPolicyUntilDate", row), + immutability_policy_mode: string_at(batch, "ImmutabilityPolicyMode", row) + .and_then(|value| value.parse().ok()), + incremental_copy: bool_at(batch, "IncrementalCopy", row), + is_sealed: bool_at(batch, "Sealed", row), + last_accessed_on: timestamp_at(batch, "LastAccessTime", row), + last_modified: timestamp_at(batch, "Last-Modified", row), + lease_duration: string_at(batch, "LeaseDuration", row).and_then(|value| value.parse().ok()), + lease_state: string_at(batch, "LeaseState", row).and_then(|value| value.parse().ok()), + lease_status: string_at(batch, "LeaseStatus", row).and_then(|value| value.parse().ok()), + legal_hold: bool_at(batch, "LegalHold", row), + rehydrate_priority: string_at(batch, "RehydratePriority", row) + .and_then(|value| value.parse().ok()), + remaining_retention_days: i32_at(batch, "RemainingRetentionDays", row), + server_encrypted: bool_at(batch, "ServerEncrypted", row), + smart_access_tier: string_at(batch, "SmartAccessTier", row) + .and_then(|value| value.parse().ok()), + tag_count: i32_at(batch, "TagCount", row), + ..Default::default() + }; + + BlobItem { + blob_tags: blob_tags_at(batch, "Tags", row), + deleted: bool_at(batch, "Deleted", row), + has_versions_only: bool_at(batch, "HasVersionsOnly", row), + is_current_version: bool_at(batch, "IsCurrentVersion", row), + metadata: blob_metadata_at(batch, "Metadata", row), + name: string_at(batch, "Name", row), + object_replication_metadata: object_replication_metadata_at(batch, "OrMetadata", row), + properties: Some(properties), + snapshot: string_at(batch, "Snapshot", row), + version_id: string_at(batch, "VersionId", row), + ..Default::default() + } +} + +fn map_entries_at(batch: &RecordBatch, name: &str, row: usize) -> Option> { + let array = column(batch, name)?; + if array.is_null(row) { + return None; + } + let map = array.as_any().downcast_ref::()?; + let offsets = map.value_offsets(); + let start = offsets[row] as usize; + let end = offsets[row + 1] as usize; + let keys = map.keys().as_any().downcast_ref::()?; + let values = map.values().as_any().downcast_ref::()?; + let mut entries = Vec::with_capacity(end - start); + for index in start..end { + if keys.is_null(index) { + continue; + } + let value = (!values.is_null(index)) + .then(|| values.value(index).to_string()) + .unwrap_or_default(); + entries.push((keys.value(index).to_string(), value)); + } + (!entries.is_empty()).then_some(entries) +} + +fn blob_tags_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let blob_tag_set = map_entries_at(batch, name, row)? + .into_iter() + .map(|(key, value)| BlobTag { + key: Some(key), + value: Some(value), + }) + .collect(); + Some(BlobTags { + blob_tag_set: Some(blob_tag_set), + }) +} + +fn blob_metadata_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + Some(BlobMetadata { + values: Some(map_entries_at(batch, name, row)?.into_iter().collect()), + encrypted: None, + }) +} + +fn object_replication_metadata_at( + batch: &RecordBatch, + name: &str, + row: usize, +) -> Option { + Some(ObjectReplicationMetadata { + additional_properties: Some(map_entries_at(batch, name, row)?.into_iter().collect()), + }) +} + +fn to_error(error: ArrowError) -> Error { + Error::new(ErrorKind::DataConversion, error) +} + +fn column<'a>(batch: &'a RecordBatch, name: &str) -> Option<&'a dyn Array> { + let index = batch.schema().index_of(name).ok()?; + Some(batch.column(index).as_ref()) +} + +fn string_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)? + .as_any() + .downcast_ref::()?; + (!array.is_null(row)).then(|| array.value(row).to_string()) +} + +fn bool_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)? + .as_any() + .downcast_ref::()?; + (!array.is_null(row)).then(|| array.value(row)) +} + +fn u64_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)?; + if array.is_null(row) { + return None; + } + if let Some(array) = array.as_any().downcast_ref::() { + Some(array.value(row)) + } else if let Some(array) = array.as_any().downcast_ref::() { + u64::try_from(array.value(row)).ok() + } else { + None + } +} + +fn i64_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)?; + if array.is_null(row) { + return None; + } + if let Some(array) = array.as_any().downcast_ref::() { + Some(array.value(row)) + } else if let Some(array) = array.as_any().downcast_ref::() { + i64::try_from(array.value(row)).ok() + } else if let Some(array) = array.as_any().downcast_ref::() { + Some(i64::from(array.value(row))) + } else if let Some(array) = array.as_any().downcast_ref::() { + Some(i64::from(array.value(row))) + } else { + None + } +} + +fn i32_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)?; + if array.is_null(row) { + return None; + } + if let Some(array) = array.as_any().downcast_ref::() { + Some(array.value(row)) + } else if let Some(array) = array.as_any().downcast_ref::() { + i32::try_from(array.value(row)).ok() + } else if let Some(array) = array.as_any().downcast_ref::() { + i32::try_from(array.value(row)).ok() + } else if let Some(array) = array.as_any().downcast_ref::() { + i32::try_from(array.value(row)).ok() + } else { + None + } +} + +fn timestamp_at(batch: &RecordBatch, name: &str, row: usize) -> Option { + let array = column(batch, name)?; + if array.is_null(row) { + return None; + } + let nanos = match array.data_type() { + DataType::Timestamp(TimeUnit::Second, _) => { + i128::from( + array + .as_any() + .downcast_ref::()? + .value(row), + ) * 1_000_000_000 + } + DataType::Timestamp(TimeUnit::Millisecond, _) => { + i128::from( + array + .as_any() + .downcast_ref::()? + .value(row), + ) * 1_000_000 + } + DataType::Timestamp(TimeUnit::Microsecond, _) => { + i128::from( + array + .as_any() + .downcast_ref::()? + .value(row), + ) * 1_000 + } + DataType::Timestamp(TimeUnit::Nanosecond, _) => i128::from( + array + .as_any() + .downcast_ref::()? + .value(row), + ), + _ => return None, + }; + OffsetDateTime::from_unix_timestamp_nanos(nanos).ok() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::{ + AccessTier, ArchiveStatus, BlobType, CopyStatus, ImmutabilityPolicyMode, LeaseDuration, + LeaseState, LeaseStatus, RehydratePriority, + }; + use arrow::array::builder::{MapBuilder, StringBuilder}; + use arrow::array::ArrayRef; + use arrow::datatypes::{Field, Schema}; + use arrow::ipc::writer::StreamWriter; + use std::collections::HashMap; + use std::sync::Arc; + + fn s(value: &str) -> ArrayRef { + Arc::new(StringArray::from(vec![Some(value.to_string())])) + } + + fn b(value: bool) -> ArrayRef { + Arc::new(BooleanArray::from(vec![Some(value)])) + } + + fn u64c(value: u64) -> ArrayRef { + Arc::new(UInt64Array::from(vec![Some(value)])) + } + + fn ts(millis: i64) -> ArrayRef { + Arc::new(TimestampMillisecondArray::from(vec![Some(millis)])) + } + + fn expected_ts(millis: i64) -> OffsetDateTime { + OffsetDateTime::from_unix_timestamp_nanos(millis as i128 * 1_000_000).unwrap() + } + + /// Builds a single-row [`RecordBatch`] from `(column_name, array)` pairs, + /// deriving each nullable [`Field`] from the array's own data type. + fn batch(columns: Vec<(&str, ArrayRef)>) -> RecordBatch { + let fields: Vec = columns + .iter() + .map(|(name, array)| Field::new(*name, array.data_type().clone(), true)) + .collect(); + let arrays: Vec = columns.into_iter().map(|(_, array)| array).collect(); + RecordBatch::try_new(Arc::new(Schema::new(fields)), arrays).unwrap() + } + + #[test] + fn populated_row_maps_every_field() { + let md5 = base64::encode([1u8, 2, 3, 4]); + let batch = batch(vec![ + // BlobItem-level columns. + ("Name", s("blob.txt")), + ("Deleted", b(false)), + ("HasVersionsOnly", b(true)), + ("IsCurrentVersion", b(true)), + ("Snapshot", s("2019-01-01T00:00:00.0000000Z")), + ("VersionId", s("2021-01-01T00:00:00.0000000Z")), + // BlobProperties columns. + ("AccessTier", s("Hot")), + ("AccessTierChangeTime", ts(3_000_000)), + ("AccessTierInferred", b(true)), + ("ArchiveStatus", s("rehydrate-pending-to-hot")), + ("x-ms-blob-sequence-number", u64c(42)), + ("BlobType", s("BlockBlob")), + ("Cache-Control", s("no-cache")), + ("Content-Disposition", s("inline")), + ("Content-Encoding", s("gzip")), + ("Content-Language", s("en-US")), + ("Content-Length", u64c(12345)), + ("Content-MD5", s(&md5)), + ("Content-Type", s("text/plain")), + ("CopyCompletionTime", ts(4_000_000)), + ("CopyId", s("copy-id")), + ("CopyProgress", s("1024/1024")), + ("CopySource", s("https://example.com/source")), + ("CopyStatus", s("success")), + ("CopyStatusDescription", s("done")), + ("Creation-Time", ts(1_000_000)), + ("DeletedTime", ts(5_000_000)), + ("CopyDestinationSnapshot", s("2020-01-01T00:00:00.0000000Z")), + ("CustomerProvidedKeySha256", s("cpk-sha")), + ("EncryptionScope", s("scope-1")), + ("Etag", s("0xETAG")), + ("Expiry-Time", ts(8_000_000)), + ("ImmutabilityPolicyUntilDate", ts(6_000_000)), + ("ImmutabilityPolicyMode", s("unlocked")), + ("IncrementalCopy", b(false)), + ("Sealed", b(true)), + ("LastAccessTime", ts(7_000_000)), + ("Last-Modified", ts(2_000_000)), + ("LeaseDuration", s("infinite")), + ("LeaseState", s("available")), + ("LeaseStatus", s("unlocked")), + ("LegalHold", b(true)), + ("RehydratePriority", s("High")), + ("RemainingRetentionDays", u64c(7)), + ("ServerEncrypted", b(true)), + ("TagCount", u64c(3)), + ]); + + let item = row_to_blob_item(&batch, 0); + + // BlobItem-level fields. + assert_eq!(Some("blob.txt".to_string()), item.name); + assert_eq!(Some(false), item.deleted); + assert_eq!(Some(true), item.has_versions_only); + assert_eq!(Some(true), item.is_current_version); + assert_eq!( + Some("2019-01-01T00:00:00.0000000Z".to_string()), + item.snapshot + ); + assert_eq!( + Some("2021-01-01T00:00:00.0000000Z".to_string()), + item.version_id + ); + + let props = item.properties.expect("properties should be set"); + assert_eq!(Some(AccessTier::Hot), props.access_tier); + assert_eq!(Some(expected_ts(3_000_000)), props.access_tier_change_time); + assert_eq!(Some(true), props.access_tier_inferred); + assert_eq!( + Some(ArchiveStatus::RehydratePendingToHot), + props.archive_status + ); + assert_eq!(Some(42), props.blob_sequence_number); + assert_eq!(Some(BlobType::BlockBlob), props.blob_type); + assert_eq!(Some("no-cache".to_string()), props.cache_control); + assert_eq!(Some("inline".to_string()), props.content_disposition); + assert_eq!(Some("gzip".to_string()), props.content_encoding); + assert_eq!(Some("en-US".to_string()), props.content_language); + assert_eq!(Some(12345), props.content_length); + assert_eq!(Some(vec![1u8, 2, 3, 4]), props.content_md5); + assert_eq!(Some("text/plain".to_string()), props.content_type); + assert_eq!(Some(expected_ts(4_000_000)), props.copy_completion_time); + assert_eq!(Some("copy-id".to_string()), props.copy_id); + assert_eq!(Some("1024/1024".to_string()), props.copy_progress); + assert_eq!( + Some("https://example.com/source".to_string()), + props.copy_source + ); + assert_eq!(Some(CopyStatus::Success), props.copy_status); + assert_eq!(Some("done".to_string()), props.copy_status_description); + assert_eq!(Some(expected_ts(1_000_000)), props.creation_time); + assert_eq!(Some(expected_ts(5_000_000)), props.deleted_time); + assert_eq!( + Some("2020-01-01T00:00:00.0000000Z".to_string()), + props.destination_snapshot + ); + assert_eq!(Some("cpk-sha".to_string()), props.encryption_key_sha256); + assert_eq!(Some("scope-1".to_string()), props.encryption_scope); + assert_eq!(Some(Etag::from("0xETAG")), props.etag); + assert_eq!(Some(expected_ts(8_000_000)), props.expires_on); + assert_eq!( + Some(expected_ts(6_000_000)), + props.immutability_policy_expires_on + ); + assert_eq!( + Some(ImmutabilityPolicyMode::Unlocked), + props.immutability_policy_mode + ); + assert_eq!(Some(false), props.incremental_copy); + assert_eq!(Some(true), props.is_sealed); + assert_eq!(Some(expected_ts(7_000_000)), props.last_accessed_on); + assert_eq!(Some(expected_ts(2_000_000)), props.last_modified); + assert_eq!(Some(LeaseDuration::Infinite), props.lease_duration); + assert_eq!(Some(LeaseState::Available), props.lease_state); + assert_eq!(Some(LeaseStatus::Unlocked), props.lease_status); + assert_eq!(Some(true), props.legal_hold); + assert_eq!(Some(RehydratePriority::High), props.rehydrate_priority); + assert_eq!(Some(7), props.remaining_retention_days); + assert_eq!(Some(true), props.server_encrypted); + assert_eq!(Some(3), props.tag_count); + } + + #[test] + fn null_values_map_to_none() { + let batch = batch(vec![ + ( + "Name", + Arc::new(StringArray::from(vec![Option::::None])), + ), + ( + "Creation-Time", + Arc::new(TimestampMillisecondArray::from(vec![Option::::None])), + ), + ( + "Content-Length", + Arc::new(UInt64Array::from(vec![Option::::None])), + ), + ( + "ServerEncrypted", + Arc::new(BooleanArray::from(vec![Option::::None])), + ), + ( + "BlobType", + Arc::new(StringArray::from(vec![Option::::None])), + ), + ( + "TagCount", + Arc::new(UInt64Array::from(vec![Option::::None])), + ), + ( + "x-ms-blob-sequence-number", + Arc::new(UInt64Array::from(vec![Option::::None])), + ), + ]); + + let item = row_to_blob_item(&batch, 0); + assert_eq!(None, item.name); + let props = item.properties.expect("properties should be set"); + assert_eq!(None, props.creation_time); + assert_eq!(None, props.content_length); + assert_eq!(None, props.server_encrypted); + assert_eq!(None, props.blob_type); + assert_eq!(None, props.tag_count); + assert_eq!(None, props.blob_sequence_number); + } + + #[test] + fn absent_columns_map_to_none() { + // Only `Name` is present; every other field must fall back to `None`. + let batch = batch(vec![("Name", s("only-name"))]); + + let item = row_to_blob_item(&batch, 0); + assert_eq!(Some("only-name".to_string()), item.name); + assert_eq!(None, item.version_id); + let props = item.properties.expect("properties should be set"); + assert_eq!(None, props.content_type); + assert_eq!(None, props.access_tier); + assert_eq!(None, props.creation_time); + assert_eq!(None, props.blob_sequence_number); + } + + // Builds a single-row `map` column from key/value pairs. + fn map_col(entries: &[(&str, &str)]) -> ArrayRef { + let mut builder = MapBuilder::new(None, StringBuilder::new(), StringBuilder::new()); + for (key, value) in entries { + builder.keys().append_value(key); + builder.values().append_value(value); + } + builder.append(true).unwrap(); + Arc::new(builder.finish()) + } + + #[test] + fn map_columns_decode() { + // `Tags`, `Metadata`, and `OrMetadata` decode from Arrow map columns. + let batch = batch(vec![ + ("Name", s("blob.txt")), + ("Tags", map_col(&[("env", "test"), ("team", "sdk")])), + ("Metadata", map_col(&[("team", "sdk")])), + ("OrMetadata", map_col(&[("policy-id", "rule-id")])), + ]); + + let item = row_to_blob_item(&batch, 0); + + let tags = item + .blob_tags + .expect("blob_tags should be populated") + .blob_tag_set + .expect("tag set should be present"); + assert!(tags + .iter() + .any(|t| t.key.as_deref() == Some("env") && t.value.as_deref() == Some("test"))); + assert!(tags + .iter() + .any(|t| t.key.as_deref() == Some("team") && t.value.as_deref() == Some("sdk"))); + + let metadata = item + .metadata + .expect("metadata should be populated") + .values + .expect("metadata values should be present"); + assert_eq!(Some(&"sdk".to_string()), metadata.get("team")); + + let or_metadata = item + .object_replication_metadata + .expect("or_metadata should be populated") + .additional_properties + .expect("or_metadata properties should be present"); + assert_eq!(Some(&"rule-id".to_string()), or_metadata.get("policy-id")); + } + + #[test] + fn absent_map_columns_are_none() { + // Map fields fall back to `None` when the columns are absent. + let batch = batch(vec![("Name", s("only-name"))]); + let item = row_to_blob_item(&batch, 0); + assert!(item.blob_tags.is_none()); + assert!(item.metadata.is_none()); + assert!(item.object_replication_metadata.is_none()); + } + + #[test] + fn decode_stream_maps_rows_and_next_marker() { + let columns = vec![ + ("Name", s("hello.txt")), + ("BlobType", s("BlockBlob")), + ("Content-Length", u64c(10)), + ("ServerEncrypted", b(true)), + ]; + let fields: Vec = columns + .iter() + .map(|(name, array)| Field::new(*name, array.data_type().clone(), true)) + .collect(); + let arrays: Vec = columns.into_iter().map(|(_, array)| array).collect(); + + let mut metadata = HashMap::new(); + metadata.insert(NEXT_MARKER_KEY.to_string(), "next-page".to_string()); + let schema = Arc::new(Schema::new_with_metadata(fields, metadata)); + let record_batch = RecordBatch::try_new(schema.clone(), arrays).unwrap(); + + let mut buffer = Vec::new(); + { + let mut writer = StreamWriter::try_new(&mut buffer, &schema).unwrap(); + writer.write(&record_batch).unwrap(); + writer.finish().unwrap(); + } + + let response = decode_arrow_list_blobs(&buffer).unwrap(); + assert_eq!(Some("next-page".to_string()), response.next_marker); + assert_eq!(1, response.blob_items.len()); + + let item = &response.blob_items[0]; + assert_eq!(Some("hello.txt".to_string()), item.name); + let props = item.properties.as_ref().expect("properties should be set"); + assert_eq!(Some(BlobType::BlockBlob), props.blob_type); + assert_eq!(Some(10), props.content_length); + assert_eq!(Some(true), props.server_encrypted); + } + + #[test] + fn arrow_next_marker_absent_is_none() { + let columns = vec![("Name", s("hello.txt"))]; + let fields: Vec = columns + .iter() + .map(|(name, array)| Field::new(*name, array.data_type().clone(), true)) + .collect(); + let arrays: Vec = columns.into_iter().map(|(_, array)| array).collect(); + let schema = Arc::new(Schema::new(fields)); + let record_batch = RecordBatch::try_new(schema.clone(), arrays).unwrap(); + + let mut buffer = Vec::new(); + { + let mut writer = StreamWriter::try_new(&mut buffer, &schema).unwrap(); + writer.write(&record_batch).unwrap(); + writer.finish().unwrap(); + } + + assert_eq!(None, arrow_next_marker(&buffer).unwrap()); + } + + #[test] + fn missing_content_type_defaults_to_xml() { + assert!(matches!(wire_format(&Headers::new()), Ok(WireFormat::Xml))); + } + + #[test] + fn arrow_contract_covers_every_model_field() { + // Compile-time guard: adding a field to `BlobItem`/`BlobProperties` breaks this + // exhaustive destructure until `row_to_blob_item` is updated to map it, preventing + // the Arrow path from silently dropping model fields the XML path would populate. + let item = row_to_blob_item(&batch(vec![("Name", s("guard"))]), 0); + let BlobItem { + blob_tags: _, + deleted: _, + has_versions_only: _, + is_current_version: _, + metadata: _, + name: _, + object_replication_metadata: _, + properties, + snapshot: _, + version_id: _, + } = item; + let BlobProperties { + access_tier: _, + access_tier_change_time: _, + access_tier_inferred: _, + archive_status: _, + blob_sequence_number: _, + blob_type: _, + cache_control: _, + content_disposition: _, + content_encoding: _, + content_language: _, + content_length: _, + content_md5: _, + content_type: _, + copy_completion_time: _, + copy_id: _, + copy_progress: _, + copy_source: _, + copy_status: _, + copy_status_description: _, + creation_time: _, + deleted_time: _, + destination_snapshot: _, + encryption_key_sha256: _, + encryption_scope: _, + etag: _, + expires_on: _, + immutability_policy_expires_on: _, + immutability_policy_mode: _, + incremental_copy: _, + is_sealed: _, + last_accessed_on: _, + last_modified: _, + lease_duration: _, + lease_state: _, + lease_status: _, + legal_hold: _, + rehydrate_priority: _, + remaining_retention_days: _, + server_encrypted: _, + smart_access_tier: _, + tag_count: _, + } = properties.expect("properties should be set"); + } + + #[test] + fn hierarchy_splits_prefix_rows() { + let names = Arc::new(StringArray::from(vec![ + Some("dir1/"), + Some("blob-a.txt"), + Some("blob-b.txt"), + ])) as ArrayRef; + let resource_types = Arc::new(StringArray::from(vec![ + Some("blobprefix"), + Option::<&str>::None, + Some("blob"), + ])) as ArrayRef; + let schema = Arc::new(Schema::new(vec![ + Field::new("Name", DataType::Utf8, true), + Field::new("ResourceType", DataType::Utf8, true), + ])); + let record_batch = + RecordBatch::try_new(schema.clone(), vec![names, resource_types]).unwrap(); + + let mut buffer = Vec::new(); + { + let mut writer = StreamWriter::try_new(&mut buffer, &schema).unwrap(); + writer.write(&record_batch).unwrap(); + writer.finish().unwrap(); + } + + let response = decode_arrow_list_blobs_hierarchy(&buffer).unwrap(); + let list = response.hierarchical_list; + assert_eq!(2, list.blob_items.len()); + assert_eq!(Some("blob-a.txt".to_string()), list.blob_items[0].name); + assert_eq!(Some("blob-b.txt".to_string()), list.blob_items[1].name); + let prefixes = list.blob_prefixes.expect("prefixes should be present"); + assert_eq!(1, prefixes.len()); + assert_eq!(Some("dir1/".to_string()), prefixes[0].name); + } + + #[test] + fn hierarchy_absent_prefixes_are_none() { + let names = Arc::new(StringArray::from(vec![Some("only-a.txt")])) as ArrayRef; + let schema = Arc::new(Schema::new(vec![Field::new("Name", DataType::Utf8, true)])); + let record_batch = RecordBatch::try_new(schema.clone(), vec![names]).unwrap(); + + let mut buffer = Vec::new(); + { + let mut writer = StreamWriter::try_new(&mut buffer, &schema).unwrap(); + writer.write(&record_batch).unwrap(); + writer.finish().unwrap(); + } + + let response = decode_arrow_list_blobs_hierarchy(&buffer).unwrap(); + assert_eq!(1, response.hierarchical_list.blob_items.len()); + assert!(response.hierarchical_list.blob_prefixes.is_none()); + } + + #[test] + fn hierarchy_contract_covers_every_model_field() { + // Compile-time guard: adding a field to the hierarchy models breaks this destructure until + // the Arrow decoder is updated, preventing silently dropped data. + let ListBlobsHierarchicalResponse { + container_name: _, + delimiter: _, + hierarchical_list, + marker: _, + max_results: _, + next_marker: _, + prefix: _, + service_endpoint: _, + } = ListBlobsHierarchicalResponse::default(); + let BlobHierarchyList { + blob_items: _, + blob_prefixes: _, + } = hierarchical_list; + let BlobPrefix { name: _ } = BlobPrefix::default(); + } +} diff --git a/sdk/storage/azure_storage_blob/src/clients/blob_container_client.rs b/sdk/storage/azure_storage_blob/src/clients/blob_container_client.rs index 693e026de84..affaaf9be04 100644 --- a/sdk/storage/azure_storage_blob/src/clients/blob_container_client.rs +++ b/sdk/storage/azure_storage_blob/src/clients/blob_container_client.rs @@ -3,13 +3,22 @@ pub use crate::generated::clients::{BlobContainerClient, BlobContainerClientOptions}; -use crate::{models::StorageErrorCode, BlobClient}; +use crate::{ + arrow_decode::{decode_next_marker, AutoFormat}, + models::{ + BlobContainerClientListBlobsHierarchicalOptions, BlobContainerClientListBlobsOptions, + ListBlobsHierarchicalResponse, ListBlobsResponse, StorageErrorCode, + }, + BlobClient, +}; use azure_core::{ credentials::TokenCredential, - error::ErrorKind, + error::{CheckSuccessOptions, ErrorKind}, http::{ + pager::{PagerContinuation, PagerResult, PagerState}, policies::{auth::BearerTokenAuthorizationPolicy, Policy}, - Pipeline, StatusCode, Url, + Method, Pager, Pipeline, PipelineSendOptions, RawResponse, Request, StatusCode, Url, + UrlExt, }, tracing, Result, }; @@ -114,21 +123,237 @@ impl BlobContainerClient { Err(e) => Err(e), } } + + /// Returns a list of the blobs in the specified container. + /// + /// Apache Arrow is requested by default, with automatic XML fallback. To require XML, set + /// [`BlobContainerClientListBlobsOptions::response_format`] to + /// [`StorageResponseFormat::Xml`](crate::models::StorageResponseFormat::Xml). See + /// [`StorageResponseFormat`](crate::models::StorageResponseFormat) for the available response + /// formats. + /// + /// Over Apache Arrow the service returns only the blob rows and next marker, so the response + /// envelope fields (`container_name`, `marker`, `max_results`, `prefix`, `service_endpoint`) + /// are `None`; request XML to populate them. + /// + /// # Arguments + /// + /// * `options` - Optional parameters for the request. + /// + pub fn list_blobs( + &self, + options: Option>, + ) -> Result> { + let options = options.unwrap_or_default().into_owned(); + let accept = options + .response_format + .unwrap_or_default() + .as_header_value(); + let pager_options = options.method_options.clone(); + let pipeline = self.pipeline.clone(); + let version = self.version.clone(); + let mut first_url = self.endpoint.clone(); + let mut query_builder = first_url.query_builder(); + query_builder + .append_pair("comp", "list") + .append_pair("restype", "container"); + if let Some(end_before) = options.end_before.as_ref() { + query_builder.set_pair("endBefore", end_before); + } + if let Some(include) = options.include.as_ref() { + query_builder.set_pair( + "include", + include + .iter() + .map(|i| i.to_string()) + .collect::>() + .join(","), + ); + } + if let Some(marker) = options.marker.as_ref() { + query_builder.set_pair("marker", marker); + } + if let Some(maxresults) = options.maxresults { + query_builder.set_pair("maxresults", maxresults.to_string()); + } + if let Some(prefix) = options.prefix.as_ref() { + query_builder.set_pair("prefix", prefix); + } + if let Some(start_from) = options.start_from.as_ref() { + query_builder.set_pair("startFrom", start_from); + } + if let Some(timeout) = options.timeout { + query_builder.set_pair("timeout", timeout.to_string()); + } + query_builder.build(); + + Ok(Pager::new( + move |state: PagerState, pager_options| { + let mut url = first_url.clone(); + if let PagerState::More(continuation) = state { + let mut query_builder = url.query_builder(); + query_builder.set_pair("marker", continuation.as_ref()); + query_builder.build(); + } + let mut request = Request::new(url, Method::Get); + request.insert_header("accept", accept); + request.insert_header("x-ms-version", &version); + let pipeline = pipeline.clone(); + Box::pin(async move { + let response = pipeline + .send( + &pager_options.context, + &mut request, + Some(PipelineSendOptions { + check_success: CheckSuccessOptions { + success_codes: &[200], + }, + ..Default::default() + }), + ) + .await?; + let (status, headers, body) = response.deconstruct(); + let next_marker = decode_next_marker(&headers, &body)?; + let response = RawResponse::from_bytes(status, headers, body).into(); + Ok(match next_marker { + Some(next_marker) => PagerResult::More { + response, + continuation: PagerContinuation::Token(next_marker), + }, + None => PagerResult::Done { response }, + }) + }) + }, + Some(pager_options), + )) + } + + /// Returns a list of the blobs in the specified container, grouping blobs under virtual + /// directories using `delimiter`. + /// + /// Apache Arrow is requested by default, with automatic XML fallback, matching + /// [`list_blobs`](Self::list_blobs). Virtual directories are returned as + /// [`BlobPrefix`](crate::models::BlobPrefix) entries on the page's + /// [`hierarchical_list`](crate::models::BlobHierarchyList). + /// + /// Over Apache Arrow the service returns only the blob rows, prefixes, and next marker, so the + /// response envelope fields (including `delimiter`, `container_name`, and `prefix`) are `None`; + /// request XML to populate them. + /// + /// # Arguments + /// + /// * `delimiter` - Groups blobs whose names share a common substring up to this separator into + /// a single `BlobPrefix` placeholder. + /// * `options` - Optional parameters for the request. + pub fn list_blobs_hierarchical( + &self, + delimiter: &str, + options: Option>, + ) -> Result> { + let options = options.unwrap_or_default().into_owned(); + let accept = options + .response_format + .unwrap_or_default() + .as_header_value(); + let pager_options = options.method_options.clone(); + let pipeline = self.pipeline.clone(); + let version = self.version.clone(); + let mut first_url = self.endpoint.clone(); + let mut query_builder = first_url.query_builder(); + query_builder + .append_pair("comp", "list") + .append_pair("restype", "container"); + query_builder.set_pair("delimiter", delimiter); + if let Some(end_before) = options.end_before.as_ref() { + query_builder.set_pair("endBefore", end_before); + } + if let Some(include) = options.include.as_ref() { + query_builder.set_pair( + "include", + include + .iter() + .map(|i| i.to_string()) + .collect::>() + .join(","), + ); + } + if let Some(marker) = options.marker.as_ref() { + query_builder.set_pair("marker", marker); + } + if let Some(maxresults) = options.maxresults { + query_builder.set_pair("maxresults", maxresults.to_string()); + } + if let Some(prefix) = options.prefix.as_ref() { + query_builder.set_pair("prefix", prefix); + } + if let Some(start_from) = options.start_from.as_ref() { + query_builder.set_pair("startFrom", start_from); + } + if let Some(timeout) = options.timeout { + query_builder.set_pair("timeout", timeout.to_string()); + } + query_builder.build(); + + Ok(Pager::new( + move |state: PagerState, pager_options| { + let mut url = first_url.clone(); + if let PagerState::More(continuation) = state { + let mut query_builder = url.query_builder(); + query_builder.set_pair("marker", continuation.as_ref()); + query_builder.build(); + } + let mut request = Request::new(url, Method::Get); + request.insert_header("accept", accept); + request.insert_header("x-ms-version", &version); + let pipeline = pipeline.clone(); + Box::pin(async move { + let response = pipeline + .send( + &pager_options.context, + &mut request, + Some(PipelineSendOptions { + check_success: CheckSuccessOptions { + success_codes: &[200], + }, + ..Default::default() + }), + ) + .await?; + let (status, headers, body) = response.deconstruct(); + let next_marker = decode_next_marker(&headers, &body)?; + let response = RawResponse::from_bytes(status, headers, body).into(); + Ok(match next_marker { + Some(next_marker) => PagerResult::More { + response, + continuation: PagerContinuation::Token(next_marker), + }, + None => PagerResult::Done { response }, + }) + }) + }, + Some(pager_options), + )) + } } #[cfg(test)] mod tests { use super::*; + use crate::models::StorageResponseFormat; + use arrow::array::{builder::StringBuilder, RecordBatch}; + use arrow::datatypes::{DataType, Field, Schema}; + use arrow::ipc::writer::StreamWriter; use azure_core::{ http::{ - headers::Headers, pager::PagerContinuation, AsyncRawResponse, ClientOptions, - StatusCode, Transport, + headers::{Headers, ACCEPT, CONTENT_TYPE}, + pager::{PagerContinuation, PagerOptions}, + AsyncRawResponse, ClientOptions, StatusCode, Transport, }, Bytes, }; use azure_core_test::http::MockHttpClient; use futures::{FutureExt as _, TryStreamExt as _}; - use std::sync::Arc; + use std::{collections::HashMap, sync::Arc}; const LIST_BLOBS_PAGE: &[u8] = br#" @@ -143,6 +368,23 @@ mod tests { page-2 "#; + const XML_PAGE_1: &[u8] = br#" + + + page1-a.txtBlockBlob + page1-b.txtBlockBlob + + page2 +"#; + + const XML_PAGE_2: &[u8] = br#" + + + page2-a.txtBlockBlob + page2-b.txtBlockBlob + +"#; + #[test] fn from_url_rejects_cannot_be_a_base_url() { let url = Url::parse("data:text/plain,hello").unwrap(); @@ -183,17 +425,7 @@ mod tests { } .boxed() })); - let client = BlobContainerClient::new( - Url::parse("https://example.blob.core.windows.net/container").unwrap(), - None, - Some(BlobContainerClientOptions { - client_options: ClientOptions { - transport: Some(Transport::new(mock_client)), - ..Default::default() - }, - ..Default::default() - }), - )?; + let client = container_client_with(mock_client); let mut pages = client.list_blobs(None)?.into_pages(); let page = pages.try_next().await?.expect("expected a page"); @@ -210,4 +442,395 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn list_blobs_mock_arrow_all_pages() -> Result<()> { + let client = container_client_with(arrow_mock_client()); + let names = collect_blob_names(client.list_blobs(None)?).await?; + assert_eq!( + names, + ["page1-a.txt", "page1-b.txt", "page2-a.txt", "page2-b.txt"] + ); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_mock_arrow_xml_fallback() -> Result<()> { + // Arrow is requested (the default), but the service replies with XML; the pager must + // still decode every blob across both pages via the XML fallback path. + let client = container_client_with(xml_mock_client_with_accept( + "application/vnd.apache.arrow.stream,application/xml", + )); + let names = collect_blob_names(client.list_blobs(None)?).await?; + assert_eq!( + names, + ["page1-a.txt", "page1-b.txt", "page2-a.txt", "page2-b.txt"] + ); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_mock_arrow_from_continuation() -> Result<()> { + let client = container_client_with(arrow_mock_client()); + let options = BlobContainerClientListBlobsOptions { + method_options: PagerOptions { + continuation: Some(PagerContinuation::Token("page2".into())), + ..Default::default() + }, + ..Default::default() + }; + let names = collect_blob_names(client.list_blobs(Some(options))?).await?; + assert_eq!(names, ["page2-a.txt", "page2-b.txt"]); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_mock_explicit_xml() -> Result<()> { + let client = container_client_with(xml_mock_client_with_accept("application/xml")); + let options = BlobContainerClientListBlobsOptions { + response_format: Some(StorageResponseFormat::Xml), + ..Default::default() + }; + let names = collect_blob_names(client.list_blobs(Some(options))?).await?; + assert_eq!( + names, + ["page1-a.txt", "page1-b.txt", "page2-a.txt", "page2-b.txt"] + ); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_mock_arrow_sends_end_before() -> Result<()> { + // The end_before option flows out as the `endBefore` query parameter. + let client = container_client_with(Arc::new(MockHttpClient::new(|req| { + assert!(req + .url() + .query_pairs() + .any(|(k, v)| k == "endBefore" && v == "cc.txt")); + let body = build_arrow_list_blobs(&["aa.txt"], None); + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/vnd.apache.arrow.stream"); + Ok(AsyncRawResponse::from_bytes(StatusCode::Ok, headers, body)) + } + .boxed() + }))); + let options = BlobContainerClientListBlobsOptions { + response_format: Some(StorageResponseFormat::Arrow), + end_before: Some("cc.txt".to_string()), + ..Default::default() + }; + let names = collect_blob_names(client.list_blobs(Some(options))?).await?; + assert_eq!(names, ["aa.txt"]); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_mock_arrow_drops_envelope() -> Result<()> { + // Arrow carries only blob rows and the next marker; envelope fields stay None. + let client = container_client_with(arrow_mock_client()); + let page = client + .list_blobs(None)? + .into_pages() + .try_next() + .await? + .expect("expected a page") + .into_model()?; + assert!(page.container_name.is_none()); + assert!(page.prefix.is_none()); + assert!(page.max_results.is_none()); + assert!(page.service_endpoint.is_none()); + Ok(()) + } + + fn container_client_with(mock: Arc) -> BlobContainerClient { + BlobContainerClient::new( + Url::parse("https://example.blob.core.windows.net/container").unwrap(), + None, + Some(BlobContainerClientOptions { + client_options: ClientOptions { + transport: Some(Transport::new(mock)), + ..Default::default() + }, + ..Default::default() + }), + ) + .unwrap() + } + + async fn collect_blob_names( + pager: Pager, + ) -> Result> { + let mut pages = pager.into_pages(); + let mut names = Vec::new(); + while let Some(page) = pages.try_next().await? { + let model = page.into_model()?; + names.extend(model.blob_items.into_iter().filter_map(|b| b.name)); + } + Ok(names) + } + + fn build_arrow_list_blobs(names: &[&str], next_marker: Option<&str>) -> Bytes { + let metadata: HashMap = next_marker + .map(|m| HashMap::from([("NextMarker".to_string(), m.to_string())])) + .unwrap_or_default(); + let schema = Arc::new(Schema::new_with_metadata( + vec![Field::new("Name", DataType::Utf8, true)], + metadata, + )); + let mut builder = StringBuilder::new(); + for name in names { + builder.append_value(name); + } + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(builder.finish())]) + .expect("valid batch"); + let mut buf = Vec::new(); + let mut writer = StreamWriter::try_new(&mut buf, &schema).expect("valid writer"); + writer.write(&batch).expect("write batch"); + writer.finish().expect("finish"); + Bytes::from(buf) + } + + fn arrow_mock_client() -> Arc { + let page1 = build_arrow_list_blobs(&["page1-a.txt", "page1-b.txt"], Some("page2")); + let page2 = build_arrow_list_blobs(&["page2-a.txt", "page2-b.txt"], None); + Arc::new(MockHttpClient::new(move |req| { + assert_eq!( + req.headers().get_str(&ACCEPT).unwrap(), + "application/vnd.apache.arrow.stream,application/xml" + ); + let is_page2 = req + .url() + .query_pairs() + .any(|(k, v)| k == "marker" && v == "page2"); + let body = if is_page2 { + page2.clone() + } else { + page1.clone() + }; + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/vnd.apache.arrow.stream"); + Ok(AsyncRawResponse::from_bytes(StatusCode::Ok, headers, body)) + } + .boxed() + })) + } + + fn xml_mock_client_with_accept(accept: &'static str) -> Arc { + Arc::new(MockHttpClient::new(move |req| { + assert_eq!(req.headers().get_str(&ACCEPT).unwrap(), accept); + let is_page2 = req + .url() + .query_pairs() + .any(|(k, v)| k == "marker" && v == "page2"); + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/xml"); + let body = if is_page2 { XML_PAGE_2 } else { XML_PAGE_1 }; + Ok(AsyncRawResponse::from_bytes( + StatusCode::Ok, + headers, + Bytes::from_static(body), + )) + } + .boxed() + })) + } + + const XML_HIERARCHY_PAGE: &[u8] = br#" + + / + + dir1/ + top.txtBlockBlob + +"#; + + fn build_arrow_hierarchy( + blobs: &[&str], + prefixes: &[&str], + next_marker: Option<&str>, + ) -> Bytes { + let metadata: HashMap = next_marker + .map(|m| HashMap::from([("NextMarker".to_string(), m.to_string())])) + .unwrap_or_default(); + let schema = Arc::new(Schema::new_with_metadata( + vec![ + Field::new("Name", DataType::Utf8, true), + Field::new("ResourceType", DataType::Utf8, true), + ], + metadata, + )); + let mut names = StringBuilder::new(); + let mut resource_types = StringBuilder::new(); + for prefix in prefixes { + names.append_value(prefix); + resource_types.append_value("blobprefix"); + } + for blob in blobs { + names.append_value(blob); + resource_types.append_value("blob"); + } + let batch = RecordBatch::try_new( + schema.clone(), + vec![Arc::new(names.finish()), Arc::new(resource_types.finish())], + ) + .expect("valid batch"); + let mut buf = Vec::new(); + let mut writer = StreamWriter::try_new(&mut buf, &schema).expect("valid writer"); + writer.write(&batch).expect("write batch"); + writer.finish().expect("finish"); + Bytes::from(buf) + } + + fn arrow_hierarchy_mock_client() -> Arc { + let page = build_arrow_hierarchy(&["top.txt"], &["dir1/", "dir2/"], None); + Arc::new(MockHttpClient::new(move |req| { + assert_eq!( + req.headers().get_str(&ACCEPT).unwrap(), + "application/vnd.apache.arrow.stream,application/xml" + ); + assert!(req + .url() + .query_pairs() + .any(|(k, v)| k == "delimiter" && v == "/")); + let page = page.clone(); + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/vnd.apache.arrow.stream"); + Ok(AsyncRawResponse::from_bytes(StatusCode::Ok, headers, page)) + } + .boxed() + })) + } + + fn xml_hierarchy_mock_client(accept: &'static str) -> Arc { + Arc::new(MockHttpClient::new(move |req| { + assert_eq!(req.headers().get_str(&ACCEPT).unwrap(), accept); + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/xml"); + Ok(AsyncRawResponse::from_bytes( + StatusCode::Ok, + headers, + Bytes::from_static(XML_HIERARCHY_PAGE), + )) + } + .boxed() + })) + } + + fn arrow_hierarchy_mock_client_paged() -> Arc { + let page1 = build_arrow_hierarchy(&["a.txt"], &["dir1/"], Some("page2")); + let page2 = build_arrow_hierarchy(&["b.txt"], &["dir2/"], None); + Arc::new(MockHttpClient::new(move |req| { + assert_eq!( + req.headers().get_str(&ACCEPT).unwrap(), + "application/vnd.apache.arrow.stream,application/xml" + ); + let is_page2 = req + .url() + .query_pairs() + .any(|(k, v)| k == "marker" && v == "page2"); + let body = if is_page2 { + page2.clone() + } else { + page1.clone() + }; + async move { + let mut headers = Headers::new(); + headers.insert(CONTENT_TYPE, "application/vnd.apache.arrow.stream"); + Ok(AsyncRawResponse::from_bytes(StatusCode::Ok, headers, body)) + } + .boxed() + })) + } + + #[tokio::test] + async fn list_blobs_hierarchical_mock_arrow() -> Result<()> { + let client = container_client_with(arrow_hierarchy_mock_client()); + let page = client + .list_blobs_hierarchical("/", None)? + .into_pages() + .try_next() + .await? + .expect("expected a page") + .into_model()?; + + let blobs: Vec<_> = page + .hierarchical_list + .blob_items + .iter() + .filter_map(|b| b.name.as_deref()) + .collect(); + assert_eq!(blobs, ["top.txt"]); + + let prefixes = page + .hierarchical_list + .blob_prefixes + .expect("prefixes should be present"); + let prefix_names: Vec<_> = prefixes.iter().filter_map(|p| p.name.as_deref()).collect(); + assert_eq!(prefix_names, ["dir1/", "dir2/"]); + + // Arrow omits the response envelope fields, including the delimiter. + assert!(page.delimiter.is_none()); + assert!(page.container_name.is_none()); + assert!(page.prefix.is_none()); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_hierarchical_mock_arrow_xml_fallback() -> Result<()> { + // Arrow requested (default) but the service replies with XML; prefixes, blobs, and the + // delimiter (which Arrow omits) all decode via the XML fallback. + let client = container_client_with(xml_hierarchy_mock_client( + "application/vnd.apache.arrow.stream,application/xml", + )); + let page = client + .list_blobs_hierarchical("/", None)? + .into_pages() + .try_next() + .await? + .expect("expected a page") + .into_model()?; + + assert_eq!(page.delimiter.as_deref(), Some("/")); + let prefixes = page + .hierarchical_list + .blob_prefixes + .expect("prefixes should be present"); + assert_eq!(prefixes[0].name.as_deref(), Some("dir1/")); + assert_eq!( + page.hierarchical_list.blob_items[0].name.as_deref(), + Some("top.txt") + ); + Ok(()) + } + + #[tokio::test] + async fn list_blobs_hierarchical_mock_arrow_all_pages() -> Result<()> { + let client = container_client_with(arrow_hierarchy_mock_client_paged()); + let mut pages = client.list_blobs_hierarchical("/", None)?.into_pages(); + + let mut blobs = Vec::new(); + let mut prefixes = Vec::new(); + while let Some(page) = pages.try_next().await? { + let page = page.into_model()?; + blobs.extend( + page.hierarchical_list + .blob_items + .into_iter() + .filter_map(|b| b.name), + ); + if let Some(page_prefixes) = page.hierarchical_list.blob_prefixes { + prefixes.extend(page_prefixes.into_iter().filter_map(|p| p.name)); + } + } + + // Blobs and prefixes from both pages aggregate across the NextMarker boundary. + assert_eq!(blobs, ["a.txt", "b.txt"]); + assert_eq!(prefixes, ["dir1/", "dir2/"]); + Ok(()) + } } diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/append_blob_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/append_blob_client.rs index 29e4247558b..b4aed22dab0 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/append_blob_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/append_blob_client.rs @@ -581,7 +581,7 @@ impl AppendBlobClient { } /// Default value for [`AppendBlobClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for AppendBlobClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/blob_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/blob_client.rs index 546fa894298..929476d8bc4 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/blob_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/blob_client.rs @@ -1771,7 +1771,7 @@ impl BlobClient { } /// Default value for [`BlobClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for BlobClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/blob_container_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/blob_container_client.rs index 38a71b79a81..ac942b441b4 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/blob_container_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/blob_container_client.rs @@ -11,19 +11,24 @@ use crate::generated::models::{ BlobContainerClientFindBlobsByTagsOptions, BlobContainerClientGetAccessPolicyOptions, BlobContainerClientGetAccountInfoOptions, BlobContainerClientGetAccountInfoResult, BlobContainerClientGetPropertiesOptions, BlobContainerClientGetPropertiesResult, - BlobContainerClientListBlobsOptions, BlobContainerClientReleaseLeaseOptions, + BlobContainerClientListBlobsHierarchicalInternalOptions, + BlobContainerClientListBlobsHierarchicalInternalResult, + BlobContainerClientListBlobsHierarchicalXmlOptions, + BlobContainerClientListBlobsInternalOptions, BlobContainerClientListBlobsInternalResult, + BlobContainerClientListBlobsXmlOptions, BlobContainerClientReleaseLeaseOptions, BlobContainerClientReleaseLeaseResult, BlobContainerClientRenewLeaseOptions, BlobContainerClientRenewLeaseResult, BlobContainerClientSetAccessPolicyOptions, - BlobContainerClientSetMetadataOptions, FilteredBlobResponse, ListBlobsResponse, - SignedIdentifiers, + BlobContainerClientSetMetadataOptions, FilteredBlobResponse, ListBlobsHierarchicalResponse, + ListBlobsResponse, SignedIdentifiers, }; use azure_core::{ error::CheckSuccessOptions, fmt::SafeDebug, http::{ pager::{PagerContinuation, PagerResult, PagerState}, - ClientOptions, Method, NoFormat, Pager, Pipeline, PipelineSendOptions, RawResponse, - Request, RequestContent, Response, Url, UrlExt, XmlFormat, + AsyncResponse, ClientOptions, Method, NoFormat, Pager, Pipeline, PipelineSendOptions, + PipelineStreamOptions, RawResponse, Request, RequestContent, Response, Url, UrlExt, + XmlFormat, }, time::to_rfc7231, tracing, xml, Result, @@ -707,15 +712,299 @@ impl BlobContainerClient { Ok(rsp.into()) } + /// Returns a list of the blobs as raw data, to be deserialized by the client. A delimiter can be used to traverse a virtual + /// hierarchy of blobs as though it were a file system. + /// + /// # Arguments + /// + /// * `accept` - The Accept header indicating the requested response media type. + /// * `delimiter` - If specified, the operation returns a BlobPrefix element that acts as a placeholder for all blobs whose + /// names begin with the same substring up to the appearance of the delimiter character. The delimiter may be a single character + /// or a string. + /// * `options` - Optional parameters for the request. + /// + /// ## Response Headers + /// + /// The returned [`AsyncResponse`](azure_core::http::AsyncResponse) implements the [`BlobContainerClientListBlobsHierarchicalInternalResultHeaders`] trait, which provides + /// access to response headers. For example: + /// + /// ```ignore + /// use azure_core::{Result, http::AsyncResponse}; + /// use azure_storage_blob::models::{BlobContainerClientListBlobsHierarchicalInternalResult, BlobContainerClientListBlobsHierarchicalInternalResultHeaders}; + /// async fn example() -> Result<()> { + /// let response: AsyncResponse = unimplemented!(); + /// // Access response headers + /// if let Some(content_type) = response.content_type()? { + /// println!("content-type: {:?}", content_type); + /// } + /// Ok(()) + /// } + /// ``` + /// + /// ### Available headers + /// * [`content_type`()](crate::generated::models::BlobContainerClientListBlobsHierarchicalInternalResultHeaders::content_type) - content-type + /// + /// [`BlobContainerClientListBlobsHierarchicalInternalResultHeaders`]: crate::generated::models::BlobContainerClientListBlobsHierarchicalInternalResultHeaders + #[tracing::function("Storage.Blob.BlobContainerClient.listBlobsHierarchicalInternal")] + pub(crate) async fn list_blobs_hierarchical_internal( + &self, + accept: String, + delimiter: &str, + options: Option>, + ) -> Result> { + let options = options.unwrap_or_default(); + let ctx = options.method_options.context.to_borrowed(); + let mut url = self.endpoint.clone(); + let mut query_builder = url.query_builder(); + query_builder + .append_pair("comp", "list") + .append_pair("restype", "container"); + query_builder.set_pair("delimiter", delimiter); + if let Some(end_before) = options.end_before.as_ref() { + query_builder.set_pair("endBefore", end_before); + } + if let Some(include) = options.include.as_ref() { + query_builder.set_pair( + "include", + include + .iter() + .map(|i| i.to_string()) + .collect::>() + .join(","), + ); + } + if let Some(marker) = options.marker.as_ref() { + query_builder.set_pair("marker", marker); + } + if let Some(maxresults) = options.maxresults { + query_builder.set_pair("maxresults", maxresults.to_string()); + } + if let Some(prefix) = options.prefix.as_ref() { + query_builder.set_pair("prefix", prefix); + } + if let Some(start_from) = options.start_from.as_ref() { + query_builder.set_pair("startFrom", start_from); + } + if let Some(timeout) = options.timeout { + query_builder.set_pair("timeout", timeout.to_string()); + } + query_builder.build(); + let mut request = Request::new(url, Method::Get); + request.insert_header("accept", accept); + request.insert_header("x-ms-version", &self.version); + let rsp = self + .pipeline + .stream( + &ctx, + &mut request, + Some(PipelineStreamOptions { + check_success: CheckSuccessOptions { + success_codes: &[200], + }, + ..Default::default() + }), + ) + .await?; + Ok(rsp.into()) + } + + /// Returns a list of the blobs in the specified container. A delimiter can be used to traverse a virtual hierarchy of blobs + /// as though it were a file system. + /// + /// # Arguments + /// + /// * `delimiter` - If specified, the operation returns a BlobPrefix element that acts as a placeholder for all blobs whose + /// names begin with the same substring up to the appearance of the delimiter character. The delimiter may be a single character + /// or a string. + /// * `options` - Optional parameters for the request. + #[tracing::function("Storage.Blob.BlobContainerClient.listBlobHierarchySegment")] + pub(crate) fn list_blobs_hierarchical_xml( + &self, + delimiter: &str, + options: Option>, + ) -> Result> { + let options = options.unwrap_or_default().into_owned(); + let pipeline = self.pipeline.clone(); + let mut first_url = self.endpoint.clone(); + let mut query_builder = first_url.query_builder(); + query_builder + .append_pair("comp", "list") + .append_pair("restype", "container"); + query_builder.set_pair("delimiter", delimiter); + if let Some(include) = options.include.as_ref() { + query_builder.set_pair( + "include", + include + .iter() + .map(|i| i.to_string()) + .collect::>() + .join(","), + ); + } + if let Some(marker) = options.marker.as_ref() { + query_builder.set_pair("marker", marker); + } + if let Some(maxresults) = options.maxresults { + query_builder.set_pair("maxresults", maxresults.to_string()); + } + if let Some(prefix) = options.prefix.as_ref() { + query_builder.set_pair("prefix", prefix); + } + if let Some(start_from) = options.start_from.as_ref() { + query_builder.set_pair("startFrom", start_from); + } + if let Some(timeout) = options.timeout { + query_builder.set_pair("timeout", timeout.to_string()); + } + query_builder.build(); + #[derive(serde::Deserialize)] + struct BlobContainerClientListBlobsHierarchicalXmlPage { + #[serde(rename = "NextMarker")] + next_marker: Option, + } + + let version = self.version.clone(); + Ok(Pager::new( + move |marker: PagerState, pager_options| { + let mut url = first_url.clone(); + if let PagerState::More(marker) = marker { + let mut query_builder = url.query_builder(); + query_builder.set_pair("marker", marker.as_ref()); + query_builder.build(); + } + let mut request = Request::new(url, Method::Get); + request.insert_header("accept", "application/xml"); + request.insert_header("x-ms-version", &version); + let pipeline = pipeline.clone(); + Box::pin(async move { + let rsp = pipeline + .send( + &pager_options.context, + &mut request, + Some(PipelineSendOptions { + check_success: CheckSuccessOptions { + success_codes: &[200], + }, + ..Default::default() + }), + ) + .await?; + let (status, headers, body) = rsp.deconstruct(); + let res: BlobContainerClientListBlobsHierarchicalXmlPage = + xml::from_xml(&body)?; + let rsp = RawResponse::from_bytes(status, headers, body).into(); + Ok(match res.next_marker { + Some(next_marker) if !next_marker.is_empty() => PagerResult::More { + response: rsp, + continuation: PagerContinuation::Token(next_marker), + }, + _ => PagerResult::Done { response: rsp }, + }) + }) + }, + Some(options.method_options), + )) + } + + /// Returns a list of the blobs as raw data, to be deserialized by the client. + /// + /// # Arguments + /// + /// * `accept` - The Accept header indicating the requested response media type. + /// * `options` - Optional parameters for the request. + /// + /// ## Response Headers + /// + /// The returned [`AsyncResponse`](azure_core::http::AsyncResponse) implements the [`BlobContainerClientListBlobsInternalResultHeaders`] trait, which provides + /// access to response headers. For example: + /// + /// ```ignore + /// use azure_core::{Result, http::AsyncResponse}; + /// use azure_storage_blob::models::{BlobContainerClientListBlobsInternalResult, BlobContainerClientListBlobsInternalResultHeaders}; + /// async fn example() -> Result<()> { + /// let response: AsyncResponse = unimplemented!(); + /// // Access response headers + /// if let Some(content_type) = response.content_type()? { + /// println!("content-type: {:?}", content_type); + /// } + /// Ok(()) + /// } + /// ``` + /// + /// ### Available headers + /// * [`content_type`()](crate::generated::models::BlobContainerClientListBlobsInternalResultHeaders::content_type) - content-type + /// + /// [`BlobContainerClientListBlobsInternalResultHeaders`]: crate::generated::models::BlobContainerClientListBlobsInternalResultHeaders + #[tracing::function("Storage.Blob.BlobContainerClient.listBlobsInternal")] + pub(crate) async fn list_blobs_internal( + &self, + accept: String, + options: Option>, + ) -> Result> { + let options = options.unwrap_or_default(); + let ctx = options.method_options.context.to_borrowed(); + let mut url = self.endpoint.clone(); + let mut query_builder = url.query_builder(); + query_builder + .append_pair("comp", "list") + .append_pair("restype", "container"); + if let Some(end_before) = options.end_before.as_ref() { + query_builder.set_pair("endBefore", end_before); + } + if let Some(include) = options.include.as_ref() { + query_builder.set_pair( + "include", + include + .iter() + .map(|i| i.to_string()) + .collect::>() + .join(","), + ); + } + if let Some(marker) = options.marker.as_ref() { + query_builder.set_pair("marker", marker); + } + if let Some(maxresults) = options.maxresults { + query_builder.set_pair("maxresults", maxresults.to_string()); + } + if let Some(prefix) = options.prefix.as_ref() { + query_builder.set_pair("prefix", prefix); + } + if let Some(start_from) = options.start_from.as_ref() { + query_builder.set_pair("startFrom", start_from); + } + if let Some(timeout) = options.timeout { + query_builder.set_pair("timeout", timeout.to_string()); + } + query_builder.build(); + let mut request = Request::new(url, Method::Get); + request.insert_header("accept", accept); + request.insert_header("x-ms-version", &self.version); + let rsp = self + .pipeline + .stream( + &ctx, + &mut request, + Some(PipelineStreamOptions { + check_success: CheckSuccessOptions { + success_codes: &[200], + }, + ..Default::default() + }), + ) + .await?; + Ok(rsp.into()) + } + /// Returns a list of the blobs in the specified container. /// /// # Arguments /// /// * `options` - Optional parameters for the request. #[tracing::function("Storage.Blob.BlobContainerClient.listBlobs")] - pub fn list_blobs( + pub(crate) fn list_blobs_xml( &self, - options: Option>, + options: Option>, ) -> Result> { let options = options.unwrap_or_default().into_owned(); let pipeline = self.pipeline.clone(); @@ -751,7 +1040,7 @@ impl BlobContainerClient { } query_builder.build(); #[derive(serde::Deserialize)] - struct BlobContainerClientListBlobsPage { + struct BlobContainerClientListBlobsXmlPage { #[serde(rename = "NextMarker")] next_marker: Option, } @@ -783,7 +1072,7 @@ impl BlobContainerClient { ) .await?; let (status, headers, body) = rsp.deconstruct(); - let res: BlobContainerClientListBlobsPage = xml::from_xml(&body)?; + let res: BlobContainerClientListBlobsXmlPage = xml::from_xml(&body)?; let rsp = RawResponse::from_bytes(status, headers, body).into(); Ok(match res.next_marker { Some(next_marker) if !next_marker.is_empty() => PagerResult::More { @@ -1061,7 +1350,7 @@ impl BlobContainerClient { } /// Default value for [`BlobContainerClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for BlobContainerClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/blob_service_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/blob_service_client.rs index f69d1c595c9..8a36b503336 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/blob_service_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/blob_service_client.rs @@ -454,7 +454,7 @@ impl BlobServiceClient { } /// Default value for [`BlobServiceClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for BlobServiceClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/block_blob_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/block_blob_client.rs index 8140b0689d8..881c2434caf 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/block_blob_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/block_blob_client.rs @@ -932,7 +932,7 @@ impl BlockBlobClient { } /// Default value for [`BlockBlobClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for BlockBlobClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/clients/page_blob_client.rs b/sdk/storage/azure_storage_blob/src/generated/clients/page_blob_client.rs index 0b5bccc4ff2..9b2b6a1f3c4 100644 --- a/sdk/storage/azure_storage_blob/src/generated/clients/page_blob_client.rs +++ b/sdk/storage/azure_storage_blob/src/generated/clients/page_blob_client.rs @@ -1005,7 +1005,7 @@ impl PageBlobClient { } /// Default value for [`PageBlobClientOptions::version`]. -pub(crate) const DEFAULT_VERSION: &str = "2026-10-06"; +pub(crate) const DEFAULT_VERSION: &str = "2026-12-06"; impl Default for PageBlobClientOptions { fn default() -> Self { diff --git a/sdk/storage/azure_storage_blob/src/generated/models/enums.rs b/sdk/storage/azure_storage_blob/src/generated/models/enums.rs index fa368fc1ac2..5e6a7143912 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/enums.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/enums.rs @@ -263,6 +263,13 @@ pub enum LeaseStatus { Unlocked, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ListBlobsHierarchicalInternalResponseContentType { + ApplicationVndApacheArrowStream, + + ApplicationXml, +} + /// Specifies additional datasets to include when listing blobs in a container. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum ListBlobsIncludeItem { @@ -297,6 +304,13 @@ pub enum ListBlobsIncludeItem { Versions, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ListBlobsInternalResponseContentType { + ApplicationVndApacheArrowStream, + + ApplicationXml, +} + /// Specifies what additional information should be returned as part of the list operation. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum ListContainersIncludeType { diff --git a/sdk/storage/azure_storage_blob/src/generated/models/enums_impl.rs b/sdk/storage/azure_storage_blob/src/generated/models/enums_impl.rs index 0ee7be028d0..3e4f22db08b 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/enums_impl.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/enums_impl.rs @@ -7,7 +7,8 @@ use super::{ AccessTier, AccountKind, ArchiveStatus, BlobCopySourceTags, BlobDeleteType, BlobType, BlockListType, CopyStatus, DeleteSnapshotsOptionType, EncryptionAlgorithmType, FileShareTokenIntent, FilterBlobsIncludeItem, GeoReplicationStatusType, ImmutabilityPolicyMode, - LeaseDuration, LeaseState, LeaseStatus, ListBlobsIncludeItem, ListContainersIncludeType, + LeaseDuration, LeaseState, LeaseStatus, ListBlobsHierarchicalInternalResponseContentType, + ListBlobsIncludeItem, ListBlobsInternalResponseContentType, ListContainersIncludeType, PremiumPageBlobAccessTier, PublicAccessType, RehydratePriority, SequenceNumberActionType, SkuName, StorageErrorCode, }; @@ -706,6 +707,47 @@ impl Display for LeaseStatus { } } +impl FromStr for ListBlobsHierarchicalInternalResponseContentType { + type Err = Error; + fn from_str(s: &str) -> ::core::result::Result::Err> { + Ok(match s { + "application/vnd.apache.arrow.stream" => { + ListBlobsHierarchicalInternalResponseContentType::ApplicationVndApacheArrowStream + } + "application/xml" => ListBlobsHierarchicalInternalResponseContentType::ApplicationXml, + _ => { + return Err(Error::with_message_fn(ErrorKind::DataConversion, || { + format!("unknown variant of ListBlobsHierarchicalInternalResponseContentType found: \"{s}\"") + })) + } + }) + } +} + +impl AsRef for ListBlobsHierarchicalInternalResponseContentType { + fn as_ref(&self) -> &str { + match self { + ListBlobsHierarchicalInternalResponseContentType::ApplicationVndApacheArrowStream => { + "application/vnd.apache.arrow.stream" + } + ListBlobsHierarchicalInternalResponseContentType::ApplicationXml => "application/xml", + } + } +} + +impl Display for ListBlobsHierarchicalInternalResponseContentType { + fn fmt(&self, f: &mut Formatter<'_>) -> ::std::fmt::Result { + match self { + ListBlobsHierarchicalInternalResponseContentType::ApplicationVndApacheArrowStream => { + Display::fmt("application/vnd.apache.arrow.stream", f) + } + ListBlobsHierarchicalInternalResponseContentType::ApplicationXml => { + Display::fmt("application/xml", f) + } + } + } +} + impl FromStr for ListBlobsIncludeItem { type Err = Error; fn from_str(s: &str) -> ::core::result::Result::Err> { @@ -763,6 +805,49 @@ impl Display for ListBlobsIncludeItem { } } +impl FromStr for ListBlobsInternalResponseContentType { + type Err = Error; + fn from_str(s: &str) -> ::core::result::Result::Err> { + Ok(match s { + "application/vnd.apache.arrow.stream" => { + ListBlobsInternalResponseContentType::ApplicationVndApacheArrowStream + } + "application/xml" => ListBlobsInternalResponseContentType::ApplicationXml, + _ => { + return Err(Error::with_message_fn(ErrorKind::DataConversion, || { + format!( + "unknown variant of ListBlobsInternalResponseContentType found: \"{s}\"" + ) + })) + } + }) + } +} + +impl AsRef for ListBlobsInternalResponseContentType { + fn as_ref(&self) -> &str { + match self { + ListBlobsInternalResponseContentType::ApplicationVndApacheArrowStream => { + "application/vnd.apache.arrow.stream" + } + ListBlobsInternalResponseContentType::ApplicationXml => "application/xml", + } + } +} + +impl Display for ListBlobsInternalResponseContentType { + fn fmt(&self, f: &mut Formatter<'_>) -> ::std::fmt::Result { + match self { + ListBlobsInternalResponseContentType::ApplicationVndApacheArrowStream => { + Display::fmt("application/vnd.apache.arrow.stream", f) + } + ListBlobsInternalResponseContentType::ApplicationXml => { + Display::fmt("application/xml", f) + } + } + } +} + impl FromStr for ListContainersIncludeType { type Err = Error; fn from_str(s: &str) -> ::core::result::Result::Err> { diff --git a/sdk/storage/azure_storage_blob/src/generated/models/enums_serde.rs b/sdk/storage/azure_storage_blob/src/generated/models/enums_serde.rs index 1929babe91c..178637860ca 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/enums_serde.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/enums_serde.rs @@ -7,7 +7,8 @@ use super::{ AccessTier, AccountKind, ArchiveStatus, BlobCopySourceTags, BlobDeleteType, BlobType, BlockListType, CopyStatus, DeleteSnapshotsOptionType, EncryptionAlgorithmType, FileShareTokenIntent, FilterBlobsIncludeItem, GeoReplicationStatusType, ImmutabilityPolicyMode, - LeaseDuration, LeaseState, LeaseStatus, ListBlobsIncludeItem, ListContainersIncludeType, + LeaseDuration, LeaseState, LeaseStatus, ListBlobsHierarchicalInternalResponseContentType, + ListBlobsIncludeItem, ListBlobsInternalResponseContentType, ListContainersIncludeType, PremiumPageBlobAccessTier, PublicAccessType, RehydratePriority, SequenceNumberActionType, SkuName, StorageErrorCode, }; @@ -336,6 +337,25 @@ impl Serialize for LeaseStatus { } } +impl<'de> Deserialize<'de> for ListBlobsHierarchicalInternalResponseContentType { + fn deserialize(deserializer: D) -> ::core::result::Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + s.parse().map_err(serde::de::Error::custom) + } +} + +impl Serialize for ListBlobsHierarchicalInternalResponseContentType { + fn serialize(&self, s: S) -> ::core::result::Result + where + S: Serializer, + { + s.serialize_str(self.as_ref()) + } +} + impl<'de> Deserialize<'de> for ListBlobsIncludeItem { fn deserialize(deserializer: D) -> ::core::result::Result where @@ -355,6 +375,25 @@ impl Serialize for ListBlobsIncludeItem { } } +impl<'de> Deserialize<'de> for ListBlobsInternalResponseContentType { + fn deserialize(deserializer: D) -> ::core::result::Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + s.parse().map_err(serde::de::Error::custom) + } +} + +impl Serialize for ListBlobsInternalResponseContentType { + fn serialize(&self, s: S) -> ::core::result::Result + where + S: Serializer, + { + s.serialize_str(self.as_ref()) + } +} + impl<'de> Deserialize<'de> for ListContainersIncludeType { fn deserialize(deserializer: D) -> ::core::result::Result where diff --git a/sdk/storage/azure_storage_blob/src/generated/models/header_traits.rs b/sdk/storage/azure_storage_blob/src/generated/models/header_traits.rs index 2251d2837f6..fe03cc53326 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/header_traits.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/header_traits.rs @@ -12,12 +12,14 @@ use super::{ BlobClientRenewLeaseResult, BlobClientStartCopyFromUrlResult, BlobContainerClientAcquireLeaseResult, BlobContainerClientBreakLeaseResult, BlobContainerClientChangeLeaseResult, BlobContainerClientGetAccountInfoResult, - BlobContainerClientGetPropertiesResult, BlobContainerClientReleaseLeaseResult, + BlobContainerClientGetPropertiesResult, BlobContainerClientListBlobsHierarchicalInternalResult, + BlobContainerClientListBlobsInternalResult, BlobContainerClientReleaseLeaseResult, BlobContainerClientRenewLeaseResult, BlobServiceClientGetAccountInfoResult, BlobType, BlockBlobClientCommitBlockListResult, BlockBlobClientStageBlockFromUrlResult, BlockBlobClientStageBlockResult, BlockBlobClientUploadBlobFromUrlResult, BlockBlobClientUploadInternalResult, BlockList, CopyStatus, ImmutabilityPolicyMode, - LeaseDuration, LeaseState, LeaseStatus, PageBlobClientClearPagesResult, + LeaseDuration, LeaseState, LeaseStatus, ListBlobsHierarchicalInternalResponseContentType, + ListBlobsInternalResponseContentType, PageBlobClientClearPagesResult, PageBlobClientCreateResult, PageBlobClientResizeResult, PageBlobClientSetSequenceNumberResult, PageBlobClientUploadPagesFromUrlResult, PageBlobClientUploadPagesResult, PageList, PublicAccessType, RehydratePriority, SignedIdentifiers, SkuName, @@ -1725,6 +1727,66 @@ impl BlobContainerClientGetPropertiesResultHeaders } } +/// Provides access to typed response headers for `BlobContainerClient::list_blobs_hierarchical_internal()` +/// +/// # Examples +/// +/// ```ignore +/// use azure_core::{Result, http::AsyncResponse}; +/// use azure_storage_blob::models::{BlobContainerClientListBlobsHierarchicalInternalResult, BlobContainerClientListBlobsHierarchicalInternalResultHeaders}; +/// async fn example() -> Result<()> { +/// let response: AsyncResponse = unimplemented!(); +/// // Access response headers +/// if let Some(content_type) = response.content_type()? { +/// println!("content-type: {:?}", content_type); +/// } +/// Ok(()) +/// } +/// ``` +pub(crate) trait BlobContainerClientListBlobsHierarchicalInternalResultHeaders: + private::Sealed +{ + fn content_type(&self) -> Result>; +} + +impl BlobContainerClientListBlobsHierarchicalInternalResultHeaders + for AsyncResponse +{ + /// Content-Type header + fn content_type(&self) -> Result> { + Headers::get_optional_as(self.headers(), &CONTENT_TYPE) + } +} + +/// Provides access to typed response headers for `BlobContainerClient::list_blobs_internal()` +/// +/// # Examples +/// +/// ```ignore +/// use azure_core::{Result, http::AsyncResponse}; +/// use azure_storage_blob::models::{BlobContainerClientListBlobsInternalResult, BlobContainerClientListBlobsInternalResultHeaders}; +/// async fn example() -> Result<()> { +/// let response: AsyncResponse = unimplemented!(); +/// // Access response headers +/// if let Some(content_type) = response.content_type()? { +/// println!("content-type: {:?}", content_type); +/// } +/// Ok(()) +/// } +/// ``` +pub(crate) trait BlobContainerClientListBlobsInternalResultHeaders: private::Sealed { + fn content_type(&self) -> Result>; +} + +impl BlobContainerClientListBlobsInternalResultHeaders + for AsyncResponse +{ + /// Content-Type header + fn content_type(&self) -> Result> { + Headers::get_optional_as(self.headers(), &CONTENT_TYPE) + } +} + /// Provides access to typed response headers for `BlobContainerClient::release_lease()` /// /// # Examples @@ -2767,11 +2829,13 @@ mod private { BlobClientStartCopyFromUrlResult, BlobContainerClientAcquireLeaseResult, BlobContainerClientBreakLeaseResult, BlobContainerClientChangeLeaseResult, BlobContainerClientGetAccountInfoResult, BlobContainerClientGetPropertiesResult, - BlobContainerClientReleaseLeaseResult, BlobContainerClientRenewLeaseResult, - BlobServiceClientGetAccountInfoResult, BlockBlobClientCommitBlockListResult, - BlockBlobClientStageBlockFromUrlResult, BlockBlobClientStageBlockResult, - BlockBlobClientUploadBlobFromUrlResult, BlockBlobClientUploadInternalResult, BlockList, - PageBlobClientClearPagesResult, PageBlobClientCreateResult, PageBlobClientResizeResult, + BlobContainerClientListBlobsHierarchicalInternalResult, + BlobContainerClientListBlobsInternalResult, BlobContainerClientReleaseLeaseResult, + BlobContainerClientRenewLeaseResult, BlobServiceClientGetAccountInfoResult, + BlockBlobClientCommitBlockListResult, BlockBlobClientStageBlockFromUrlResult, + BlockBlobClientStageBlockResult, BlockBlobClientUploadBlobFromUrlResult, + BlockBlobClientUploadInternalResult, BlockList, PageBlobClientClearPagesResult, + PageBlobClientCreateResult, PageBlobClientResizeResult, PageBlobClientSetSequenceNumberResult, PageBlobClientUploadPagesFromUrlResult, PageBlobClientUploadPagesResult, PageList, SignedIdentifiers, }; @@ -2780,6 +2844,8 @@ mod private { pub trait Sealed {} impl Sealed for AsyncResponse {} + impl Sealed for AsyncResponse {} + impl Sealed for AsyncResponse {} impl Sealed for Response {} impl Sealed for Response {} impl Sealed for Response {} diff --git a/sdk/storage/azure_storage_blob/src/generated/models/method_options.rs b/sdk/storage/azure_storage_blob/src/generated/models/method_options.rs index 3e3bcae3b64..c8198849b80 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/method_options.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/method_options.rs @@ -1088,37 +1088,142 @@ pub struct BlobContainerClientGetPropertiesOptions<'a> { pub timeout: Option, } -/// Options to be passed to `BlobContainerClient::list_blobs()` +/// Options to be passed to `BlobContainerClient::list_blobs_hierarchical_internal()` #[derive(Clone, Default, SafeDebug)] -pub struct BlobContainerClientListBlobsOptions<'a> { +pub(crate) struct BlobContainerClientListBlobsHierarchicalInternalOptions<'a> { + /// Filters the results to return only names that are ordered before this value. Currently only applies to Apache Arrow scenario. + pub(crate) end_before: Option, + /// Specify to include additional, optional information. - pub include: Option>, + pub(crate) include: Option>, /// An opaque string value that identifies the portion of the result set to return with this operation. - pub marker: Option, + pub(crate) marker: Option, /// Specifies the maximum number of resources to return. If the request does not specify maxresults, or specifies a value /// greater than 5000, the server will return up to 5000 items. - pub maxresults: Option, + pub(crate) maxresults: Option, /// Allows customization of the method call. - pub method_options: PagerOptions<'a>, + pub(crate) method_options: ClientMethodOptions<'a>, /// Filters the results to return only resources whose name begins with the specified prefix. - pub prefix: Option, + pub(crate) prefix: Option, /// Specifies the relative path to list paths from. For non-recursive list, only one entity level is supported; for recursive /// list, multiple entity levels are supported. (Inclusive) - pub start_from: Option, + pub(crate) start_from: Option, /// The timeout parameter is expressed in seconds. For more information, see [Setting Timeouts for Blob Service Operations.](\"") - pub timeout: Option, + pub(crate) timeout: Option, +} + +/// Options to be passed to `BlobContainerClient::list_blobs_hierarchical_xml()` +#[derive(Clone, Default, SafeDebug)] +pub(crate) struct BlobContainerClientListBlobsHierarchicalXmlOptions<'a> { + /// Specify to include additional, optional information. + pub(crate) include: Option>, + + /// An opaque string value that identifies the portion of the result set to return with this operation. + pub(crate) marker: Option, + + /// Specifies the maximum number of resources to return. If the request does not specify maxresults, or specifies a value + /// greater than 5000, the server will return up to 5000 items. + pub(crate) maxresults: Option, + + /// Allows customization of the method call. + pub(crate) method_options: PagerOptions<'a>, + + /// Filters the results to return only resources whose name begins with the specified prefix. + pub(crate) prefix: Option, + + /// Specifies the relative path to list paths from. For non-recursive list, only one entity level is supported; for recursive + /// list, multiple entity levels are supported. (Inclusive) + pub(crate) start_from: Option, + + /// The timeout parameter is expressed in seconds. For more information, see [Setting Timeouts for Blob Service Operations.](\"") + pub(crate) timeout: Option, +} + +impl BlobContainerClientListBlobsHierarchicalXmlOptions<'_> { + /// Transforms this [`BlobContainerClientListBlobsHierarchicalXmlOptions`] into a new `BlobContainerClientListBlobsHierarchicalXmlOptions` that owns the underlying data, cloning it if necessary. + pub fn into_owned(self) -> BlobContainerClientListBlobsHierarchicalXmlOptions<'static> { + BlobContainerClientListBlobsHierarchicalXmlOptions { + include: self.include, + marker: self.marker, + maxresults: self.maxresults, + method_options: PagerOptions { + context: self.method_options.context.into_owned(), + ..self.method_options + }, + prefix: self.prefix, + start_from: self.start_from, + timeout: self.timeout, + } + } +} + +/// Options to be passed to `BlobContainerClient::list_blobs_internal()` +#[derive(Clone, Default, SafeDebug)] +pub(crate) struct BlobContainerClientListBlobsInternalOptions<'a> { + /// Filters the results to return only names that are ordered before this value. Currently only applies to Apache Arrow scenario. + pub(crate) end_before: Option, + + /// Specify to include additional, optional information. + pub(crate) include: Option>, + + /// An opaque string value that identifies the portion of the result set to return with this operation. + pub(crate) marker: Option, + + /// Specifies the maximum number of resources to return. If the request does not specify maxresults, or specifies a value + /// greater than 5000, the server will return up to 5000 items. + pub(crate) maxresults: Option, + + /// Allows customization of the method call. + pub(crate) method_options: ClientMethodOptions<'a>, + + /// Filters the results to return only resources whose name begins with the specified prefix. + pub(crate) prefix: Option, + + /// Specifies the relative path to list paths from. For non-recursive list, only one entity level is supported; for recursive + /// list, multiple entity levels are supported. (Inclusive) + pub(crate) start_from: Option, + + /// The timeout parameter is expressed in seconds. For more information, see [Setting Timeouts for Blob Service Operations.](\"") + pub(crate) timeout: Option, +} + +/// Options to be passed to `BlobContainerClient::list_blobs_xml()` +#[derive(Clone, Default, SafeDebug)] +pub(crate) struct BlobContainerClientListBlobsXmlOptions<'a> { + /// Specify to include additional, optional information. + pub(crate) include: Option>, + + /// An opaque string value that identifies the portion of the result set to return with this operation. + pub(crate) marker: Option, + + /// Specifies the maximum number of resources to return. If the request does not specify maxresults, or specifies a value + /// greater than 5000, the server will return up to 5000 items. + pub(crate) maxresults: Option, + + /// Allows customization of the method call. + pub(crate) method_options: PagerOptions<'a>, + + /// Filters the results to return only resources whose name begins with the specified prefix. + pub(crate) prefix: Option, + + /// Specifies the relative path to list paths from. For non-recursive list, only one entity level is supported; for recursive + /// list, multiple entity levels are supported. (Inclusive) + pub(crate) start_from: Option, + + /// The timeout parameter is expressed in seconds. For more information, see [Setting Timeouts for Blob Service Operations.](\"") + pub(crate) timeout: Option, } -impl BlobContainerClientListBlobsOptions<'_> { - /// Transforms this [`BlobContainerClientListBlobsOptions`] into a new `BlobContainerClientListBlobsOptions` that owns the underlying data, cloning it if necessary. - pub fn into_owned(self) -> BlobContainerClientListBlobsOptions<'static> { - BlobContainerClientListBlobsOptions { +impl BlobContainerClientListBlobsXmlOptions<'_> { + /// Transforms this [`BlobContainerClientListBlobsXmlOptions`] into a new `BlobContainerClientListBlobsXmlOptions` that owns the underlying data, cloning it if necessary. + pub fn into_owned(self) -> BlobContainerClientListBlobsXmlOptions<'static> { + BlobContainerClientListBlobsXmlOptions { include: self.include, marker: self.marker, maxresults: self.maxresults, diff --git a/sdk/storage/azure_storage_blob/src/generated/models/models.rs b/sdk/storage/azure_storage_blob/src/generated/models/models.rs index d7089c2be0d..d3cb48461fc 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/models.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/models.rs @@ -120,6 +120,14 @@ pub struct BlobContainerClientGetAccountInfoResult; #[derive(SafeDebug)] pub struct BlobContainerClientGetPropertiesResult; +/// Contains results for `BlobContainerClient::list_blobs_hierarchical_internal()` +#[derive(SafeDebug)] +pub(crate) struct BlobContainerClientListBlobsHierarchicalInternalResult; + +/// Contains results for `BlobContainerClient::list_blobs_internal()` +#[derive(SafeDebug)] +pub(crate) struct BlobContainerClientListBlobsInternalResult; + /// Contains results for `BlobContainerClient::release_lease()` #[derive(SafeDebug)] pub struct BlobContainerClientReleaseLeaseResult; @@ -128,6 +136,19 @@ pub struct BlobContainerClientReleaseLeaseResult; #[derive(SafeDebug)] pub struct BlobContainerClientRenewLeaseResult; +/// Represents an array of blobs. +#[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] +#[non_exhaustive] +pub struct BlobHierarchyList { + /// The blob items. + #[serde(default, rename = "Blob")] + pub blob_items: Vec, + + /// The blob prefixes. + #[serde(rename = "BlobPrefix", skip_serializing_if = "Option::is_none")] + pub blob_prefixes: Option>, +} + /// Represents a blob. #[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] #[non_exhaustive] @@ -191,6 +212,19 @@ pub struct BlobName { pub encoded: Option, } +/// Represents a blob prefix. +#[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] +#[non_exhaustive] +pub struct BlobPrefix { + /// The blob name. + #[serde( + deserialize_with = "crate::models::blob_name::option::deserialize", + rename = "Name", + skip_serializing_if = "Option::is_none" + )] + pub name: Option, +} + /// The properties of a blob. #[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] #[non_exhaustive] @@ -901,6 +935,45 @@ pub struct KeyInfo { pub start: Option, } +/// The result of the List Blobs Hierarchical API. +#[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] +#[non_exhaustive] +#[serde(rename = "EnumerationResults")] +pub struct ListBlobsHierarchicalResponse { + /// The container name. + #[serde(rename = "@ContainerName", skip_serializing_if = "Option::is_none")] + pub container_name: Option, + + /// The delimiter of the blobs. + #[serde(rename = "Delimiter", skip_serializing_if = "Option::is_none")] + pub delimiter: Option, + + /// The list of hierarchical blobs. + #[serde(default, rename = "Blobs")] + pub hierarchical_list: BlobHierarchyList, + + /// An opaque string value that identifies the portion of the result set returned with this operation. + #[serde(rename = "Marker", skip_serializing_if = "Option::is_none")] + pub marker: Option, + + /// The maximum number of blobs to be returned with this operation. + #[serde(rename = "MaxResults", skip_serializing_if = "Option::is_none")] + pub max_results: Option, + + /// An opaque string value that identifies the portion of the result set to be returned with the next operation. Use this + /// value in the next request to continue the listing operation. + #[serde(rename = "NextMarker", skip_serializing_if = "Option::is_none")] + pub next_marker: Option, + + /// The prefix of the blobs. + #[serde(rename = "Prefix", skip_serializing_if = "Option::is_none")] + pub prefix: Option, + + /// The service endpoint. + #[serde(rename = "@ServiceEndpoint", skip_serializing_if = "Option::is_none")] + pub service_endpoint: Option, +} + /// The result of the List Blobs API. #[derive(Clone, Default, Deserialize, SafeDebug, Serialize)] #[non_exhaustive] diff --git a/sdk/storage/azure_storage_blob/src/generated/models/models_impl.rs b/sdk/storage/azure_storage_blob/src/generated/models/models_impl.rs index 8912359fcfa..765e0abf336 100644 --- a/sdk/storage/azure_storage_blob/src/generated/models/models_impl.rs +++ b/sdk/storage/azure_storage_blob/src/generated/models/models_impl.rs @@ -5,7 +5,8 @@ use super::{ BlobItem, BlobServiceProperties, BlobTags, BlockLookupList, ContainerItem, FilterBlobItem, - FilteredBlobResponse, KeyInfo, ListBlobsResponse, ListContainersResponse, SignedIdentifiers, + FilteredBlobResponse, KeyInfo, ListBlobsHierarchicalResponse, ListBlobsResponse, + ListContainersResponse, SignedIdentifiers, }; use async_trait::async_trait; use azure_core::{ @@ -23,6 +24,15 @@ impl Page for FilteredBlobResponse { } } +#[async_trait] +impl Page for ListBlobsHierarchicalResponse { + type Item = BlobItem; + type IntoIter = as IntoIterator>::IntoIter; + async fn into_items(self) -> Result { + Ok(self.hierarchical_list.blob_items.into_iter()) + } +} + #[async_trait] impl Page for ListBlobsResponse { type Item = BlobItem; diff --git a/sdk/storage/azure_storage_blob/src/lib.rs b/sdk/storage/azure_storage_blob/src/lib.rs index 38aa86eb831..8fb03e5ff68 100644 --- a/sdk/storage/azure_storage_blob/src/lib.rs +++ b/sdk/storage/azure_storage_blob/src/lib.rs @@ -7,6 +7,7 @@ #![allow(dead_code)] #![cfg_attr(docsrs, feature(doc_cfg))] +mod arrow_decode; pub(crate) mod buffers; pub mod clients; #[allow(unused_imports)] diff --git a/sdk/storage/azure_storage_blob/src/models/method_options.rs b/sdk/storage/azure_storage_blob/src/models/method_options.rs index f89c9f1a232..4456ede25c4 100644 --- a/sdk/storage/azure_storage_blob/src/models/method_options.rs +++ b/sdk/storage/azure_storage_blob/src/models/method_options.rs @@ -5,13 +5,15 @@ use std::{collections::HashMap, num::NonZero}; use azure_core::{ fmt::SafeDebug, - http::{ClientMethodOptions, Etag}, + http::{pager::PagerOptions, ClientMethodOptions, Etag}, }; use time::OffsetDateTime; use crate::models::{ - AccessTier, BlobClientDownloadInternalOptions, EncryptionAlgorithmType, HttpRange, - ImmutabilityPolicyMode, + AccessTier, BlobClientDownloadInternalOptions, + BlobContainerClientListBlobsHierarchicalInternalOptions, + BlobContainerClientListBlobsInternalOptions, EncryptionAlgorithmType, HttpRange, + ImmutabilityPolicyMode, ListBlobsIncludeItem, }; /// Options to be passed to `BlobClient::download()` @@ -109,6 +111,158 @@ impl<'a> From> for BlobClientDownloadInternalOptio } } +/// The response format requested by [`BlobContainerClient::list_blobs`](crate::BlobContainerClient::list_blobs). +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum StorageResponseFormat { + /// Prefer Apache Arrow and allow the service to fall back to XML. + #[default] + Arrow, + + /// Request XML only. + Xml, +} + +impl StorageResponseFormat { + pub(crate) fn as_header_value(self) -> &'static str { + match self { + Self::Arrow => "application/vnd.apache.arrow.stream,application/xml", + Self::Xml => "application/xml", + } + } +} + +/// Options to be passed to [`BlobContainerClient::list_blobs`](crate::BlobContainerClient::list_blobs). +#[derive(Clone, Default, SafeDebug)] +pub struct BlobContainerClientListBlobsOptions<'a> { + /// Selects the response format. Defaults to [`StorageResponseFormat::Arrow`]. + pub response_format: Option, + + /// Filters the results to return only names that are ordered before this value. Only applies to the Apache Arrow scenario. + pub end_before: Option, + + /// Specify to include additional, optional information. + pub include: Option>, + + /// An opaque string value that identifies the portion of the result set to return with this operation. + pub marker: Option, + + /// Specifies the maximum number of resources to return. + pub maxresults: Option, + + /// Allows customization of the method call. + pub method_options: PagerOptions<'a>, + + /// Filters the results to return only resources whose name begins with the specified prefix. + pub prefix: Option, + + /// Specifies the relative path to list paths from. + pub start_from: Option, + + /// The timeout parameter is expressed in seconds. + pub timeout: Option, +} + +impl BlobContainerClientListBlobsOptions<'_> { + pub(crate) fn into_owned(self) -> BlobContainerClientListBlobsOptions<'static> { + BlobContainerClientListBlobsOptions { + response_format: self.response_format, + end_before: self.end_before, + include: self.include, + marker: self.marker, + maxresults: self.maxresults, + method_options: PagerOptions { + context: self.method_options.context.into_owned(), + ..self.method_options + }, + prefix: self.prefix, + start_from: self.start_from, + timeout: self.timeout, + } + } + + pub(crate) fn to_internal( + &self, + method_options: ClientMethodOptions<'static>, + ) -> BlobContainerClientListBlobsInternalOptions<'static> { + BlobContainerClientListBlobsInternalOptions { + end_before: self.end_before.clone(), + include: self.include.clone(), + marker: self.marker.clone(), + maxresults: self.maxresults, + method_options, + prefix: self.prefix.clone(), + start_from: self.start_from.clone(), + timeout: self.timeout, + } + } +} + +/// Options to be passed to [`BlobContainerClient::list_blobs_hierarchical`](crate::BlobContainerClient::list_blobs_hierarchical). +#[derive(Clone, Default, SafeDebug)] +pub struct BlobContainerClientListBlobsHierarchicalOptions<'a> { + /// Selects the response format. Defaults to [`StorageResponseFormat::Arrow`]. + pub response_format: Option, + + /// Filters the results to return only names that are ordered before this value. Only applies to the Apache Arrow scenario. + pub end_before: Option, + + /// Specify to include additional, optional information. + pub include: Option>, + + /// An opaque string value that identifies the portion of the result set to return with this operation. + pub marker: Option, + + /// Specifies the maximum number of resources to return. + pub maxresults: Option, + + /// Allows customization of the method call. + pub method_options: PagerOptions<'a>, + + /// Filters the results to return only resources whose name begins with the specified prefix. + pub prefix: Option, + + /// Specifies the relative path to list paths from. + pub start_from: Option, + + /// The timeout parameter is expressed in seconds. + pub timeout: Option, +} + +impl BlobContainerClientListBlobsHierarchicalOptions<'_> { + pub(crate) fn into_owned(self) -> BlobContainerClientListBlobsHierarchicalOptions<'static> { + BlobContainerClientListBlobsHierarchicalOptions { + response_format: self.response_format, + end_before: self.end_before, + include: self.include, + marker: self.marker, + maxresults: self.maxresults, + method_options: PagerOptions { + context: self.method_options.context.into_owned(), + ..self.method_options + }, + prefix: self.prefix, + start_from: self.start_from, + timeout: self.timeout, + } + } + + pub(crate) fn to_internal( + &self, + method_options: ClientMethodOptions<'static>, + ) -> BlobContainerClientListBlobsHierarchicalInternalOptions<'static> { + BlobContainerClientListBlobsHierarchicalInternalOptions { + end_before: self.end_before.clone(), + include: self.include.clone(), + marker: self.marker.clone(), + maxresults: self.maxresults, + method_options, + prefix: self.prefix.clone(), + start_from: self.start_from.clone(), + timeout: self.timeout, + } + } +} + /// Options to be passed to `BlockBlobClient::upload()` #[derive(Clone, Default, SafeDebug)] pub struct BlockBlobClientUploadOptions<'a> { diff --git a/sdk/storage/azure_storage_blob/src/models/mod.rs b/sdk/storage/azure_storage_blob/src/models/mod.rs index af94a4a4da3..26bd4b56a1f 100644 --- a/sdk/storage/azure_storage_blob/src/models/mod.rs +++ b/sdk/storage/azure_storage_blob/src/models/mod.rs @@ -14,6 +14,7 @@ pub use http_ranges::HttpRange; pub(crate) mod response_ext; mod upload_result; +pub use crate::arrow_decode::AutoFormat; pub use crate::generated::models::*; pub use download_result::{ BlobClientDownloadIntoResult, BlobClientDownloadResult, BlobDownloadProperties, @@ -21,6 +22,10 @@ pub use download_result::{ pub use method_options::BlobClientDownloadOptions; pub use method_options::BlockBlobClientUploadOptions; pub use method_options::BlockBlobClientUploadOptions as BlobClientUploadOptions; +pub use method_options::{ + BlobContainerClientListBlobsHierarchicalOptions, BlobContainerClientListBlobsOptions, + StorageResponseFormat, +}; pub use upload_result::BlockBlobClientUploadResult; pub use upload_result::BlockBlobClientUploadResult as BlobClientUploadResult; diff --git a/sdk/storage/azure_storage_blob/tests/blob_container_client.rs b/sdk/storage/azure_storage_blob/tests/blob_container_client.rs index 86d108d8f7a..27c20df82de 100644 --- a/sdk/storage/azure_storage_blob/tests/blob_container_client.rs +++ b/sdk/storage/azure_storage_blob/tests/blob_container_client.rs @@ -7,21 +7,24 @@ use azure_core::{ http::{RequestContent, StatusCode}, time::{parse_rfc3339, to_rfc3339, OffsetDateTime}, }; -use azure_core_test::{recorded, Matcher, TestContext, VarOptions}; +use azure_core_test::{recorded, Matcher, TestContext, TestMode, VarOptions}; use azure_storage_blob::format_filter_expression; use azure_storage_blob::models::{ - AccessPolicy, AccountKind, BlobContainerClientAcquireLeaseResultHeaders, - BlobContainerClientBreakLeaseOptions, BlobContainerClientChangeLeaseResultHeaders, - BlobContainerClientCreateOptions, BlobContainerClientFindBlobsByTagsOptions, - BlobContainerClientGetAccountInfoResultHeaders, BlobContainerClientGetPropertiesResultHeaders, + AccessPolicy, AccessTier, AccountKind, ArchiveStatus, BlobClientGetPropertiesResultHeaders, + BlobClientSetImmutabilityPolicyOptions, BlobClientSetTierOptions, + BlobContainerClientAcquireLeaseResultHeaders, BlobContainerClientBreakLeaseOptions, + BlobContainerClientChangeLeaseResultHeaders, BlobContainerClientCreateOptions, + BlobContainerClientFindBlobsByTagsOptions, BlobContainerClientGetAccountInfoResultHeaders, + BlobContainerClientGetPropertiesResultHeaders, BlobContainerClientListBlobsHierarchicalOptions, BlobContainerClientListBlobsOptions, BlobContainerClientSetMetadataOptions, BlobType, - BlockBlobClientUploadOptions, LeaseState, ListBlobsIncludeItem, SignedIdentifiers, - StorageErrorCode, + BlockBlobClientUploadOptions, CopyStatus, ImmutabilityPolicyMode, LeaseDuration, LeaseState, + LeaseStatus, ListBlobsIncludeItem, PageBlobClientSetSequenceNumberOptions, RehydratePriority, + SequenceNumberActionType, SignedIdentifiers, StorageErrorCode, StorageResponseFormat, }; use azure_storage_blob::StorageError; use common::{ create_test_blob, get_blob_name, get_blob_service_client, get_container_client, - get_container_name, poll_until, StorageAccount, + get_container_name, get_valid_encryption_scope, list_blobs_arrow, poll_until, StorageAccount, }; use futures::{StreamExt, TryStreamExt}; use std::{collections::HashMap, error::Error, time::Duration}; @@ -142,6 +145,640 @@ async fn test_list_blobs(ctx: TestContext) -> Result<(), Box> { Ok(()) } +#[recorded::test] +async fn test_list_blobs_arrow_populates_properties( + ctx: TestContext, +) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, false, StorageAccount::Standard, None).await?; + container_client.create(None).await?; + + // Arrange: upload a blob populating as many listable properties as possible. + let blob_name = get_blob_name(recording); + let payload = b"arrow phase a payload".to_vec(); + // Base64 MD5 of `payload`; the service validates x-ms-blob-content-md5 against the body. + let content_md5 = azure_core::base64::decode("IU+6Y1iDGdD2YaCH1kdRpg==")?; + let metadata = HashMap::from([("team".to_string(), "sdk".to_string())]); + let upload_options = BlockBlobClientUploadOptions { + blob_cache_control: Some("max-age=3600".to_string()), + blob_content_disposition: Some("inline".to_string()), + blob_content_encoding: Some("gzip".to_string()), + blob_content_language: Some("en-US".to_string()), + blob_content_md5: Some(content_md5.clone()), + blob_content_type: Some("text/plain".to_string()), + metadata: Some(metadata.clone()), + tier: Some(AccessTier::Hot), + ..Default::default() + } + .with_tags(HashMap::from([("env".to_string(), "test".to_string())])); + create_test_blob( + &container_client.blob_client(&blob_name), + Some(RequestContent::from(payload.clone())), + Some(upload_options), + ) + .await?; + + // Act: request the Apache Arrow stream. The SDK transparently decodes Arrow or + // falls back to XML, so this exercises the field mapping on whichever wire + // format the live service returns. + let page = container_client + .list_blobs(Some(BlobContainerClientListBlobsOptions { + response_format: Some(StorageResponseFormat::Arrow), + include: Some(vec![ + ListBlobsIncludeItem::Metadata, + ListBlobsIncludeItem::Tags, + ]), + ..Default::default() + }))? + .into_pages() + .try_next() + .await? + .unwrap() + .into_model()?; + + // Assert: the scalar/timestamp/enum properties round-trip through the mapping. + let blob = page + .blob_items + .iter() + .find(|b| b.name.as_deref() == Some(blob_name.as_str())) + .expect("expected uploaded blob in listing"); + let props = blob.properties.as_ref().expect("expected blob properties"); + + assert_eq!(Some(BlobType::BlockBlob), props.blob_type); + assert!(props.etag.is_some()); + assert_eq!(Some(payload.len() as u64), props.content_length); + assert_eq!(Some("text/plain".to_string()), props.content_type); + assert_eq!(Some("gzip".to_string()), props.content_encoding); + assert_eq!(Some("en-US".to_string()), props.content_language); + assert_eq!(Some("inline".to_string()), props.content_disposition); + assert_eq!(Some("max-age=3600".to_string()), props.cache_control); + assert_eq!(Some(content_md5), props.content_md5); + assert!(props.creation_time.is_some()); + assert!(props.last_modified.is_some()); + assert!(props.access_tier.is_some()); + assert!(props.access_tier_change_time.is_some()); + assert_eq!(Some(LeaseState::Available), props.lease_state); + assert_eq!(Some(LeaseStatus::Unlocked), props.lease_status); + assert_eq!(Some(true), props.server_encrypted); + assert_eq!(Some(1), props.tag_count); + + // Map-typed columns decode from the Arrow `map` columns. + let blob_meta = blob + .metadata + .as_ref() + .expect("metadata should be populated"); + assert_eq!(Some(&metadata), blob_meta.values.as_ref()); + let tags = blob + .blob_tags + .as_ref() + .expect("blob_tags should be populated") + .blob_tag_set + .as_ref() + .expect("tag set should be present"); + assert!(tags + .iter() + .any(|t| t.key.as_deref() == Some("env") && t.value.as_deref() == Some("test"))); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_hierarchical_arrow(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, false, StorageAccount::Standard, None).await?; + container_client.create(None).await?; + + // Arrange: two blobs under a virtual directory plus one at the container root. + for name in ["dir1/a.txt", "dir1/b.txt", "top.txt"] { + create_test_blob(&container_client.blob_client(name), None, None).await?; + } + + // Act: list hierarchically over Apache Arrow, grouping the directory with "/". + let page = container_client + .list_blobs_hierarchical( + "/", + Some(BlobContainerClientListBlobsHierarchicalOptions { + response_format: Some(StorageResponseFormat::Arrow), + ..Default::default() + }), + )? + .into_pages() + .try_next() + .await? + .expect("expected a page") + .into_model()?; + + // Assert: the directory collapses into a BlobPrefix and the root blob is listed. + let prefixes = page + .hierarchical_list + .blob_prefixes + .expect("expected blob prefixes"); + assert!(prefixes.iter().any(|p| p.name.as_deref() == Some("dir1/"))); + assert!(page + .hierarchical_list + .blob_items + .iter() + .any(|b| b.name.as_deref() == Some("top.txt"))); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_arrow_end_before(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, false, StorageAccount::Standard, None).await?; + container_client.create(None).await?; + + // Arrange: four lexicographically ordered blobs. + for name in ["aa.txt", "bb.txt", "cc.txt", "dd.txt"] { + create_test_blob(&container_client.blob_client(name), None, None).await?; + } + + // Act: Apache Arrow range listing stops before "cc.txt" (exclusive). + let page = container_client + .list_blobs(Some(BlobContainerClientListBlobsOptions { + response_format: Some(StorageResponseFormat::Arrow), + end_before: Some("cc.txt".to_string()), + ..Default::default() + }))? + .into_pages() + .try_next() + .await? + .expect("expected a page") + .into_model()?; + + // Assert: only names ordered before "cc.txt" are returned. + let names: Vec<_> = page + .blob_items + .iter() + .filter_map(|b| b.name.as_deref()) + .collect(); + assert!(names.contains(&"aa.txt")); + assert!(names.contains(&"bb.txt")); + assert!(!names.contains(&"cc.txt")); + assert!(!names.contains(&"dd.txt")); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_arrow_stateful_properties(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Standard, None).await?; + + // Blob with an infinite lease -> Leased / Locked / Infinite. + let leased_name = get_blob_name(recording); + let leased_client = container_client.blob_client(&leased_name); + create_test_blob(&leased_client, None, None).await?; + leased_client.acquire_lease(-1, None).await?; + + // Sealed append blob -> is_sealed, blob_type = AppendBlob. + let sealed_name = get_blob_name(recording); + let append_client = container_client + .blob_client(&sealed_name) + .append_blob_client(); + append_client.create(None).await?; + append_client.seal(None).await?; + + // Page blob with a sequence number -> blob_sequence_number, blob_type = PageBlob. + let page_name = get_blob_name(recording); + let page_client = container_client.blob_client(&page_name).page_blob_client(); + page_client.create(1024, None).await?; + page_client + .set_sequence_number( + SequenceNumberActionType::Update, + Some(PageBlobClientSetSequenceNumberOptions { + blob_sequence_number: Some(7), + ..Default::default() + }), + ) + .await?; + + // Blob with a snapshot -> snapshot row present with the Snapshots include. + let snapshot_name = get_blob_name(recording); + let snapshot_client = container_client.blob_client(&snapshot_name); + create_test_blob(&snapshot_client, None, None).await?; + snapshot_client.create_snapshot(None).await?; + + // Blob uploaded with an encryption scope -> encryption_scope. + let scope_name = get_blob_name(recording); + let scope_client = container_client.blob_client(&scope_name); + create_test_blob( + &scope_client, + None, + Some(BlockBlobClientUploadOptions { + encryption_scope: Some(get_valid_encryption_scope()), + ..Default::default() + }), + ) + .await?; + + // Archived blob rehydrating to Hot -> access_tier = Archive, archive_status, rehydrate_priority. + let archive_name = get_blob_name(recording); + let archive_client = container_client.blob_client(&archive_name); + create_test_blob(&archive_client, None, None).await?; + archive_client.set_tier(AccessTier::Archive, None).await?; + archive_client + .set_tier( + AccessTier::Hot, + Some(BlobClientSetTierOptions { + rehydrate_priority: Some(RehydratePriority::High), + ..Default::default() + }), + ) + .await?; + + // Blob uploaded without an explicit tier -> access_tier_inferred. + let inferred_name = get_blob_name(recording); + let inferred_client = container_client.blob_client(&inferred_name); + create_test_blob(&inferred_client, None, None).await?; + + // Soft-deleted blob -> deleted, deleted_time, remaining_retention_days with the Deleted include. + let deleted_name = get_blob_name(recording); + let deleted_client = container_client.blob_client(&deleted_name); + create_test_blob(&deleted_client, None, None).await?; + deleted_client.delete(None).await?; + + // A single Arrow list call covers every blob staged above. + let items = list_blobs_arrow( + &container_client, + Some(vec![ + ListBlobsIncludeItem::Snapshots, + ListBlobsIncludeItem::Deleted, + ]), + ) + .await?; + let find = |name: &str| { + items + .iter() + .find(|b| b.name.as_deref() == Some(name) && b.snapshot.is_none()) + .unwrap_or_else(|| panic!("expected blob {name} in listing")) + }; + + // Lease. + let props = find(&leased_name) + .properties + .as_ref() + .expect("leased properties"); + assert_eq!(Some(LeaseState::Leased), props.lease_state); + assert_eq!(Some(LeaseStatus::Locked), props.lease_status); + assert_eq!(Some(LeaseDuration::Infinite), props.lease_duration); + + // Sealed append blob. + let props = find(&sealed_name) + .properties + .as_ref() + .expect("sealed properties"); + assert_eq!(Some(true), props.is_sealed); + assert_eq!(Some(BlobType::AppendBlob), props.blob_type); + + // Page blob sequence number. + let props = find(&page_name) + .properties + .as_ref() + .expect("page properties"); + assert_eq!(Some(7), props.blob_sequence_number); + assert_eq!(Some(BlobType::PageBlob), props.blob_type); + + // Encryption scope. + let props = find(&scope_name) + .properties + .as_ref() + .expect("scope properties"); + assert_eq!(Some(get_valid_encryption_scope()), props.encryption_scope); + + // Archive + rehydrate. + let props = find(&archive_name) + .properties + .as_ref() + .expect("archive properties"); + assert_eq!(Some(AccessTier::Archive), props.access_tier); + assert_eq!( + Some(ArchiveStatus::RehydratePendingToHot), + props.archive_status + ); + assert_eq!(Some(RehydratePriority::High), props.rehydrate_priority); + + // Inferred tier. + let props = find(&inferred_name) + .properties + .as_ref() + .expect("inferred properties"); + assert_eq!(Some(true), props.access_tier_inferred); + + // Soft-deleted blob. + let deleted = find(&deleted_name); + assert_eq!(Some(true), deleted.deleted); + let props = deleted.properties.as_ref().expect("deleted properties"); + assert!(props.deleted_time.is_some()); + assert!(props.remaining_retention_days.is_some()); + + // Snapshot row (distinct from the base blob row). + assert!( + items + .iter() + .any(|b| b.name.as_deref() == Some(snapshot_name.as_str()) && b.snapshot.is_some()), + "expected a snapshot row for {snapshot_name}" + ); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_arrow_version_properties(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Versioned, None).await?; + + // Blob with two versions, current version retained -> version_id, is_current_version. + let versioned_name = get_blob_name(recording); + let versioned_client = container_client.blob_client(&versioned_name); + create_test_blob( + &versioned_client, + Some(RequestContent::from(b"version 1".to_vec())), + None, + ) + .await?; + create_test_blob( + &versioned_client, + Some(RequestContent::from(b"version 2".to_vec())), + None, + ) + .await?; + + let items = list_blobs_arrow( + &container_client, + Some(vec![ListBlobsIncludeItem::Versions]), + ) + .await?; + + // The retained blob exposes version_id on every row and exactly one current version. + let versions: Vec<_> = items + .iter() + .filter(|b| b.name.as_deref() == Some(versioned_name.as_str())) + .collect(); + assert!( + versions.len() >= 2, + "expected at least two versions for {versioned_name}, got {}", + versions.len() + ); + assert!( + versions.iter().all(|b| b.version_id.is_some()), + "every version row should carry a version_id" + ); + assert_eq!( + 1, + versions + .iter() + .filter(|b| b.is_current_version == Some(true)) + .count(), + "exactly one row should be the current version" + ); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_arrow_has_versions_only(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Versioned, None).await?; + + // Create two versions, then delete the base blob so only versions remain. Listing with + // DeletedWithVersions surfaces the blob as a root entry with has_versions_only set. + let name = get_blob_name(recording); + let blob_client = container_client.blob_client(&name); + create_test_blob( + &blob_client, + Some(RequestContent::from(b"v1".to_vec())), + None, + ) + .await?; + create_test_blob( + &blob_client, + Some(RequestContent::from(b"v2".to_vec())), + None, + ) + .await?; + blob_client.delete(None).await?; + + let items = list_blobs_arrow( + &container_client, + Some(vec![ListBlobsIncludeItem::DeletedWithVersions]), + ) + .await?; + + let blob = items + .iter() + .find(|b| b.name.as_deref() == Some(name.as_str())) + .expect("expected versions-only blob in listing"); + assert_eq!(Some(true), blob.has_versions_only); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test] +async fn test_list_blobs_arrow_copy_properties(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Standard, None).await?; + + // Async Copy Blob populates the copy_* properties on the destination blob. + let source_name = get_blob_name(recording); + let source_client = container_client.blob_client(&source_name); + create_test_blob( + &source_client, + Some(RequestContent::from(b"arrow copy source".to_vec())), + None, + ) + .await?; + + let dest_name = get_blob_name(recording); + let dest_client = container_client.blob_client(&dest_name); + dest_client + .start_copy_from_url(source_client.url().as_str().into(), None) + .await?; + + // Wait for the async copy to reach a terminal state so the listing reports it. + let mut copy_status = None; + for _ in 0..10 { + copy_status = dest_client.get_properties(None).await?.copy_status()?; + if copy_status != Some(CopyStatus::Pending) { + break; + } + if recording.test_mode() == TestMode::Live || recording.test_mode() == TestMode::Record { + time::sleep(Duration::from_secs(1)).await; + } + } + assert_eq!(Some(CopyStatus::Success), copy_status); + + let items = list_blobs_arrow(&container_client, Some(vec![ListBlobsIncludeItem::Copy])).await?; + let blob = items + .iter() + .find(|b| b.name.as_deref() == Some(dest_name.as_str())) + .expect("expected destination blob in listing"); + let props = blob.properties.as_ref().expect("expected blob properties"); + + // copy_status_description, incremental_copy, and destination_snapshot are not emitted by a + // successful non-incremental copy; they remain covered by the decoder unit tests. + assert!(props.copy_id.is_some()); + assert_eq!(Some(CopyStatus::Success), props.copy_status); + assert!(props.copy_source.is_some()); + assert!(props.copy_progress.is_some()); + assert!(props.copy_completion_time.is_some()); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test(playback)] +async fn test_list_blobs_arrow_immutability_properties( + ctx: TestContext, +) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Standard, None).await?; + + let blob_name = get_blob_name(recording); + let blob_client = container_client.blob_client(&blob_name); + create_test_blob(&blob_client, None, None).await?; + + let expiry = parse_rfc3339( + recording + .var( + "IMMUTABILITY_EXPIRY", + Some(VarOptions { + default_value: Some( + to_rfc3339( + &(OffsetDateTime::now_utc() + Duration::from_secs(60 * 60 * 24)), + ) + .into(), + ), + ..Default::default() + }), + ) + .as_str(), + )?; + blob_client + .set_immutability_policy( + &expiry, + Some(BlobClientSetImmutabilityPolicyOptions { + immutability_policy_mode: Some(ImmutabilityPolicyMode::Unlocked), + ..Default::default() + }), + ) + .await?; + blob_client.set_legal_hold(true, None).await?; + + let items = list_blobs_arrow( + &container_client, + Some(vec![ + ListBlobsIncludeItem::ImmutabilityPolicy, + ListBlobsIncludeItem::LegalHold, + ]), + ) + .await?; + let blob = items + .iter() + .find(|b| b.name.as_deref() == Some(blob_name.as_str())) + .expect("expected blob in listing"); + let props = blob.properties.as_ref().expect("expected blob properties"); + + // Assert + assert_eq!( + Some(ImmutabilityPolicyMode::Unlocked), + props.immutability_policy_mode + ); + assert!(props.immutability_policy_expires_on.is_some()); + assert_eq!(Some(true), props.legal_hold); + + blob_client.set_legal_hold(false, None).await?; + blob_client.delete_immutability_policy(None).await?; + blob_client.delete(None).await?; + let _ = container_client.delete(None).await; + Ok(()) +} + +#[recorded::test(playback)] +async fn test_list_blobs_arrow_last_accessed_on(ctx: TestContext) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let container_client = + get_container_client(recording, true, StorageAccount::Standard, None).await?; + + let blob_name = get_blob_name(recording); + let blob_client = container_client.blob_client(&blob_name); + create_test_blob(&blob_client, None, None).await?; + let _ = blob_client.download(None).await?.body.collect().await?; + + let items = list_blobs_arrow(&container_client, None).await?; + let blob = items + .iter() + .find(|b| b.name.as_deref() == Some(blob_name.as_str())) + .expect("expected blob in listing"); + let props = blob.properties.as_ref().expect("expected blob properties"); + + // Assert + assert!(props.last_accessed_on.is_some()); + + container_client.delete(None).await?; + Ok(()) +} + +#[recorded::test(playback)] +async fn test_list_blobs_arrow_object_replication_metadata( + ctx: TestContext, +) -> Result<(), Box> { + // Recording Setup + let recording = ctx.recording(); + let service_client = get_blob_service_client(recording, StorageAccount::Standard, None)?; + + let container_client = service_client.blob_container_client("test1"); + let blobs = list_blobs_arrow(&container_client, None).await?; + let blob = blobs + .iter() + .find(|blob| { + blob.name.as_deref() == Some("bla.txt") + && blob.version_id.as_deref() == Some("2022-08-29T21:54:26.5412339Z") + && blob.is_current_version == Some(true) + }) + .expect("expected configured object-replication source blob version"); + + let properties = blob + .object_replication_metadata + .as_ref() + .and_then(|metadata| metadata.additional_properties.as_ref()) + .expect("expected object replication metadata on configured source blob"); + + // Assert + assert!(!properties.is_empty()); + assert_eq!( + Some("complete"), + properties + .get("or-c570de93-3a83-4718-8ebe-f17b20d38a4f_49f6dc14-f5f7-4471-bf13-da984b86d136") + .map(String::as_str) + ); + Ok(()) +} + #[recorded::test] async fn test_list_blobs_with_continuation(ctx: TestContext) -> Result<(), Box> { // Recording Setup diff --git a/sdk/storage/azure_storage_blob/tests/common/mod.rs b/sdk/storage/azure_storage_blob/tests/common/mod.rs index 444493518c5..cbd96a0f1d6 100644 --- a/sdk/storage/azure_storage_blob/tests/common/mod.rs +++ b/sdk/storage/azure_storage_blob/tests/common/mod.rs @@ -29,12 +29,14 @@ use azure_core::{ use azure_core_test::{Recording, TestMode}; use azure_storage_blob::{ models::{ - BlockBlobClientUploadOptions, BlockBlobClientUploadResult, BlockLookupList, - EncryptionAlgorithmType, + BlobContainerClientListBlobsOptions, BlobItem, BlockBlobClientUploadOptions, + BlockBlobClientUploadResult, BlockLookupList, EncryptionAlgorithmType, + ListBlobsIncludeItem, StorageResponseFormat, }, BlobClient, BlobClientOptions, BlobContainerClient, BlobContainerClientOptions, BlobServiceClient, BlobServiceClientOptions, }; +use futures::TryStreamExt; pub const KB: usize = 1024; pub const MB: usize = KB * 1024; @@ -238,6 +240,35 @@ pub async fn create_test_blob( } } +pub async fn list_blobs_page( + container_client: &BlobContainerClient, + accept: StorageResponseFormat, + include: Option>, +) -> Result> { + let page = container_client + .list_blobs(Some(BlobContainerClientListBlobsOptions { + response_format: Some(accept), + include, + ..Default::default() + }))? + .into_pages() + .try_next() + .await? + .expect("list_blobs returned at least one page") + .into_model()?; + Ok(page.blob_items) +} + +/// Lists blobs in `container_client` using the Apache Arrow accept format and returns the +/// decoded blob items from the first page. Used by Arrow field-mapping tests to verify +/// listed properties over the wire. +pub async fn list_blobs_arrow( + container_client: &BlobContainerClient, + include: Option>, +) -> Result> { + list_blobs_page(container_client, StorageResponseFormat::Arrow, include).await +} + pub trait ClientOptionsExt { fn with_per_call_policy(self, policy: Arc) -> Self; fn with_per_try_policy(self, policy: Arc) -> Self; diff --git a/sdk/storage/azure_storage_blob/tsp-location.yaml b/sdk/storage/azure_storage_blob/tsp-location.yaml index b6a20314b82..14b34c6fc48 100644 --- a/sdk/storage/azure_storage_blob/tsp-location.yaml +++ b/sdk/storage/azure_storage_blob/tsp-location.yaml @@ -1,4 +1,4 @@ directory: specification/storage/data-plane/BlobStorage -commit: 7efab4ca79331ad2ac48eb67fb944d030745918d +commit: 12daaad818581eecbf1a33055365d72734f87d26 repo: Azure/azure-rest-api-specs -additionalDirectories: +additionalDirectories: