azure-storage-blob
azure-storage-blob is Microsoft's official Python client for Azure Blob Storage, the object store that plays the same role in Azure that S3 does in AWS. The API is a hierarchy of three clients: BlobServiceClient for the account, ContainerClient for one container, and BlobClient for one blob, each able to hand you the next one down. Uploads and downloads chunk themselves automatically, so a multi-gigabyte file goes up in parallel blocks without you writing that logic. Beyond get and put it covers the parts of blob storage people actually hit in production: shared access signatures, blob tags and metadata, access tiers, leases, snapshots and versions, immutability policies, append and page blobs, server-side copy, and client-side encryption. A complete asyncio version of every client lives in azure.storage.blob.aio with the same method names.
The default and effectively only complete client for Azure Blob Storage from Python, well typed and consistent with the rest of the Azure SDK once the RBAC groundwork is in place. If your work is mostly reading and writing files, an fsspec layer on top will save you a lot of client plumbing.
Use it if
- You are on Azure and want managed identity instead of secrets: DefaultAzureCredential from azure-identity means no account keys in config, and the same code runs locally under your az login and in production under a workload identity
- You move large files and want the chunking, parallelism, and retries handled: pass max_concurrency and the client splits the transfer into blocks, with a progress_hook if you need a progress bar
- You need blob features beyond read and write, such as time-based SAS URLs, blob tags you can query with find_blobs_by_tags, access tier changes, leases for mutual exclusion, or legal holds
- You have an asyncio service: azure.storage.blob.aio mirrors the sync API method for method, so the two versions of a codebase stay readable side by side
- Your account has hierarchical namespace turned on and you need real directories, atomic renames, or POSIX-style ACLs. Those are Data Lake Gen2 operations, and azure-storage-file-datalake is the package that exposes them
- You want cloud-agnostic code or a file-like object to hand to pandas. adlfs gives you an fsspec filesystem and smart-open gives you an open() that speaks several clouds; both are far less code than three nested client objects
- You are optimising a slim container image. This pulls azure-core, cryptography, isodate, and typing-extensions, and cryptography carries compiled extensions, which is real weight if the only thing you do is fetch one JSON file
- You expect fast issue triage. The tracker is the whole azure-sdk-for-python monorepo covering every Azure service, with roughly 795 open issues plus more PRs, so a storage-specific report competes with everything else
- You are bulk-migrating terabytes. Driving that from Python is slower and more fragile than running azcopy, which is built for exactly that job
Setup reality
pip install azure-storage-blob needs Python 3.9 or newer, and in practice you also install azure-identity for token auth and use the aio extra when you want the async clients, since those need an async HTTP transport such as aiohttp. The install is the easy part. The blocker is nearly always RBAC: subscription Owner does not grant data-plane access, so you need the Storage Blob Data Reader or Storage Blob Data Contributor role on the account, and a fresh role assignment can take several minutes to take effect while your code returns 403. Many organisations also disable shared key access at the account level, which turns every connection-string example on the internet into an AuthorizationFailure. Two API defaults surprise people: upload_blob refuses to overwrite unless you pass overwrite=True, and download_blob returns a stream object rather than bytes, so you have to call readall, readinto, or chunks.
Patterns
Connect with Entra ID instead of keysclient-managed-identity
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
credential = DefaultAzureCredential()
service = BlobServiceClient(
account_url="https://myaccount.blob.core.windows.net",
credential=credential,
)
container = service.get_container_client("uploads")This needs the Storage Blob Data Contributor role on the account, not just Owner on the subscription. A new assignment can take a few minutes to apply, so an immediate 403 is often propagation rather than a bug in your code.
Connect with a connection stringclient-connection-string
import os
from azure.storage.blob import BlobServiceClient
service = BlobServiceClient.from_connection_string(
conn_str=os.environ["AZURE_STORAGE_CONNECTION_STRING"]
)
blob = service.get_blob_client(container="uploads", blob="report.csv")Quickest path locally and the one every tutorial uses, but it embeds an account key that grants full access to the account. Plenty of organisations disable shared key auth entirely, in which case this fails with AuthorizationFailure no matter how correct the string is.
Upload a file, overwriting if presentupload-file
from azure.storage.blob import ContentSettings
blob = container.get_blob_client("reports/2026-08.csv")
with open("local.csv", "rb") as f:
blob.upload_blob(
f,
overwrite=True,
content_settings=ContentSettings(content_type="text/csv"),
)overwrite defaults to False and raises ResourceExistsError, which catches nearly everyone once. Without content_settings the blob gets application/octet-stream, so browsers download it instead of displaying it.
Download straight into a filedownload-to-file
blob = container.get_blob_client("reports/2026-08.csv")
with open("local.csv", "wb") as f:
blob.download_blob(max_concurrency=4).readinto(f)
# small blobs, in memory
text = container.get_blob_client("note.txt").download_blob(encoding="utf-8").readall()download_blob returns a StorageStreamDownloader, not bytes. readinto streams to a file handle without holding the blob in memory; readall does hold all of it, which is fine for small objects and disastrous for large ones.
Process a large blob chunk by chunkstream-in-chunks
downloader = blob.download_blob()
for chunk in downloader.chunks():
process(chunk) # bytes, max_chunk_get_size each (4 MiB by default)chunks() is the memory-safe way to handle a blob larger than RAM. Adjust the size with the max_chunk_get_size keyword on the client if 4 MiB requests are too chatty for your network.
List blobs under a prefixlist-blobs
for props in container.list_blobs(name_starts_with="reports/2026/"):
print(props.name, props.size, props.last_modified)
# names only, cheaper to iterate
names = list(container.list_blob_names(name_starts_with="reports/"))The result is an ItemPaged that fetches more pages as you iterate, so wrapping it in list() on a large container makes many requests. list_blob_names skips deserialising full properties when you only need keys.
Treat prefixes as folderswalk-virtual-directories
from azure.storage.blob import BlobPrefix
for item in container.walk_blobs(name_starts_with="reports/", delimiter="/"):
if isinstance(item, BlobPrefix):
print("dir :", item.name)
else:
print("blob:", item.name)Flat blob storage has no directories; walk_blobs fakes them by grouping on the delimiter. Real directory semantics, including atomic rename, only exist on accounts with a hierarchical namespace through the Data Lake package.
Delete only if it is therehandle-missing-blobs
from azure.core.exceptions import ResourceNotFoundError
try:
blob.delete_blob(delete_snapshots="include")
except ResourceNotFoundError:
pass # already gone
if blob.exists():
props = blob.get_blob_properties()Deleting a blob that has snapshots fails unless you pass delete_snapshots. exists() costs a round trip, so in a loop prefer catching ResourceNotFoundError over checking first, which also avoids a race.
Time-limited download link without account keysuser-delegation-sas
from datetime import datetime, timedelta, timezone
from azure.storage.blob import BlobSasPermissions, generate_blob_sas
now = datetime.now(timezone.utc)
key = service.get_user_delegation_key(now, now + timedelta(hours=1))
token = generate_blob_sas(
account_name=service.account_name,
container_name="uploads",
blob_name="report.csv",
user_delegation_key=key,
permission=BlobSasPermissions(read=True),
expiry=now + timedelta(minutes=15),
)
url = f"{blob.url}?{token}"A user delegation key needs a token credential on the service client and is signed by Entra ID rather than the account key, so revoking the identity kills the links. The SAS cannot outlive the delegation key, and clock skew makes very short expiries unreliable.
Tune chunking and watch progresstune-large-transfers
service = BlobServiceClient(
account_url=url,
credential=credential,
max_single_put_size=8 * 1024 * 1024, # default 64 MiB
max_block_size=8 * 1024 * 1024, # default 4 MiB
)
def on_progress(current, total):
print(f"{current}/{total}")
with open("big.tar", "rb") as f:
blob.upload_blob(f, overwrite=True, max_concurrency=8, progress_hook=on_progress)Anything at or under max_single_put_size goes in one PUT; above it the file is split into blocks of max_block_size and max_concurrency of them fly at once. Lower the single-put size on flaky networks so a failure retries a block instead of the whole file.
Attach metadata and searchable tagsmetadata-and-tags
blob.upload_blob(
data,
overwrite=True,
metadata={"source": "nightly-etl", "rows": "12043"},
tags={"stage": "raw", "dataset": "orders"},
)
# tags are queryable across the whole container
for hit in container.find_blobs_by_tags("\"stage\" = 'raw'"):
print(hit.name)Metadata is free-form and only readable per blob; tags are indexed and queryable but capped at 10 per blob with a restricted character set. Note that set_blob_metadata replaces the whole dictionary rather than merging into it.
The asyncio versionasync-client
import asyncio
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient
async def main():
async with DefaultAzureCredential() as credential:
async with BlobServiceClient(account_url=url, credential=credential) as service:
container = service.get_container_client("uploads")
async for props in container.list_blobs():
print(props.name)
asyncio.run(main())Install the aio extra so an async transport is present, and use the async credential from azure.identity.aio rather than the sync one. Both clients and credentials hold connections, so skipping async with leaks sockets and prints unclosed-session warnings on shutdown.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| azure-storage-file-datalake | PyPI | The account uses a hierarchical namespace and you need directories, renames, or ACLs. |
| adlfs | PyPI | You want an fsspec filesystem so pandas, dask, and pyarrow can read Azure paths directly. |
| smart-open | PyPI | You just want open() over a URL and would rather not learn a cloud-specific client hierarchy. |