pyrit.models.AzureBlobStorageIO

pyrit.models.AzureBlobStorageIO#

class AzureBlobStorageIO(*, container_url: str | None = None, sas_token: str | None = None, blob_content_type: SupportedContentType = SupportedContentType.PLAIN_TEXT)[source]#

Bases: StorageIO

Implementation of StorageIO for Azure Blob Storage.

__init__(*, container_url: str | None = None, sas_token: str | None = None, blob_content_type: SupportedContentType = SupportedContentType.PLAIN_TEXT) → None[source]#

Initialize an Azure Blob Storage I/O adapter.

Parameters:

  • container_url (Optional[str]) – Azure Blob container URL.

  • sas_token (Optional[str]) – Optional SAS token.

  • blob_content_type (SupportedContentType) – Blob content type for uploads.

Raises:

ValueError – If container_url is missing.

Methods

__init__(*[, container_url, sas_token, ...])

Initialize an Azure Blob Storage I/O adapter.

create_directory_if_not_exists(directory_path)

Log a no-op directory creation for Azure Blob Storage.

is_file(path)

Check whether the path refers to a file (blob) in Azure Blob Storage.

parse_blob_url(file_path)

Parse a blob URL to extract the container and blob name.

path_exists(path)

Check whether a given path exists in the Azure Blob Storage container.

read_file(path)

Asynchronously reads the content of a file (blob) from Azure Blob Storage.

write_file(path, data)

Write data to Azure Blob Storage at the specified path.
async create_directory_if_not_exists(directory_path: Path | str) → None[source]#

Log a no-op directory creation for Azure Blob Storage.

Parameters:

directory_path (Union[Path, str]) – Requested directory path.

async is_file(path: Path | str) → bool[source]#

Check whether the path refers to a file (blob) in Azure Blob Storage.

Parameters:

path (Union[Path, str]) – Blob URL or path to test.

Returns:

True when the blob exists and has non-zero content size.

Return type:

bool

parse_blob_url(file_path: str) → tuple[str, str][source]#

Parse a blob URL to extract the container and blob name.

Parameters:

file_path (str) – Full blob URL.

Returns:

Container name and blob name.

Return type:

tuple[str, str]

Raises:

ValueError – If file_path is not a valid blob URL.

async path_exists(path: Path | str) → bool[source]#

Check whether a given path exists in the Azure Blob Storage container.

Parameters:

path (Union[Path, str]) – Blob URL or path to test.

Returns:

True when the path exists.

Return type:

bool

async read_file(path: Path | str) → bytes[source]#

Asynchronously reads the content of a file (blob) from Azure Blob Storage.

If the provided path is a full URL (e.g., “https://account.blob.core.windows.net/container/dir1/dir2/sample.png”), it extracts the relative blob path (e.g., “dir1/dir2/sample.png”) to correctly access the blob. If a relative path is provided, it will use it as-is.

Parameters:

path (str) – The path to the file (blob) in Azure Blob Storage. This can be either a full URL or a relative path.

Returns:

The content of the file (blob) as bytes.

Return type:

bytes

Raises:

Exception – If there is an error in reading the blob file, an exception will be logged and re-raised.

Example

file_content = await read_file(”https://account.blob.core.windows.net/container/dir2/1726627689003831.png”) # Or using a relative path: file_content = await read_file(“dir1/dir2/1726627689003831.png”)

async write_file(path: Path | str, data: bytes) → None[source]#

Write data to Azure Blob Storage at the specified path.

Parameters:

  • path (str) – The full Azure Blob Storage URL

  • data (bytes) – The data to write.

Contents