ts_shape.loader.timeseries.azure_blob_loader

Classes:

• AzureBlobFlexibleFileLoader –

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

• AzureBlobParquetLoader –

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

AzureBlobFlexibleFileLoader

AzureBlobFlexibleFileLoader(container_name: str, *, connection_string: Optional[str] = None, account_url: Optional[str] = None, credential: Optional[object] = 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.

Methods:

• fetch_files_by_time_range –

  Download files that match extensions within [start, end] hour prefixes.

• fetch_files_by_time_range_and_basenames –

  Download files whose basename (final path segment) is in basenames,

• iter_file_names_by_time_range –

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

• list_files_by_time_range –

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

• stream_files_by_time_range –

  Stream matching files as (blob_name, bytes-or-parsed) within [start, end].

• stream_files_by_time_range_and_basenames –

  Stream files whose basename is in basenames within [start, end].

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def __init__(
    self,
    container_name: str,
    *,
    connection_string: Optional[str] = None,
    account_url: Optional[str] = None,
    credential: Optional[object] = None,
    prefix: str = "",
    max_workers: int = 8,
    hour_pattern: str = "{Y}/{m}/{d}/{H}/",
) -> None:
    try:
        from azure.storage.blob import ContainerClient  # type: ignore
    except Exception as exc:  # pragma: no cover - import guard
        raise ImportError(
            "azure-storage-blob is required for AzureBlobFlexibleFileLoader. "
            "Install with `pip install azure-storage-blob`."
        ) from exc

    # Prefer AAD credential path if account_url provided or credential is given
    if account_url or (credential is not None and not connection_string):
        if not account_url:
            raise ValueError("account_url must be provided when using AAD credential auth")
        if credential is None:
            raise ValueError("credential must be provided when using AAD credential auth")
        self.container_client = ContainerClient(account_url=account_url, container_name=container_name, credential=credential)
    else:
        if not connection_string:
            raise ValueError("Either connection_string or (account_url + credential) must be provided")
        self.container_client = ContainerClient.from_connection_string(
            conn_str=connection_string, container_name=container_name
        )
    self.prefix = prefix
    self.max_workers = max_workers if max_workers > 0 else 1
    # Pattern for hour-level subpath; tokens: {Y} {m} {d} {H}
    self.hour_pattern = hour_pattern
    # Initialize parsers lazily once per process
    if not AzureBlobFlexibleFileLoader._parsers_initialized:
        self._enable_builtin_parsers()
        AzureBlobFlexibleFileLoader._parsers_initialized = True

fetch_files_by_time_range

fetch_files_by_time_range(start_timestamp: str | Timestamp, end_timestamp: str | Timestamp, *, extensions: Optional[Iterable[str]] = 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.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def fetch_files_by_time_range(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    *,
    extensions: Optional[Iterable[str]] = 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.
    """
    blob_names = self.list_files_by_time_range(start_timestamp, end_timestamp, extensions=extensions)
    if not blob_names:
        return {}
    results: Dict[str, Any] = {}
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name = {executor.submit(self._download_bytes, n): n for n in blob_names}
        for fut in as_completed(future_to_name):
            name = future_to_name[fut]
            content = fut.result()
            if content is not None:
                results[name] = self._parse_bytes(name, content) if parse else content
    return results

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: Optional[Iterable[str]] = 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.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def fetch_files_by_time_range_and_basenames(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    basenames: Iterable[str],
    *,
    extensions: Optional[Iterable[str]] = 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.
    """
    base_set = {str(b).strip() for b in basenames if str(b).strip()}
    allowed_exts = self._normalize_exts(extensions)
    candidates: List[str] = []
    for pfx in (self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)):
        blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
        for b in blob_iter:  # type: ignore[attr-defined]
            name = str(b.name)
            base = name.rsplit('/', 1)[-1]
            if base not in base_set:
                continue
            if allowed_exts is not None and not any(name.lower().endswith(ext) for ext in allowed_exts):
                continue
            candidates.append(name)

    if not candidates:
        return {}

    results: Dict[str, Any] = {}
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name = {executor.submit(self._download_bytes, n): n for n in candidates}
        for fut in as_completed(future_to_name):
            name = future_to_name[fut]
            content = fut.result()
            if content is not None:
                results[name] = self._parse_bytes(name, content) if parse else content
    return results

iter_file_names_by_time_range

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

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

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def iter_file_names_by_time_range(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    *,
    extensions: Optional[Iterable[str]] = None,
) -> Iterator[str]:
    """
    Yield blob names under each hourly prefix within [start, end].
    Uses server-side prefix listing and client-side extension filtering.
    """
    allowed_exts = self._normalize_exts(extensions)
    for pfx in (self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)):
        blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
        for b in blob_iter:  # type: ignore[attr-defined]
            name = str(b.name)
            if allowed_exts is not None:
                if not any(name.lower().endswith(ext) for ext in allowed_exts):
                    continue
            yield name

list_files_by_time_range

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

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

Parameters:

• extensions

  (Optional[Iterable[str]], default: None ) –

  Optional set/list of file extensions (e.g., {"json", ".bmp"}). Case-insensitive.

• limit

  (Optional[int], default: None ) –

  Optional cap on number of files collected.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def list_files_by_time_range(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    *,
    extensions: Optional[Iterable[str]] = None,
    limit: Optional[int] = None,
) -> List[str]:
    """
    List blob names under each hourly prefix within [start, end].

    Args:
        extensions: Optional set/list of file extensions (e.g., {"json", ".bmp"}). Case-insensitive.
        limit: Optional cap on number of files collected.
    """
    allowed_exts = self._normalize_exts(extensions)
    names: List[str] = []
    collected = 0
    for pfx in (self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)):
        blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
        for b in blob_iter:  # type: ignore[attr-defined]
            name = str(b.name)
            if allowed_exts is not None:
                lower_name = name.lower()
                if not any(lower_name.endswith(ext) for ext in allowed_exts):
                    continue
            names.append(name)
            collected += 1
            if limit is not None and collected >= limit:
                return names
    return names

stream_files_by_time_range

stream_files_by_time_range(start_timestamp: str | Timestamp, end_timestamp: str | Timestamp, *, extensions: Optional[Iterable[str]] = 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.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def stream_files_by_time_range(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    *,
    extensions: Optional[Iterable[str]] = 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.
    """
    names_iter = self.iter_file_names_by_time_range(start_timestamp, end_timestamp, extensions=extensions)

    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name: Dict[Any, str] = {}
        # initial fill
        try:
            while len(future_to_name) < self.max_workers:
                n = next(names_iter)
                future_to_name[executor.submit(self._download_bytes, n)] = n
        except StopIteration:
            pass

        while future_to_name:
            # Drain
            for fut in as_completed(list(future_to_name.keys())):
                name = future_to_name.pop(fut)
                try:
                    content = fut.result()
                except Exception:
                    content = None
                if content is not None:
                    yield (name, self._parse_bytes(name, content) if parse else content)

            # Refill
            try:
                while len(future_to_name) < self.max_workers:
                    n = next(names_iter)
                    future_to_name[executor.submit(self._download_bytes, n)] = n
            except StopIteration:
                pass

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: Optional[Iterable[str]] = 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.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def stream_files_by_time_range_and_basenames(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    basenames: Iterable[str],
    *,
    extensions: Optional[Iterable[str]] = 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.
    """
    base_set = {str(b).strip() for b in basenames if str(b).strip()}
    allowed_exts = self._normalize_exts(extensions)

    def _names_iter() -> Iterator[str]:
        for pfx in (self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)):
            blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
            for b in blob_iter:  # type: ignore[attr-defined]
                name = str(b.name)
                base = name.rsplit('/', 1)[-1]
                if base not in base_set:
                    continue
                if allowed_exts is not None and not any(name.lower().endswith(ext) for ext in allowed_exts):
                    continue
                yield name

    names_iter = _names_iter()
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name: Dict[Any, str] = {}
        # initial fill
        try:
            while len(future_to_name) < self.max_workers:
                n = next(names_iter)
                future_to_name[executor.submit(self._download_bytes, n)] = n
        except StopIteration:
            pass

        while future_to_name:
            for fut in as_completed(list(future_to_name.keys())):
                name = future_to_name.pop(fut)
                try:
                    content = fut.result()
                except Exception:
                    content = None
                if content is not None:
                    yield (name, self._parse_bytes(name, content) if parse else content)

            try:
                while len(future_to_name) < self.max_workers:
                    n = next(names_iter)
                    future_to_name[executor.submit(self._download_bytes, n)] = n
            except StopIteration:
                pass

AzureBlobParquetLoader

AzureBlobParquetLoader(container_name: str, *, connection_string: Optional[str] = None, account_url: Optional[str] = None, credential: Optional[object] = 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

Parameters:

• connection_string

  (Optional[str], default: None ) –

  Azure Storage connection string.

• container_name

  (str) –

  Target container name.

• prefix

  (str, default: '' ) –

  Optional path prefix to narrow listing (e.g. "year/month/").

• max_workers

  (int, default: 8 ) –

  Max concurrent downloads/reads.

Methods:

• from_account_name –

  Construct a loader using AAD credentials with an account name.

• list_structure –

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

• load_all_files –

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

• load_by_time_range –

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

• load_files_by_time_range_and_uuids –

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

• stream_by_time_range –

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

• stream_files_by_time_range_and_uuids –

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

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def __init__(
    self,
    container_name: str,
    *,
    connection_string: Optional[str] = None,
    account_url: Optional[str] = None,
    credential: Optional[object] = None,
    prefix: str = "",
    max_workers: int = 8,
    hour_pattern: str = "{Y}/{m}/{d}/{H}/",
) -> None:
    """
    Initialize the loader with Azure connection details.

    Args:
        connection_string: Azure Storage connection string.
        container_name: Target container name.
        prefix: Optional path prefix to narrow listing (e.g. "year/month/").
        max_workers: Max concurrent downloads/reads.
    """
    try:
        from azure.storage.blob import ContainerClient  # type: ignore
    except Exception as exc:  # pragma: no cover - import guard
        raise ImportError(
            "azure-storage-blob is required for AzureBlobParquetLoader. "
            "Install with `pip install azure-storage-blob`."
        ) from exc

    self._ContainerClient = ContainerClient
    # Prefer AAD credential path if account_url provided or credential is given
    if account_url or (credential is not None and not connection_string):
        if not account_url:
            raise ValueError("account_url must be provided when using AAD credential auth")
        if credential is None:
            raise ValueError("credential must be provided when using AAD credential auth")
        self.container_client = ContainerClient(account_url=account_url, container_name=container_name, credential=credential)
    else:
        if not connection_string:
            raise ValueError("Either connection_string or (account_url + credential) must be provided")
        self.container_client = ContainerClient.from_connection_string(
            conn_str=connection_string, container_name=container_name
        )
    self.prefix = prefix
    self.max_workers = max_workers if max_workers > 0 else 1
    # Pattern for hour-level subpath; tokens: {Y} {m} {d} {H}
    # Default matches many data lake layouts: YYYY/MM/DD/HH/
    self.hour_pattern = hour_pattern

from_account_name classmethod

from_account_name(account_name: str, container_name: str, *, credential: Optional[object] = 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:

• account_name

  (str) –

  Storage account name.

• container_name

  (str) –

  Target container.

• credential

  (Optional[object], default: None ) –

  Optional Azure credential (DefaultAzureCredential if None).

• endpoint_suffix

  (str, default: 'blob.core.windows.net' ) –

  DNS suffix for the blob endpoint (e.g., for sovereign clouds).

• prefix

  (str, default: '' ) –

  Optional listing prefix (e.g., "parquet/").

• max_workers

  (int, default: 8 ) –

  Concurrency for downloads.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

@classmethod
def from_account_name(
    cls,
    account_name: str,
    container_name: str,
    *,
    credential: Optional[object] = 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.

    Args:
        account_name: Storage account name.
        container_name: Target container.
        credential: Optional Azure credential (DefaultAzureCredential if None).
        endpoint_suffix: DNS suffix for the blob endpoint (e.g., for sovereign clouds).
        prefix: Optional listing prefix (e.g., "parquet/").
        max_workers: Concurrency for downloads.
    """
    account_url = f"https://{account_name}.{endpoint_suffix}"
    if credential is None:
        raise ValueError("credential must be provided when using AAD credential auth")
    return cls(
        container_name=container_name,
        account_url=account_url,
        credential=credential,
        prefix=prefix,
        max_workers=max_workers,
    )

list_structure

list_structure(parquet_only: bool = True, limit: Optional[int] = None) -> Dict[str, List[str]]

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

Parameters:

• parquet_only

  (bool, default: True ) –

  If True, only include blobs ending with .parquet.

• limit

  (Optional[int], default: None ) –

  Optional cap on number of files collected for quick inspection.

Returns:

• 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

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def list_structure(self, parquet_only: bool = True, limit: Optional[int] = None) -> Dict[str, List[str]]:
    """
    List folder prefixes (hours) and blob names under the configured `prefix`.

    Args:
        parquet_only: If True, only include blobs ending with .parquet.
        limit: Optional cap on number of files collected for quick inspection.

    Returns:
        A dict with:
        - folders: Sorted unique hour-level prefixes like 'parquet/YYYY/MM/DD/HH/'
        - files: Sorted blob names (full paths) matching the filter
    """
    folders: Set[str] = set()
    files: List[str] = []
    collected = 0

    blob_iter = self.container_client.list_blobs(name_starts_with=self.prefix or None)
    for b in blob_iter:
        name = str(b.name)
        if parquet_only and not name.endswith(".parquet"):
            continue
        files.append(name)
        # Derive hour-level folder prefix
        if "/" in name:
            folders.add(name.rsplit("/", 1)[0].rstrip("/") + "/")
        collected += 1
        if limit is not None and collected >= limit:
            break

    return {
        "folders": sorted(folders),
        "files": sorted(files),
    }

load_all_files

load_all_files() -> DataFrame

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

Returns:

• DataFrame –

  A concatenated DataFrame of all parquet blobs. Returns an empty DataFrame

• DataFrame –

  if none are found.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def load_all_files(self) -> pd.DataFrame:
    """
    Load all parquet blobs in the container (optionally under `prefix`).

    Returns:
        A concatenated DataFrame of all parquet blobs. Returns an empty DataFrame
        if none are found.
    """
    # List all parquet blob names using optional prefix for server-side filtering
    blob_iter = self.container_client.list_blobs(name_starts_with=self.prefix or None)
    blob_names = [b.name for b in blob_iter if str(b.name).endswith(".parquet")]  # type: ignore[attr-defined]
    if not blob_names:
        return pd.DataFrame()

    frames: List[pd.DataFrame] = []
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name = {executor.submit(self._download_parquet, name): name for name in blob_names}
        for future in as_completed(future_to_name):
            df = future.result()
            if df is not None and not df.empty:
                frames.append(df)

    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()

load_by_time_range

load_by_time_range(start_timestamp: str | Timestamp, end_timestamp: str | Timestamp) -> 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.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def load_by_time_range(self, start_timestamp: str | pd.Timestamp, end_timestamp: str | pd.Timestamp) -> 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.
    """
    hour_prefixes = [self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)]
    blob_names: List[str] = []
    for pfx in hour_prefixes:
        blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
        blob_names.extend([b.name for b in blob_iter if str(b.name).endswith(".parquet")])  # type: ignore[attr-defined]

    if not blob_names:
        return pd.DataFrame()

    frames: List[pd.DataFrame] = []
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name = {executor.submit(self._download_parquet, name): name for name in blob_names}
        for future in as_completed(future_to_name):
            df = future.result()
            if df is not None and not df.empty:
                frames.append(df)

    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()

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]) -> DataFrame

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

Strategy: 1) Construct direct blob paths assuming pattern prefix/YYYY/MM/DD/HH/{uuid}.parquet (fast path, no listing). 2) For robustness, also list each hour prefix and include any blob whose basename equals one of the requested UUID variants (handles case differences and extra subfolders below the hour level).

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def load_files_by_time_range_and_uuids(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    uuid_list: List[str],
) -> pd.DataFrame:
    """
    Load parquet blobs for given UUIDs within [start, end] hours.

    Strategy:
    1) Construct direct blob paths assuming pattern prefix/YYYY/MM/DD/HH/{uuid}.parquet
       (fast path, no listing).
    2) For robustness, also list each hour prefix and include any blob whose basename
       equals one of the requested UUID variants (handles case differences and extra
       subfolders below the hour level).
    """
    if not uuid_list:
        return pd.DataFrame()

    # Sanitize and deduplicate UUIDs while preserving order
    def _clean_uuid(u: object) -> str:
        s = str(u).strip().strip("{}").strip()
        return s

    raw = [_clean_uuid(u) for u in uuid_list]
    # Include lowercase variants to be tolerant of case differences in filenames
    variants_ordered: List[str] = []
    seen: Set[str] = set()
    for u in raw:
        for v in (u, u.lower()):
            if v and v not in seen:
                seen.add(v)
                variants_ordered.append(v)

    hour_prefixes = [self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)]

    # 1) Fast path: build direct blob names
    direct_names = [f"{pfx}{u}.parquet" for pfx in hour_prefixes for u in variants_ordered]

    # 2) Robust path: list each hour prefix and filter by basename match
    basenames = {f"{u}.parquet" for u in variants_ordered}
    listed_names: List[str] = []
    try:
        for pfx in hour_prefixes:
            blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
            for b in blob_iter:  # type: ignore[attr-defined]
                name = str(b.name)
                if not name.endswith(".parquet"):
                    continue
                base = name.rsplit("/", 1)[-1]
                if base in basenames:
                    listed_names.append(name)
    except Exception:
        # If listing fails for any reason, continue with direct names only
        pass

    # Merge and preserve order, avoid duplicates
    all_blob_names = list(dict.fromkeys([*direct_names, *listed_names]))

    if not all_blob_names:
        return pd.DataFrame()

    frames: List[pd.DataFrame] = []
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        future_to_name = {executor.submit(self._download_parquet, name): name for name in all_blob_names}
        for future in as_completed(future_to_name):
            df = future.result()
            if df is not None and not df.empty:
                frames.append(df)

    return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()

stream_by_time_range

stream_by_time_range(start_timestamp: str | Timestamp, end_timestamp: str | Timestamp) -> Iterator[Tuple[str, DataFrame]]

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

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

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def stream_by_time_range(self, start_timestamp: str | pd.Timestamp, end_timestamp: str | pd.Timestamp) -> 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.
    """
    hour_prefixes = [self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)]

    def _names_iter() -> Iterator[str]:
        for pfx in hour_prefixes:
            blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
            for b in blob_iter:  # type: ignore[attr-defined]
                name = str(b.name)
                if name.endswith(".parquet"):
                    yield name

    names_iter = _names_iter()
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        futures: Dict[Any, str] = {}
        # initial fill
        try:
            while len(futures) < self.max_workers:
                n = next(names_iter)
                futures[executor.submit(self._download_parquet, n)] = n
        except StopIteration:
            pass

        while futures:
            # Drain current batch
            for fut in as_completed(list(futures.keys())):
                name = futures.pop(fut)
                try:
                    df = fut.result()
                except Exception:
                    df = None
                if df is not None and not df.empty:
                    yield (name, df)

            # Refill
            try:
                while len(futures) < self.max_workers:
                    n = next(names_iter)
                    futures[executor.submit(self._download_parquet, n)] = n
            except StopIteration:
                pass

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]) -> Iterator[Tuple[str, DataFrame]]

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

Yields (blob_name, DataFrame) as they arrive. Uses direct names plus per-hour listing fallback.

Source code in src/ts_shape/loader/timeseries/azure_blob_loader.py

def stream_files_by_time_range_and_uuids(
    self,
    start_timestamp: str | pd.Timestamp,
    end_timestamp: str | pd.Timestamp,
    uuid_list: List[str],
) -> Iterator[Tuple[str, pd.DataFrame]]:
    """
    Stream parquet DataFrames for given UUIDs within [start, end] hours.

    Yields (blob_name, DataFrame) as they arrive. Uses direct names plus per-hour listing fallback.
    """
    if not uuid_list:
        return iter(())

    def _clean_uuid(u: object) -> str:
        return str(u).strip().strip("{}").strip()

    raw = [_clean_uuid(u) for u in uuid_list]
    variants_ordered: List[str] = []
    seen: Set[str] = set()
    for u in raw:
        for v in (u, u.lower()):
            if v and v not in seen:
                seen.add(v)
                variants_ordered.append(v)

    hour_prefixes = [self._hour_prefix(ts) for ts in self._hourly_slots(start_timestamp, end_timestamp)]
    direct_names = [f"{pfx}{u}.parquet" for pfx in hour_prefixes for u in variants_ordered]

    basenames = {f"{u}.parquet" for u in variants_ordered}

    def _names_iter() -> Iterator[str]:
        # yield direct first
        yielded: Set[str] = set()
        for n in direct_names:
            yielded.add(n)
            yield n
        # then list per-hour
        for pfx in hour_prefixes:
            blob_iter = self.container_client.list_blobs(name_starts_with=pfx)
            for b in blob_iter:  # type: ignore[attr-defined]
                name = str(b.name)
                if not name.endswith(".parquet"):
                    continue
                base = name.rsplit("/", 1)[-1]
                if base in basenames and name not in yielded:
                    yielded.add(name)
                    yield name

    names_iter = _names_iter()
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        futures: Dict[Any, str] = {}
        try:
            while len(futures) < self.max_workers:
                n = next(names_iter)
                futures[executor.submit(self._download_parquet, n)] = n
        except StopIteration:
            pass

        while futures:
            for fut in as_completed(list(futures.keys())):
                name = futures.pop(fut)
                try:
                    df = fut.result()
                except Exception:
                    df = None
                if df is not None and not df.empty:
                    yield (name, df)

            try:
                while len(futures) < self.max_workers:
                    n = next(names_iter)
                    futures[executor.submit(self._download_parquet, n)] = n
            except StopIteration:
                pass