StorageHandler

Module: handler

Description

Handles interactions with both local filesystem and Blob Storage.

This class provides a unified interface for reading, writing, and managing data regardless of whether it’s stored locally or in Azure. The storage backend is determined by the config argument:

config=None: local filesystem (./data/ directory)
config=<AzureStorageConfig>: cloud storage using that configuration

Build an AzureStorageConfig directly (e.g. from values loaded from a TOML file) or via AzureStorageConfig.from_env_file(...).

Authentication for Azure: Uses either a device code token, a SAS token (with user delegation key) or client secret credentials.

`StorageHandler.read_json_blob()`

read_json_blob(self, blob_name: str, **kwargs) -> pl.DataFrame

Reads a JSON file from storage into a Polars DataFrame.

Argument	Type	Description
`blob_name`	str	The path/name of the file to read.
`**kwargs`		Additional keyword arguments passed to `polars.read_json`.

`StorageHandler.read_csv_blob()`

read_csv_blob(self, blob_name: str, **kwargs) -> pl.DataFrame

Reads a CSV file from storage into a Polars DataFrame.

Argument	Type	Description
`blob_name`	str	The path/name of the file to read.
`**kwargs`		Additional keyword arguments passed to `polars.read_csv`.

`StorageHandler.read_excel_blob()`

read_excel_blob(self, blob_name: str, **kwargs) -> pl.DataFrame

Reads an Excel file from storage into a Polars DataFrame.

Argument	Type	Description
`blob_name`	str	The path/name of the file to read.
`**kwargs`		Additional keyword arguments passed to `polars.read_excel`.

`StorageHandler.read_parquet_blob()`

read_parquet_blob(self, blob_name: str) -> Tuple[pl.DataFrame, Dict]

Reads a Parquet file from storage into a Polars DataFrame.

Argument	Type	Description
`blob_name`	str	The path/name of the file to read.

`StorageHandler.write_json_blob()`

write_json_blob(self, blob_name: str, df: pl.DataFrame, **kwargs) -> None

Writes a Polars DataFrame to a JSON file in storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to write to.
`df`	pl.DataFrame	The DataFrame to write.
`**kwargs`		Additional keyword arguments passed to `DataFrame.write_json`.

`StorageHandler.write_csv_blob()`

write_csv_blob(self, blob_name: str, df: pl.DataFrame, **kwargs) -> None

Writes a Polars DataFrame to a CSV file in storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to write to.
`df`	pl.DataFrame	The DataFrame to write.
`**kwargs`		Additional keyword arguments passed to `DataFrame.write_csv`.

`StorageHandler.write_excel_blob()`

write_excel_blob(self, blob_name: str, df: pl.DataFrame, **kwargs) -> None

Writes a Polars DataFrame to an Excel file in storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to write to.
`df`	pl.DataFrame	The DataFrame to write.
`**kwargs`		Additional keyword arguments passed to `DataFrame.write_excel`.

`StorageHandler.write_parquet_blob()`

write_parquet_blob(self, blob_name: str, df: pl.DataFrame, **kwargs) -> None

Writes a Polars DataFrame to a Parquet file in storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to write to.
`df`	pl.DataFrame	The DataFrame to write.
`**kwargs`		Additional keyword arguments passed to `DataFrame.write_parquet`.

`StorageHandler.read_blob_as_bytes()`

read_blob_as_bytes(self, blob_name: str) -> bytes

Read a blob’s raw bytes from local or Azure storage.

Argument	Type	Description
`blob_name`	str	The path/name of the blob to read.

`StorageHandler.get_device_code_auth_status()`

get_device_code_auth_status(token_cache_path: str = DEFAULT_TOKEN_CACHE, auth_record_cache_path: str = DEFAULT_AUTH_RECORD_CACHE) -> Dict

Inspect device code auth cache files on disk.

Argument	Type	Description
`token_cache_path`	str	Path to the token cache file. Defaults to DEFAULT_TOKEN_CACHE.
`auth_record_cache_path`	str	Path to the auth record file. Defaults to DEFAULT_AUTH_RECORD_CACHE.

`StorageHandler.verify_device_code_token()`

verify_device_code_token(tenant_id: str, client_id: str, scopes: Optional[List] = None, token_cache_path: str = DEFAULT_TOKEN_CACHE, auth_record_cache_path: str = DEFAULT_AUTH_RECORD_CACHE) -> bool

Attempt silent token acquisition from cache.

Argument	Type	Description
`tenant_id`	str	Azure tenant ID.
`client_id`	str	Azure client ID.
`scopes`	Optional[List]	OAuth scopes to request. Defaults to DEFAULT_SCOPES.
`token_cache_path`	str	Path to the token cache file. Defaults to DEFAULT_TOKEN_CACHE.
`auth_record_cache_path`	str	Path to the auth record file. Defaults to DEFAULT_AUTH_RECORD_CACHE.

`StorageHandler.clear_device_code_caches()`

clear_device_code_caches(cls, token_cache_path: str = DEFAULT_TOKEN_CACHE, auth_record_cache_path: str = DEFAULT_AUTH_RECORD_CACHE) -> List

Delete cached device code auth files from disk.

Argument	Type	Description
`token_cache_path`	str	Path to the token cache file. Defaults to DEFAULT_TOKEN_CACHE.
`auth_record_cache_path`	str	Path to the auth record file. Defaults to DEFAULT_AUTH_RECORD_CACHE.

`StorageHandler.delete_device_code_caches()`

delete_device_code_caches(self, token_cache_path: str = DEFAULT_TOKEN_CACHE, auth_record_cache_path: str = DEFAULT_AUTH_RECORD_CACHE) -> None

Delete cached refresh token and records for device code auth.

Argument	Type	Description
`token_cache_path`	str	Path to the device code refresh token cache file. Defaults to DEFAULT_TOKEN_CACHE.
`auth_record_cache_path`	str	Path to the authentication record file. Defaults to DEFAULT_AUTH_RECORD_CACHE.

`StorageHandler.list_blobs()`

list_blobs(self) -> List

Lists all blobs in the container with metadata.

`StorageHandler.print_blob_tree()`

print_blob_tree(self, blobs: List, container_label: str, account_label: Optional[str] = None, environment_label: Optional[str] = None) -> None

Renders a rich file tree from a flat list of blob paths.

Argument	Type	Description
`blobs`	List	List of BlobInfo objects to display.
`container_label`	str	Label for the root tree node.
`account_label`	Optional[str]	Storage account name.
`environment_label`	Optional[str]	Environment name.

`StorageHandler.scan_csv_blob()`

scan_csv_blob(self, blob_name: str, **kwargs) -> pl.LazyFrame

Scans a CSV file or multiple files from storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to scan.
`**kwargs`		Additional keyword arguments passed to `pl.scan_csv`.

`StorageHandler.scan_parquet_blob()`

scan_parquet_blob(self, blob_name: str, **kwargs) -> Tuple[pl.LazyFrame, Dict]

Scans a Parquet file from storage as a Polars LazyFrame.

Argument	Type	Description
`blob_name`	str	The path/name of the file to scan.
`**kwargs`		Additional keyword arguments passed to `pl.scan_parquet()`.

`StorageHandler.download_blob()`

download_blob(self, blob_name: str, progress_hook: Optional[Callable] = None) -> Optional[Path]

Downloads a blob from Azure Storage to a local file.

Argument	Type	Description
`blob_name`	str	The name of the blob to download.
`progress_hook`	Optional[Callable]	Optional callback invoked after each chunk with (bytes_downloaded, total_bytes).

`StorageHandler.download_dir()`

download_dir(self, dir_prefix: str, local_dir: Optional[Union[str, Path]] = None, progress_hook: Optional[Callable] = None) -> List[Path]

Download all blobs under a prefix to a local directory.

Argument	Type	Description
`dir_prefix`	str	The storage path/prefix to download recursively. A trailing `/` is optional.
`local_dir`	Optional[Union[str, Path]]	The local destination directory. Defaults to `./data` (mirroring :meth:`download_blob`). Created if it does not exist.
`progress_hook`	Optional[Callable]	Optional callback invoked after each file finishes download with `(file_path, bytes_downloaded, total_bytes)`.

`StorageHandler.write_blob_as_bytes()`

write_blob_as_bytes(self, blob_name: str, data: bytes) -> Optional[Dict[str, Any]]

Write raw bytes to a blob in local or Azure storage.

Argument	Type	Description
`blob_name`	str	The path/name of the blob to write.
`data`	bytes	The raw bytes to write.

`StorageHandler.sink_parquet_blob()`

sink_parquet_blob(self, blob_name: str, lf: pl.LazyFrame, **kwargs) -> None

Streams a Polars LazyFrame to a Parquet file in storage.

Argument	Type	Description
`blob_name`	str	The path/name of the file to write to.
`lf`	pl.LazyFrame	The LazyFrame to sink.
`**kwargs`		Additional keyword arguments passed to `LazyFrame.sink_parquet`.

`StorageHandler.upload_blob()`

upload_blob(self, blob_name: str, file_path: Union[str, Path], progress_hook: Optional[Callable] = None) -> None

Uploads a local file to Azure Blob Storage.

Argument	Type	Description
`blob_name`	str	The destination path/name in the container.
`file_path`	Union[str, Path]	Path to the local file to upload.
`progress_hook`	Optional[Callable]	Optional callback invoked during upload with (bytes_uploaded, total_bytes).

`StorageHandler.upload_dir()`

upload_dir(self, local_dir: Union[str, Path], blob_prefix: str = '', progress_hook: Optional[Callable] = None) -> List[str]

Upload all files from a local directory to storage.

Argument	Type	Description
`local_dir`	Union[str, Path]	The local directory to upload recursively.
`blob_prefix`	str	The destination prefix in storage. Defaults to empty (root of container).
`progress_hook`	Optional[Callable]	Optional callback invoked during each file upload with `(blob_name, bytes_uploaded, total_bytes)`.

`StorageHandler.delete_blob()`

delete_blob(self, blob_name: str) -> List

Delete a blob or prefix from storage.

Argument	Type	Description
`blob_name`	str	The name/path of the blob or prefix to delete.

`StorageHandler.move_blob()`

move_blob(self, source: str, destination: str, overwrite: bool = False) -> None

Move (rename) a blob within the same container.

Argument	Type	Description
`source`	str	Current path/name of the blob.
`destination`	str	New path/name for the blob.
`overwrite`	bool	If True, overwrite an existing blob at the destination. Defaults to False.