azure_blob_loader

azure_blob_loader ¤

AzureBlobParquetLoader ¤

AzureBlobParquetLoader(
    container_name: str | None = None,
    *,
    connection_string: str | None = None,
    account_url: str | None = None,
    credential: Any | None = None,
    sas_url: str | None = None,
    prefix: str = "",
    max_workers: int = 8,
    hour_pattern: str = "{Y}/{m}/{d}/{H}/"
)

Load parquet files from an Azure Blob Storage container filtered by a list of UUIDs.

Optimized for speed by: - Using server-side prefix filtering when provided - Streaming blob listings and filtering client-side by UUID containment - Downloading and parsing parquet files concurrently

Initialize the loader with Azure connection details.

Supports three authentication methods:

SAS URL (simplest)::

loader = AzureBlobParquetLoader( sas_url="https://account.blob.core.windows.net/container?sv=...&sig=..." )

A SAS URL is also auto-detected when passed as connection_string.

Connection string::

loader = AzureBlobParquetLoader( connection_string="DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...", container_name="mycontainer", )
AAD credential (account_url + credential)::

loader = AzureBlobParquetLoader( account_url="https://account.blob.core.windows.net", container_name="mycontainer", credential=DefaultAzureCredential(), )

Parameters:

Name	Type	Description	Default
`container_name`	`str \| None`	Target container name (not needed when using sas_url).	`None`
`connection_string`	`str \| None`	Azure Storage connection string.	`None`
`account_url`	`str \| None`	Full account URL for AAD auth.	`None`
`credential`	`Any \| None`	Azure credential object for AAD auth.	`None`
`sas_url`	`str \| None`	Full Blob SAS URL including container and SAS token.	`None`
`prefix`	`str`	Optional path prefix to narrow listing (e.g. "year/month/").	`''`
`max_workers`	`int`	Max concurrent downloads/reads.	`8`
`hour_pattern`	`str`	Pattern for hour-level subpath; tokens: {Y} {m} {d} {H}.	`'{Y}/{m}/{d}/{H}/'`

from_account_name `classmethod` ¤

from_account_name(
    account_name: str,
    container_name: str,
    *,
    credential: Any | None = None,
    endpoint_suffix: str = "blob.core.windows.net",
    prefix: str = "",
    max_workers: int = 8
) -> AzureBlobParquetLoader

Construct a loader using AAD credentials with an account name.

Parameters:

Name	Type	Description	Default
`account_name`	`str`	Storage account name.	required
`container_name`	`str`	Target container.	required
`credential`	`Any \| None`	Optional Azure credential (DefaultAzureCredential if None).	`None`
`endpoint_suffix`	`str`	DNS suffix for the blob endpoint (e.g., for sovereign clouds).	`'blob.core.windows.net'`
`prefix`	`str`	Optional listing prefix (e.g., "parquet/").	`''`
`max_workers`	`int`	Concurrency for downloads.	`8`

load_all_files ¤

load_all_files(
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load all parquet blobs in the container (optionally under prefix).

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

Returns:

Type	Description
`DataFrame`	A concatenated DataFrame of all parquet blobs. Returns an empty DataFrame
`DataFrame`	if none are found.

load_by_time_range ¤

load_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load all parquet blobs under hourly folders within [start, end].

Assumes container structure: prefix/year/month/day/hour/{file}.parquet Listing is constrained per-hour for speed.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

stream_by_time_range ¤

stream_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    columns: list[str] | None = None,
    filters: list | None = None,
) -> Iterator[tuple[str, pd.DataFrame]]

Stream parquet DataFrames under hourly folders within [start, end].

Yields (blob_name, DataFrame) one by one to avoid holding everything in memory.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

load_files_by_time_range_and_uuids ¤

load_files_by_time_range_and_uuids(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
) -> pd.DataFrame

Load parquet blobs for given UUIDs within [start, end] hours.

Strategy: 1) List each hour prefix once and keep blobs whose basename matches a requested UUID variant. When listing succeeds it is authoritative, so only blobs that actually exist are downloaded. 2) Only when listing fails entirely (e.g. a SAS token without list permission) fall back to constructing direct blob paths assuming the pattern prefix/YYYY/MM/DD/HH/{uuid}.parquet and downloading those blindly.

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

stream_files_by_time_range_and_uuids ¤

stream_files_by_time_range_and_uuids(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    uuid_list: list[str],
    columns: list[str] | None = None,
    filters: list | None = None,
) -> Iterator[tuple[str, pd.DataFrame]]

Stream parquet DataFrames for given UUIDs within [start, end] hours.

Yields (blob_name, DataFrame) as they arrive. Listing is authoritative when available; direct blob names are used only as a fallback when listing fails (see :meth:load_files_by_time_range_and_uuids).

Parameters:

Name	Type	Description	Default
`columns`	`list[str] \| None`	Subset of parquet columns to read; None reads all columns.	`None`
`filters`	`list \| None`	pyarrow-style predicate pushdown in DNF form, e.g. `[("value_double", ">", 0.0)]`; None reads all rows.	`None`

list_structure ¤

list_structure(
    parquet_only: bool = True, limit: int | None = None
) -> dict[str, list[str]]

List folder prefixes (hours) and blob names under the configured prefix.

Parameters:

Name	Type	Description	Default
`parquet_only`	`bool`	If True, only include blobs ending with .parquet.	`True`
`limit`	`int \| None`	Optional cap on number of files collected for quick inspection.	`None`

Returns:

Type	Description
`dict[str, list[str]]`	A dict with:
`dict[str, list[str]]`	folders: Sorted unique hour-level prefixes like 'parquet/YYYY/MM/DD/HH/'
`dict[str, list[str]]`	files: Sorted blob names (full paths) matching the filter

AzureBlobFlexibleFileLoader ¤

AzureBlobFlexibleFileLoader(
    container_name: str | None = None,
    *,
    connection_string: str | None = None,
    account_url: str | None = None,
    credential: Any | None = None,
    sas_url: str | None = None,
    prefix: str = "",
    max_workers: int = 8,
    hour_pattern: str = "{Y}/{m}/{d}/{H}/"
)

Load arbitrary file types from Azure Blob Storage under time-structured folders.

Designed for containers with paths like: prefix/YYYY/MM/DD/HH//file.ext This class lists by per-hour prefix and can filter by extensions and/or basenames, then downloads files concurrently as raw bytes.

Initialize the loader with Azure connection details.

Supports SAS URL, connection string, and AAD credential auth. See :class:AzureBlobParquetLoader for full parameter docs.

A SAS URL is auto-detected when passed as connection_string.

list_files_by_time_range ¤

list_files_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    *,
    extensions: Iterable[str] | None = None,
    limit: int | None = None
) -> list[str]

List blob names under each hourly prefix within [start, end].

Parameters:

Name	Type	Description	Default
`extensions`	`Iterable[str] \| None`	Optional set/list of file extensions (e.g., {"json", ".bmp"}). Case-insensitive.	`None`
`limit`	`int \| None`	Optional cap on number of files collected.	`None`

iter_file_names_by_time_range ¤

iter_file_names_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    *,
    extensions: Iterable[str] | None = None
) -> Iterator[str]

Yield blob names under each hourly prefix within [start, end]. Uses server-side prefix listing and client-side extension filtering.

fetch_files_by_time_range ¤

fetch_files_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    *,
    extensions: Iterable[str] | None = None,
    parse: bool = False
) -> dict[str, Any]

Download files that match extensions within [start, end] hour prefixes. Returns a dict mapping blob_name -> parsed object (if parse=True and a parser exists), otherwise raw bytes.

stream_files_by_time_range ¤

stream_files_by_time_range(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    *,
    extensions: Iterable[str] | None = None,
    parse: bool = False
) -> Iterator[tuple[str, Any]]

Stream matching files as (blob_name, bytes-or-parsed) within [start, end]. Maintains up to max_workers concurrent downloads while yielding incrementally.

fetch_files_by_time_range_and_basenames ¤

fetch_files_by_time_range_and_basenames(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    basenames: Iterable[str],
    *,
    extensions: Iterable[str] | None = None,
    parse: bool = False
) -> dict[str, Any]

Download files whose basename (final path segment) is in basenames, optionally filtered by extensions, within [start, end] hour prefixes. Returns blob_name -> parsed object (if parse=True and a parser exists), otherwise raw bytes.

stream_files_by_time_range_and_basenames ¤

stream_files_by_time_range_and_basenames(
    start_timestamp: str | Timestamp,
    end_timestamp: str | Timestamp,
    basenames: Iterable[str],
    *,
    extensions: Iterable[str] | None = None,
    parse: bool = False
) -> Iterator[tuple[str, Any]]

Stream files whose basename is in basenames within [start, end]. Yields (blob_name, bytes-or-parsed) incrementally with bounded concurrency.

azure_blob_loader