orionis.storage.drivers.AzureStorageDriver

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class AzureStorageDriver(IStorageDriver): (source)

Constructor: AzureStorageDriver(config)

View In Hierarchy

Storage driver backed by Azure Blob Storage.

Uses the official Azure SDK for Python (azure-storage-blob), which is an optional dependency: it is not installed with the framework. Install it before using this driver:

uv add 'orionis[azure]'

The SDK is imported lazily on first operation, and every blocking call runs on a worker thread via asyncio.to_thread. Directories are virtual: prefixes are inferred from blob names and explicit directories are stored as zero-byte path/ markers. Azure has no per-blob visibility: visibility reflects the container access level and setVisibility is unsupported.

Method	`__blob`	Return the blob client for normalized.
Method	`__containerClient`	Return the container client, bootstrapping it on first use.
Method	`__contentSettings`	Build the content settings for an upload, when derivable.
Method	`__init__`	Initialize the driver from an Azure disk configuration entity.
Method	`__listKeysSync`	List every blob name under prefix.
Method	`__missing`	Build the framework exception for a missing blob.
Method	`__propsSync`	Fetch blob properties or raise when the blob is absent.
Async Method	`__spool`	Buffer an async byte stream into a spooled temporary file.
Method	`__urlFor`	Compose the public URL of a blob without SDK involvement.
Async Method	`copy`	Copy the blob at source to target.
Async Method	`createDirectory`	Create a zero-byte directory marker at path.
Async Method	`delete`	Delete the blob at path.
Async Method	`deleteDirectory`	Recursively delete every blob under path.
Async Method	`directories`	List the directory prefixes contained under path.
Async Method	`directoryExists`	Check whether any blob exists under path.
Async Method	`download`	Download the blob at path to the local filesystem.
Async Method	`exists`	Check whether a blob exists at path.
Async Method	`files`	List the blob names that represent files under path.
Async Method	`hash`	Compute the content hash of the blob at path.
Async Method	`info`	Collect a metadata snapshot for the blob at path.
Async Method	`lastModified`	Return the last-modification timestamp of the blob at path.
Async Method	`mimeType`	Return the MIME type of the blob at path.
Async Method	`move`	Move the blob at source to target.
Method	`open`	Open an asynchronous binary stream for the blob at path.
Async Method	`read`	Read the full contents of the blob at path.
Async Method	`readStream`	Stream the contents of the blob at path in chunks.
Async Method	`setVisibility`	Change the visibility of the blob at path.
Async Method	`size`	Return the size in bytes of the blob at path.
Async Method	`temporaryUrl`	Build a SAS URL for the blob at path.
Async Method	`url`	Build the public URL for the blob at path.
Async Method	`visibility`	Return the effective visibility of the blob at path.
Async Method	`write`	Write contents to path, replacing any existing blob.
Async Method	`writeStream`	Write the chunks produced by stream to path.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_account_key`	Undocumented
Instance Variable	`_account_name`	Undocumented
Instance Variable	`_base_url`	Undocumented
Instance Variable	`_connection_string`	Undocumented
Instance Variable	`_container`	Undocumented
Instance Variable	`_container_name`	Undocumented
Instance Variable	`_http_error`	Undocumented
Instance Variable	`_not_found`	Undocumented
Instance Variable	`_sdk`	Undocumented

def __blob(self, normalized: str) -> Any: (source) ¶

Return the blob client for normalized.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`Any`	`BlobClient` bound to the blob.

def __containerClient(self) -> Any: (source) ¶

Return the container client, bootstrapping it on first use.

Returns
`Any`	Configured `ContainerClient` instance.

Raises
`MissingStorageDependencyException`	If `azure-storage-blob` is not installed.

def __contentSettings(self, normalized: str) -> Any: (source) ¶

Build the content settings for an upload, when derivable.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`Any`	`ContentSettings` with the guessed MIME type, or `None` when no type could be guessed.

def __init__(self, config: object): (source) ¶

Initialize the driver from an Azure disk configuration entity.

No SDK import or network activity happens here; the client is bootstrapped lazily on first use. When a connection string is provided, the account name and key are parsed from it so URLs and SAS tokens can be produced.

Parameters
config:`object`	Disk configuration exposing `connection_string`, `account_name`, `account_key`, `container`, and `url`.

Returns
`None`	Undocumented

def __listKeysSync(self, prefix: str) -> list[str]: (source) ¶

List every blob name under prefix.

Parameters
prefix:`str`	Name prefix to filter by; empty string lists everything.

Returns
`list[str]`	All matching blob names, including directory markers.

def __missing(self, normalized: str) -> StorageFileNotFoundException: (source) ¶

Build the framework exception for a missing blob.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`StorageFileNotFoundException`	Exception instance ready to be raised.

def __propsSync(self, normalized: str) -> Any: (source) ¶

Fetch blob properties or raise when the blob is absent.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`Any`	`BlobProperties` for the blob.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def __spool(self, stream: AsyncIterable[bytes]) -> BinaryIO: (source) ¶

Buffer an async byte stream into a spooled temporary file.

Parameters
stream:`AsyncIterable[bytes]`	Asynchronous byte-chunk producer.

Returns
`BinaryIO`	Rewound buffer holding the full payload.

def __urlFor(self, normalized: str) -> str: (source) ¶

Compose the public URL of a blob without SDK involvement.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`str`	URL derived from the configured base URL or the canonical Azure Blob endpoint.

async def copy(self, source: str, target: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.copy

Copy the blob at source to target.

The content is streamed through a spooled local buffer, which works with any authentication mode and never loads large blobs fully into memory.

Parameters
source:`str`	Root-relative path of the existing blob.
target:`str`	Root-relative destination path.

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the source blob does not exist.

async def createDirectory(self, path: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.createDirectory

Create a zero-byte directory marker at path.

Parameters
path:`str`	Root-relative directory path.

Returns
`None`	Undocumented

async def delete(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.delete

Delete the blob at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bool`	`True` if the blob existed and was removed.

async def deleteDirectory(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.deleteDirectory

Recursively delete every blob under path.

Parameters
path:`str`	Root-relative directory path. The empty string clears the whole container prefix space.

Returns
`bool`	`True` if at least one blob was removed.

async def directories(self, path: str = '', *, recursive: bool = False) -> list[str]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.directories

List the directory prefixes contained under path.

Parameters
path:`str`	Root-relative directory path. Empty string for the root.
recursive:`bool`	When `True`, include all nested prefixes.

Returns
`list[str]`	Sorted root-relative directory paths.

async def directoryExists(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.directoryExists

Check whether any blob exists under path.

Parameters
path:`str`	Root-relative directory path. The empty string denotes the disk root.

Returns
`bool`	`True` if the prefix contains at least one blob.

async def download(self, path: str, destination: str | Path) -> Path: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.download

Download the blob at path to the local filesystem.

Parameters
path:`str`	Root-relative file path on the disk.
destination:`str \| Path`	Local target. When it points to an existing directory the file keeps its original name inside that directory.

Returns
`Path`	Absolute local path of the downloaded file.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def exists(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.exists

Check whether a blob exists at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bool`	`True` if a blob exists at the given name.

async def files(self, path: str = '', *, recursive: bool = False) -> list[str]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.files

List the blob names that represent files under path.

Parameters
path:`str`	Root-relative directory path. Empty string for the root.
recursive:`bool`	When `True`, include files from all nested prefixes.

Returns
`list[str]`	Sorted root-relative file paths.

async def hash(self, path: str, algorithm: str = 'sha256') -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.hash

Compute the content hash of the blob at path.

The blob is streamed in chunks, so large files never load fully into memory.

Parameters
path:`str`	Root-relative file path.
algorithm:`str`	Any algorithm name accepted by `hashlib.new`.

Returns
`str`	Hexadecimal digest of the blob contents.

Raises
`StorageFileNotFoundException`	If the blob does not exist.
`UnsupportedStorageOperationException`	If algorithm is not available.

async def info(self, path: str) -> FileInfo: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.info

Collect a metadata snapshot for the blob at path.

The snapshot is built from blob properties only; checksum holds the Content-MD5 stored by Azure when available.

Parameters
path:`str`	Root-relative file path.

Returns
`FileInfo`	Immutable entity with size, MIME type, timestamps, ETag, visibility, and URL.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def lastModified(self, path: str) -> datetime: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.lastModified

Return the last-modification timestamp of the blob at path.

Parameters
path:`str`	Root-relative file path.

Returns
`datetime`	Timezone-aware modification timestamp.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def mimeType(self, path: str) -> str | None: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.mimeType

Return the MIME type of the blob at path.

Prefers the content type stored in Azure and falls back to a guess based on the file extension.

Parameters
path:`str`	Root-relative file path.

Returns
`str \| None`	MIME type, or `None` when it cannot be determined.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def move(self, source: str, target: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.move

Move the blob at source to target.

Implemented as a copy followed by a delete of the source blob.

Parameters
source:`str`	Root-relative path of the existing blob.
target:`str`	Root-relative destination path.

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the source blob does not exist.

def open(self, path: str, mode: str = 'rb') -> AsyncStream: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.open

Open an asynchronous binary stream for the blob at path.

Read-oriented modes download the blob into a spooled temporary buffer; writable modes upload the buffered content back to Azure when the stream is closed.

Parameters
path:`str`	Root-relative file path.
mode:`str`	Binary mode: `'rb'`, `'wb'`, `'ab'`, `'rb+'`, `'wb+'`, or `'ab+'`.

Returns
`AsyncStream`	Lazily opened stream; use it as an async context manager.

Raises
`UnsupportedStorageOperationException`	If mode is not a supported binary mode.

async def read(self, path: str) -> bytes: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.read

Read the full contents of the blob at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bytes`	Complete blob contents.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def readStream(self, path: str, chunk_size: int = _CHUNK_SIZE) -> AsyncIterator[bytes]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.readStream

Stream the contents of the blob at path in chunks.

Chunk sizing follows the SDK transfer configuration; the chunk_size parameter is advisory for this driver.

Parameters
path:`str`	Root-relative file path.
chunk_size:`int`	Advisory chunk size in bytes.

Returns
`AsyncIterator[bytes]`	Undocumented

Yields
`bytes`	Consecutive chunks of the blob contents.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def setVisibility(self, path: str, visibility: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.setVisibility

Change the visibility of the blob at path.

Azure Blob Storage does not support per-blob visibility, so this operation always fails. Adjust the container access level from the Azure portal or management SDK instead.

Parameters
path:`str`	Root-relative file path.
visibility:`str`	Requested visibility level.

Returns
`None`	Never returned by this driver.

Raises
`UnsupportedStorageOperationException`	Always, since Azure has no per-blob visibility.

async def size(self, path: str) -> int: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.size

Return the size in bytes of the blob at path.

Parameters
path:`str`	Root-relative file path.

Returns
`int`	Blob size in bytes.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def temporaryUrl(self, path: str, expires_in: int) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.temporaryUrl

Build a SAS URL for the blob at path.

Requires the storage account key, either configured explicitly or embedded in the connection string.

Parameters
path:`str`	Root-relative file path.
expires_in:`int`	Lifetime of the URL in seconds.

Returns
`str`	Read-only SAS URL valid for expires_in seconds.

Raises
`UnsupportedStorageOperationException`	If no account key is available for signing.
`MissingStorageDependencyException`	If `azure-storage-blob` is not installed.

async def url(self, path: str) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.url

Build the public URL for the blob at path.

Parameters
path:`str`	Root-relative file path.

Returns
`str`	URL derived from the configured base URL or the canonical Azure Blob endpoint.

async def visibility(self, path: str) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.visibility

Return the effective visibility of the blob at path.

Azure controls access at container level, so the result reflects the container access policy rather than a per-blob ACL.

Parameters
path:`str`	Root-relative file path.

Returns
`str`	`'public'` when the container allows anonymous access, otherwise `'private'`.

Raises
`StorageFileNotFoundException`	If the blob does not exist.

async def write(self, path: str, contents: bytes | str, visibility: str | None = None): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.write

Write contents to path, replacing any existing blob.

Azure Blob Storage has no per-blob visibility; access is governed by the container access level, so visibility is accepted for interface compatibility but ignored.

Parameters
path:`str`	Root-relative file path.
contents:`bytes \| str`	Data to persist. Strings are encoded as UTF-8.
visibility:`str \| None`	Ignored by this driver.

Returns
`None`	Undocumented

async def writeStream(self, path: str, stream: AsyncIterable[bytes], visibility: str | None = None): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.writeStream

Write the chunks produced by stream to path.

The payload is buffered into a spooled temporary file (spilling to disk past 8 MiB) and uploaded in blocks by the SDK.

Parameters
path:`str`	Root-relative file path.
stream:`AsyncIterable[bytes]`	Asynchronous byte-chunk producer.
visibility:`str \| None`	Ignored by this driver.