API Reference

download_dataset is the canonical public API for downloading dataset archives. save_dataset_to_disk is still available as a deprecated alias for backward compatibility.

datacollective.datasets

download_dataset(dataset_id, download_directory=None, show_progress=True, overwrite_existing=False, enable_logging=False, extract=False, overwrite_extracted=False)

Download the dataset archive to a local directory and return the archive path. Skips download if the target file already exists (unless overwrite_existing=True).

Automatically resumes interrupted downloads if a matching .checksum file exists from a previous attempt.

With extract=True, also unpacks the archive into a directory beside it, named after the archive without its extension, and returns that directory instead. An existing directory is reused unless overwrite_extracted=True, or the archive itself was downloaded again with overwrite_existing=True. This is the directory load_dataset extracts to, so the two never unpack the same archive twice.

Note: Previously called save_dataset_to_disk, which remains available as a deprecated alias for backward compatibility.

Parameters:

Name	Type	Description	Default
dataset_id	str	The dataset ID (as shown in MDC platform) or slug.	required
download_directory	str | None	Directory where to save the downloaded archive file. If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.	None
show_progress	bool	Whether to show a progress bar during download.	True
overwrite_existing	bool	Whether to overwrite the existing archive file. With extract=True the extracted directory is then unpacked afresh too.	False
enable_logging	bool	Whether to enable SDK logging to console and a local log file.	False
extract	bool	Whether to also unpack the archive and return the extracted directory.	False
overwrite_extracted	bool	With extract=True, whether to unpack the archive again even if the extracted directory already exists. Does not download the archive again.	False

Returns:

Type	Description
Path	Path to the downloaded dataset archive, or to the extracted directory if
Path	extract=True.

Raises:

Type	Description
ValueError	If dataset_id is empty, or extract=True and the archive is not a .tar.gz / .tgz file.
FileExistsError	If extract=True and a file, rather than a directory, sits where the archive would be extracted.
FileNotFoundError	If the dataset does not exist (404).
ResourceRemovedError	If the dataset is no longer available (410).
PermissionError	If access is denied (403) or download directory is not writable.
RuntimeError	If rate limit is exceeded (429) or unexpected response format.
HTTPError	For other non-2xx responses.

Source code in src/datacollective/datasets.py

def download_dataset(
    dataset_id: str,
    download_directory: str | None = None,
    show_progress: bool = True,
    overwrite_existing: bool = False,
    enable_logging: bool = False,
    extract: bool = False,
    overwrite_extracted: bool = False,
) -> Path:
    """
    Download the dataset archive to a local directory and return the archive path.
    Skips download if the target file already exists (unless `overwrite_existing=True`).

    Automatically resumes interrupted downloads if a matching .checksum file exists from a
    previous attempt.

    With `extract=True`, also unpacks the archive into a directory beside it, named after
    the archive without its extension, and returns that directory instead. An existing
    directory is reused unless `overwrite_extracted=True`, or the archive itself was
    downloaded again with `overwrite_existing=True`. This is the directory `load_dataset`
    extracts to, so the two never unpack the same archive twice.

    Note: Previously called `save_dataset_to_disk`, which remains available as a
    deprecated alias for backward compatibility.

    Args:
        dataset_id: The dataset ID (as shown in MDC platform) or slug.
        download_directory: Directory where to save the downloaded archive file.
            If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.
        show_progress: Whether to show a progress bar during download.
        overwrite_existing: Whether to overwrite the existing archive file. With
            `extract=True` the extracted directory is then unpacked afresh too.
        enable_logging: Whether to enable SDK logging to console and a local log file.
        extract: Whether to also unpack the archive and return the extracted directory.
        overwrite_extracted: With `extract=True`, whether to unpack the archive again even
            if the extracted directory already exists. Does not download the archive again.

    Returns:
        Path to the downloaded dataset archive, or to the extracted directory if
        `extract=True`.

    Raises:
        ValueError: If dataset_id is empty, or `extract=True` and the archive is not a
            `.tar.gz` / `.tgz` file.
        FileExistsError: If `extract=True` and a file, rather than a directory, sits
            where the archive would be extracted.
        FileNotFoundError: If the dataset does not exist (404).
        ResourceRemovedError: If the dataset is no longer available (410).
        PermissionError: If access is denied (403) or download directory is not writable.
        RuntimeError: If rate limit is exceeded (429) or unexpected response format.
        requests.HTTPError: For other non-2xx responses.
    """
    return _download_archive(
        dataset_id,
        download_directory,
        show_progress,
        overwrite_existing,
        enable_logging,
        download_source=DOWNLOAD_SOURCE_DOWNLOAD,
        extract=extract,
        overwrite_extracted=overwrite_extracted,
    )

download_sample(dataset_id, download_directory=None, show_progress=True, overwrite_existing=False, enable_logging=False, extract=False, overwrite_extracted=False)

Download the dataset's sample archive to a local directory and return the archive path. The sample is a small excerpt of the dataset that its publisher provides for evaluation.

The archive is saved under samples/<dataset_id>/ inside the download directory, so it can never be mistaken for the full dataset archive even when the two share a filename. Skips download if the sample the API currently serves is already there, unless overwrite_existing=True. A sample the publisher has since replaced is downloaded under its new filename, leaving the earlier file in place.

Unlike download_dataset, an interrupted sample download starts again rather than resuming: the API returns no checksum for samples to match a partial file against.

With extract=True, also unpacks the archive into a directory beside it, named after the archive without its extension, and returns that directory instead. An existing directory is reused unless overwrite_extracted=True, or the sample itself was downloaded again with overwrite_existing=True.

Requires an API key whose organization has agreed to the dataset's terms on the MDC platform. Buying the dataset is not required.

Parameters:

Name	Type	Description	Default
dataset_id	str	The dataset ID (as shown in MDC platform) or slug.	required
download_directory	str | None	Directory under which to save the sample archive. If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.	None
show_progress	bool	Whether to show a progress bar during download.	True
overwrite_existing	bool	Whether to download the sample again, replacing the existing one. With extract=True the extracted directory is then unpacked afresh too.	False
enable_logging	bool	Whether to enable SDK logging to console and a local log file.	False
extract	bool	Whether to also unpack the archive and return the extracted directory.	False
overwrite_extracted	bool	With extract=True, whether to unpack the sample again even if the extracted directory already exists. Does not download the sample again.	False

Returns:

Type	Description
Path	Path to the downloaded sample archive, or to the extracted directory if
Path	extract=True.

Raises:

Type	Description
ValueError	If dataset_id is empty, or extract=True and the sample is not a .tar.gz / .tgz file.
FileExistsError	If extract=True and a file, rather than a directory, sits where the sample would be extracted.
FileNotFoundError	If the dataset does not exist, or publishes no sample (404).
ResourceRemovedError	If the dataset is no longer available (410).
PermissionError	If the dataset's terms have not been agreed to or the dataset is private to another organization (403), or download directory is not writable.
RuntimeError	If rate limit is exceeded (429) or unexpected response format.
HTTPError	For other non-2xx responses.

Source code in src/datacollective/datasets.py

def download_sample(
    dataset_id: str,
    download_directory: str | None = None,
    show_progress: bool = True,
    overwrite_existing: bool = False,
    enable_logging: bool = False,
    extract: bool = False,
    overwrite_extracted: bool = False,
) -> Path:
    """
    Download the dataset's sample archive to a local directory and return the archive path.
    The sample is a small excerpt of the dataset that its publisher provides for evaluation.

    The archive is saved under `samples/<dataset_id>/` inside the download directory, so it
    can never be mistaken for the full dataset archive even when the two share a filename.
    Skips download if the sample the API currently serves is already there, unless
    `overwrite_existing=True`. A sample the publisher has since replaced is downloaded under
    its new filename, leaving the earlier file in place.

    Unlike `download_dataset`, an interrupted sample download starts again rather than
    resuming: the API returns no checksum for samples to match a partial file against.

    With `extract=True`, also unpacks the archive into a directory beside it, named after
    the archive without its extension, and returns that directory instead. An existing
    directory is reused unless `overwrite_extracted=True`, or the sample itself was
    downloaded again with `overwrite_existing=True`.

    Requires an API key whose organization has agreed to the dataset's terms on the MDC
    platform. Buying the dataset is not required.

    Args:
        dataset_id: The dataset ID (as shown in MDC platform) or slug.
        download_directory: Directory under which to save the sample archive.
            If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.
        show_progress: Whether to show a progress bar during download.
        overwrite_existing: Whether to download the sample again, replacing the existing one.
            With `extract=True` the extracted directory is then unpacked afresh too.
        enable_logging: Whether to enable SDK logging to console and a local log file.
        extract: Whether to also unpack the archive and return the extracted directory.
        overwrite_extracted: With `extract=True`, whether to unpack the sample again even
            if the extracted directory already exists. Does not download the sample again.

    Returns:
        Path to the downloaded sample archive, or to the extracted directory if
        `extract=True`.

    Raises:
        ValueError: If dataset_id is empty, or `extract=True` and the sample is not a
            `.tar.gz` / `.tgz` file.
        FileExistsError: If `extract=True` and a file, rather than a directory, sits
            where the sample would be extracted.
        FileNotFoundError: If the dataset does not exist, or publishes no sample (404).
        ResourceRemovedError: If the dataset is no longer available (410).
        PermissionError: If the dataset's terms have not been agreed to or the dataset is
            private to another organization (403), or download directory is not writable.
        RuntimeError: If rate limit is exceeded (429) or unexpected response format.
        requests.HTTPError: For other non-2xx responses.
    """
    _enable_logging(enable_logging)
    logger.info(f"Downloading sample of dataset {dataset_id}")

    dataset_details = get_dataset_details(dataset_id)

    archive_path = _download_sample(
        dataset_id=dataset_details.id,
        download_directory=download_directory,
        show_progress=show_progress,
        overwrite_existing=overwrite_existing,
        download_source=DOWNLOAD_SOURCE_SAMPLE,
    )
    if not extract:
        return archive_path
    return _extract_beside(archive_path, overwrite_existing, overwrite_extracted)

get_dataset_details(dataset_id)

Return dataset details from the MDC API.

This is a public endpoint: no API key (MDC_API_KEY) is required and none is sent.

Parameters:

Name	Type	Description	Default
dataset_id	str	The dataset ID (as shown in MDC platform) or slug.	required

Returns:

Type	Description
DatasetDetails	A DatasetDetails model with the dataset details as returned by the API.

Raises:

Type	Description
ValueError	If dataset_id is empty.
FileNotFoundError	If the dataset does not exist (404).
PermissionError	If the dataset is private to another organization (403).
ResourceRemovedError	If the dataset is no longer available (410).
RuntimeError	If rate limit is exceeded (429).
HTTPError	For other non-2xx responses.
ValidationError	If the API response is missing the id field.

Source code in src/datacollective/datasets.py

def get_dataset_details(dataset_id: str) -> DatasetDetails:
    """
    Return dataset details from the MDC API.

    This is a public endpoint: no API key (`MDC_API_KEY`) is required and none is sent.

    Args:
        dataset_id: The dataset ID (as shown in MDC platform) or slug.

    Returns:
        A DatasetDetails model with the dataset details as returned by the API.

    Raises:
        ValueError: If dataset_id is empty.
        FileNotFoundError: If the dataset does not exist (404).
        PermissionError: If the dataset is private to another organization (403).
        ResourceRemovedError: If the dataset is no longer available (410).
        RuntimeError: If rate limit is exceeded (429).
        requests.HTTPError: For other non-2xx responses.
        pydantic.ValidationError: If the API response is missing the `id` field.
    """
    if not dataset_id or not dataset_id.strip():
        raise ValueError("`dataset_id` must be a non-empty string")

    url = f"{_get_api_url()}/datasets/{dataset_id}"
    resp = _send_api_request(method="GET", url=url, include_auth_headers=False)
    return DatasetDetails.model_validate(resp.json())

list_dataset_filters()

Return the filter values that list_datasets currently accepts.

This is a public endpoint: no API key (MDC_API_KEY) is required.

Every value except a task is drawn from the published catalog, so it appears only while some dataset carries it; the task vocabulary is fixed.

Returns:

Type	Description
DatasetFilters	A DatasetFilters model with the available tasks, locales, licenses
DatasetFilters	and formats.

Raises:

Type	Description
RuntimeError	If rate limit is exceeded (429).
HTTPError	For other non-2xx responses.
ValidationError	If the API response does not match the expected shape.

Source code in src/datacollective/datasets.py

def list_dataset_filters() -> DatasetFilters:
    """
    Return the filter values that `list_datasets` currently accepts.

    This is a public endpoint: no API key (`MDC_API_KEY`) is required.

    Every value except a task is drawn from the published catalog, so it appears
    only while some dataset carries it; the task vocabulary is fixed.

    Returns:
        A DatasetFilters model with the available `tasks`, `locales`, `licenses`
        and `formats`.

    Raises:
        RuntimeError: If rate limit is exceeded (429).
        requests.HTTPError: For other non-2xx responses.
        pydantic.ValidationError: If the API response does not match the expected shape.
    """
    url = f"{_get_api_url()}/datasets/filters"
    resp = _send_api_request(method="GET", url=url, include_auth_headers=False)
    return DatasetFilters.model_validate(resp.json())

list_datasets(query=None, *, results_per_page=None, page_number=None, task=None, locale=None, license_abbr=None, file_format=None, sort=None, sort_direction=None, upload_date=None, has_sample=None, pricing=None)

List or search the public dataset catalog of the MDC platform.

This is a public endpoint: no API key (MDC_API_KEY) is required. Results are paginated; use page_number together with the returned total to walk through the whole catalog. The values accepted by the task, locale, license_abbr and file_format filters can be discovered with list_dataset_filters. Filtering on a value absent from those lists matches nothing rather than failing.

Parameters:

Name	Type	Description	Default
query	str | None	Free-text search string (e.g. swahili speech). Matches all datasets when omitted.	None
results_per_page	int | None	Results per page, between 1 and 100. The platform defaults to 24.	None
page_number	int | None	1-based page number. Defaults to the first page.	None
task	Task | str | Sequence[Task | str] | None	ML task(s) to keep, either a Task enum value or its string value (e.g. ASR). Pass a list to match any of several tasks.	None
locale	str | Sequence[str] | None	Language/locale code(s) to keep (e.g. en, de-DE). Pass a list to match any.	None
license_abbr	License | str | Sequence[License | str] | None	License abbreviation(s) to keep, either a License enum value or a string (e.g. CC0-1.0). Pass a list to match any.	None
file_format	str | Sequence[str] | None	File format(s) to keep (e.g. WAV, TSV). Pass a list to match any.	None
sort	Literal['relevance', 'newest', 'size'] | None	Result ordering: relevance, newest or size.	None
sort_direction	Literal['asc', 'desc'] | None	asc or desc, applied to sort.	None
upload_date	Literal['today', 'thisWeek', 'thisMonth', 'thisYear'] | None	Only datasets published within the given window: today, thisWeek, thisMonth or thisYear.	None
has_sample	bool | None	When True, only datasets that provide a sample file.	None
pricing	Literal['compensated', 'free'] | None	free or compensated datasets	None

Returns:

Type	Description
DatasetList	A DatasetList with the matching items for the requested page and the
DatasetList	total number of matches across all pages.

Raises:

Type	Description
ValueError	If an argument is out of range or not one of the accepted options.
RuntimeError	If rate limit is exceeded (429).
HTTPError	For other non-2xx responses (e.g. 400 for a filter value the API rejects).
ValidationError	If the API response does not match the expected shape.

Source code in src/datacollective/datasets.py

def list_datasets(
    query: str | None = None,
    *,
    results_per_page: int | None = None,
    page_number: int | None = None,
    task: Task | str | Sequence[Task | str] | None = None,
    locale: str | Sequence[str] | None = None,
    license_abbr: License | str | Sequence[License | str] | None = None,
    file_format: str | Sequence[str] | None = None,
    sort: Literal["relevance", "newest", "size"] | None = None,
    sort_direction: Literal["asc", "desc"] | None = None,
    upload_date: Literal["today", "thisWeek", "thisMonth", "thisYear"] | None = None,
    has_sample: bool | None = None,
    pricing: Literal["compensated", "free"] | None = None,
) -> DatasetList:
    """
    List or search the public dataset catalog of the MDC platform.

    This is a public endpoint: no API key (`MDC_API_KEY`) is required.
    Results are paginated; use `page_number` together with the returned `total` to walk
    through the whole catalog. The values accepted by the `task`, `locale`,
    `license_abbr` and `file_format` filters can be discovered with `list_dataset_filters`.
    Filtering on a value absent from those lists matches nothing rather than failing.

    Args:
        query: Free-text search string (e.g. `swahili speech`). Matches all datasets when omitted.
        results_per_page: Results per page, between 1 and 100. The platform defaults to 24.
        page_number: 1-based page number. Defaults to the first page.
        task: ML task(s) to keep, either a `Task` enum value or its string value (e.g. `ASR`).
            Pass a list to match any of several tasks.
        locale: Language/locale code(s) to keep (e.g. `en`, `de-DE`). Pass a list to match any.
        license_abbr: License abbreviation(s) to keep, either a `License` enum value or a string
            (e.g. `CC0-1.0`). Pass a list to match any.
        file_format: File format(s) to keep (e.g. `WAV`, `TSV`). Pass a list to match any.
        sort: Result ordering: `relevance`, `newest` or `size`.
        sort_direction: `asc` or `desc`, applied to `sort`.
        upload_date: Only datasets published within the given window:
            `today`, `thisWeek`, `thisMonth` or `thisYear`.
        has_sample: When True, only datasets that provide a sample file.
        pricing: free or compensated datasets

    Returns:
        A DatasetList with the matching `items` for the requested page and the
        `total` number of matches across all pages.

    Raises:
        ValueError: If an argument is out of range or not one of the accepted options.
        RuntimeError: If rate limit is exceeded (429).
        requests.HTTPError: For other non-2xx responses (e.g. 400 for a filter value the API rejects).
        pydantic.ValidationError: If the API response does not match the expected shape.
    """
    if query is not None and not query.strip():
        raise ValueError("`query` must be a non-empty string when provided")
    if results_per_page is not None and not 1 <= results_per_page <= MAX_PAGE_SIZE:
        raise ValueError(
            f"`results_per_page` must be between 1 and {MAX_PAGE_SIZE}, got {results_per_page}"
        )
    if page_number is not None and page_number < 1:
        raise ValueError(
            f"`page_number` must be a positive (1-based) integer, got {page_number}"
        )
    _validate_option("sort", sort, SORT_OPTIONS)
    _validate_option("sort_direction", sort_direction, SORT_DIRECTIONS)
    _validate_option("upload_date", upload_date, UPLOAD_DATE_OPTIONS)
    _validate_option("pricing", pricing, PRICING_OPTIONS)

    params: dict[str, Any] = {
        "q": query.strip() if query is not None else None,
        "limit": results_per_page,
        "page": page_number,
        "task": _normalize_filter_values("task", task),
        "locale": _normalize_filter_values("locale", locale),
        "license": _normalize_filter_values("license", license_abbr),
        "format": _normalize_filter_values("format", file_format),
        "sort": sort,
        "sortDirection": sort_direction,
        "uploadDate": upload_date,
        # The API treats any truthy string as "has a sample"; omitting it means "no filter".
        "sample": "true" if has_sample else None,
        "pricing": pricing,
    }
    params = {key: value for key, value in params.items() if value is not None}

    url = f"{_get_api_url()}/datasets"
    resp = _send_api_request(
        method="GET", url=url, params=params, include_auth_headers=False
    )
    return DatasetList.model_validate(resp.json())

load_dataset(dataset_id, download_directory=None, show_progress=True, overwrite_existing=False, overwrite_extracted=False, enable_logging=False, return_format='pandas')

load_dataset(dataset_id: str, download_directory: str | None = None, show_progress: bool = True, overwrite_existing: bool = False, overwrite_extracted: bool = False, enable_logging: bool = False, return_format: Literal['pandas'] = 'pandas') -> pd.DataFrame

load_dataset(dataset_id: str, download_directory: str | None = None, show_progress: bool = True, overwrite_existing: bool = False, overwrite_extracted: bool = False, enable_logging: bool = False, *, return_format: Literal['hf']) -> Dataset | DatasetDict

Download (if needed), extract (if not already extracted), and load the dataset into memory.

By default, the dataset is returned as a pandas DataFrame. Pass return_format="hf" to get a HuggingFace datasets object instead (requires the optional dependency datacollective[hf]).

If the dataset archive already exists in the download directory, it will not be re-downloaded unless overwrite_existing=True.

If there is a directory with the same name as the archive file without the suffix extension, we assume it has already been extracted, and it will not be re-extracted unless overwrite_extracted=True or the archive was downloaded again with overwrite_existing=True.

Uses the dataset schema to determine the loading strategy.

Automatically resumes interrupted downloads if a .checksum file exists from a previous attempt.

Parameters:

Name	Type	Description	Default
dataset_id	str	The dataset ID (as shown in MDC platform) or slug.	required
download_directory	str | None	Directory where to save the downloaded archive file. If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.	None
show_progress	bool	Whether to show a progress bar during download.	True
overwrite_existing	bool	Whether to overwrite existing archive.	False
overwrite_extracted	bool	Whether to overwrite existing extracted files by re-extracting the archive file. Only makes sense when overwrite_existing is False. Will check in the download directory for existing extracted files with the default naming of the folder.	False
enable_logging	bool	Whether to enable SDK logging to console and a local log file.	False
return_format	Literal['pandas', 'hf']	Format of the returned object. "pandas" (default) returns a pandas DataFrame. "hf" returns a HuggingFace Dataset, or a DatasetDict keyed by split name for datasets with multiple splits.	'pandas'

Returns: A pandas DataFrame with the loaded dataset, or a HuggingFace Dataset / DatasetDict when return_format="hf".

Raises:

Type	Description
ValueError	If dataset_id is empty, schema is unsupported, or return_format is invalid.
MissingDependencyError	If return_format="hf" and the HuggingFace datasets library is not installed.
FileExistsError	If a file, rather than a directory, sits where the archive would be extracted.
FileNotFoundError	If the dataset does not exist (404).
ResourceRemovedError	If the dataset is no longer available (410).
PermissionError	If access is denied (403) or download directory is not writable.
RuntimeError	If rate limit is exceeded (429) or unexpected response format.
HTTPError	For other non-2xx responses.

Source code in src/datacollective/datasets.py

def load_dataset(
    dataset_id: str,
    download_directory: str | None = None,
    show_progress: bool = True,
    overwrite_existing: bool = False,
    overwrite_extracted: bool = False,
    enable_logging: bool = False,
    return_format: Literal["pandas", "hf"] = "pandas",
) -> pd.DataFrame | Dataset | DatasetDict:
    """
    Download (if needed), extract (if not already extracted), and load the dataset into memory.

    By default, the dataset is returned as a pandas DataFrame. Pass `return_format="hf"`
    to get a HuggingFace `datasets` object instead (requires the optional dependency datacollective[hf]).

    If the dataset archive already exists in the download directory, it will not be re-downloaded
    unless `overwrite_existing=True`.

    If there is a directory with the same name as the archive file without the suffix extension, we assume
    it has already been extracted, and it will not be re-extracted unless `overwrite_extracted=True`
    or the archive was downloaded again with `overwrite_existing=True`.

    Uses the dataset schema to determine the loading strategy.

    Automatically resumes interrupted downloads if a .checksum file exists from a
    previous attempt.

    Args:
        dataset_id: The dataset ID (as shown in MDC platform) or slug.
        download_directory: Directory where to save the downloaded archive file.
            If None or empty, falls back to env MDC_DOWNLOAD_PATH or default.
        show_progress: Whether to show a progress bar during download.
        overwrite_existing: Whether to overwrite existing archive.
        overwrite_extracted: Whether to overwrite existing extracted files by re-extracting the archive file.
            Only makes sense when overwrite_existing is False.
            Will check in the download directory for existing extracted files with the default naming of the folder.
        enable_logging: Whether to enable SDK logging to console and a local log file.
        return_format: Format of the returned object. `"pandas"` (default) returns a
            pandas DataFrame. `"hf"` returns a HuggingFace `Dataset`, or a `DatasetDict`
            keyed by split name for datasets with multiple splits.
    Returns:
        A pandas DataFrame with the loaded dataset, or a HuggingFace `Dataset` /
        `DatasetDict` when `return_format="hf"`.

    Raises:
        ValueError: If dataset_id is empty, schema is unsupported, or `return_format`
            is invalid.
        MissingDependencyError: If `return_format="hf"` and the HuggingFace `datasets`
            library is not installed.
        FileExistsError: If a file, rather than a directory, sits where the archive
            would be extracted.
        FileNotFoundError: If the dataset does not exist (404).
        ResourceRemovedError: If the dataset is no longer available (410).
        PermissionError: If access is denied (403) or download directory is not writable.
        RuntimeError: If rate limit is exceeded (429) or unexpected response format.
        requests.HTTPError: For other non-2xx responses.
    """
    if return_format not in RETURN_FORMATS:
        raise ValueError(
            f"Invalid return_format '{return_format}'. "
            f"Supported formats: {', '.join(RETURN_FORMATS)}"
        )
    if return_format == "hf":
        # Raise error here if the optional dependency is missing before any download
        _require_datasets()

    _enable_logging(enable_logging)
    logger.info(f"Loading dataset {dataset_id}")

    dataset_details = get_dataset_details(dataset_id)
    archive_filename = _require_archive_filename(dataset_details)
    _id = dataset_details.id
    archive_checksum = dataset_details.checksum or None

    # try to fetch schema from registry
    schema = _get_dataset_schema(_id)
    if schema is None:
        raise RuntimeError(
            f"Dataset '{_id}' exists but is not supported by load_dataset yet. "
            f"You can download the raw archive with: download_dataset('{_id}'). "
            f"If you are the data owner consider submitting a schema for your dataset via"
            f" the registry: https://mozilla-data-collective.github.io/dataset-schema-registry/"
        )

    archive_path = _download_dataset(
        dataset_id=_id,
        archive_filename=archive_filename,
        download_directory=download_directory,
        show_progress=show_progress,
        overwrite_existing=overwrite_existing,
        download_source=DOWNLOAD_SOURCE_LOAD,
    )

    extract_dir = _extract_beside(archive_path, overwrite_existing, overwrite_extracted)

    schema = _resolve_schema(_id, extract_dir, archive_checksum)
    df = _load_dataset_from_schema(schema, extract_dir)

    if return_format == "hf":
        return _convert_to_hf(df, schema)
    return df

save_dataset_to_disk(dataset_id, download_directory=None, show_progress=True, overwrite_existing=False, enable_logging=False)

Deprecated alias for download_dataset.

Use download_dataset instead. This name is kept for backward compatibility.

Source code in src/datacollective/datasets.py

def save_dataset_to_disk(
    dataset_id: str,
    download_directory: str | None = None,
    show_progress: bool = True,
    overwrite_existing: bool = False,
    enable_logging: bool = False,
) -> Path:
    """
    Deprecated alias for `download_dataset`.

    Use `download_dataset` instead. This name is kept for backward compatibility.
    """
    warnings.warn(
        "`save_dataset_to_disk` is deprecated and will be removed in a future "
        "release. Use `download_dataset` instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return _download_archive(
        dataset_id,
        download_directory,
        show_progress,
        overwrite_existing,
        enable_logging,
        download_source=DOWNLOAD_SOURCE_SAVE,
    )

datacollective.download

datacollective.api_utils

datacollective.submissions

create_submission_draft(submission)

Create a draft dataset submission.

Parameters:

Name	Type	Description	Default
submission	DatasetSubmission	Dataset submission model containing name. In this function only the name field is sent and every other field will be ignored and must be sent instead with update_submission before the submission can be submitted for review.	required

Returns:

Type	Description
dict[str, Any]	The full API response dict (contains a submission key with
dict[str, Any]	the created submission).

Source code in src/datacollective/submissions.py

def create_submission_draft(submission: DatasetSubmission) -> dict[str, Any]:
    """
    Create a draft dataset submission.

    Args:
        submission: Dataset submission model containing `name`. In this function
            only the `name` field is sent and every other field will be ignored and
            must be sent instead with `update_submission` before the submission can
            be submitted for review.

    Returns:
        The full API response dict (contains a ``submission`` key with
        the created submission).
    """
    submission = _ensure_submission_model(submission)
    payload = _payload_for_fields(submission, DRAFT_FIELDS)
    if "name" not in payload:
        raise ValueError("`submission` must include `name`")

    url = f"{_get_api_url()}/submissions"
    resp = _send_api_request("POST", url, json_body=payload)
    return dict(resp.json())

create_submission_with_upload(file_path, submission, state_path=None, enable_logging=False, part_size=DEFAULT_PART_SIZE, sample_file_path=None, sample_state_path=None, max_workers=DEFAULT_MAX_WORKERS)

Single point function to create a submission, update metadata, upload a file, and submit for review. Allows for resuming an upload if interrupted by persisting state to a file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to dataset archive.	required
submission	DatasetSubmission	Dataset submission model with metadata fields.	required
state_path	str | None	Optional path to persist upload state.	None
enable_logging	bool	Whether to enable detailed logging during the process.	False
part_size	int	Multipart part size in bytes. Ignored when resuming an existing upload, which keeps the part size recorded in its state file.	DEFAULT_PART_SIZE
sample_file_path	str | None	Optional path to a sample archive to upload alongside the dataset archive. A sample file is not required to submit a dataset.	None
sample_state_path	str | None	Optional path to persist the sample upload state.	None
max_workers	int	Number of parts uploaded concurrently, for both the dataset archive and the sample file. Use 1 to upload parts sequentially.	DEFAULT_MAX_WORKERS

Source code in src/datacollective/submissions.py

def create_submission_with_upload(
    file_path: str,
    submission: DatasetSubmission,
    state_path: str | None = None,
    enable_logging: bool = False,
    part_size: int = DEFAULT_PART_SIZE,
    sample_file_path: str | None = None,
    sample_state_path: str | None = None,
    max_workers: int = DEFAULT_MAX_WORKERS,
) -> dict[str, Any]:
    """
    Single point function to create a submission, update metadata, upload a file, and submit for review.
    Allows for resuming an upload if interrupted by persisting state to a file.

    Args:
        file_path: Path to dataset archive.
        submission: Dataset submission model with metadata fields.
        state_path: Optional path to persist upload state.
        enable_logging: Whether to enable detailed logging during the process.
        part_size: Multipart part size in bytes. Ignored when resuming an existing upload,
            which keeps the part size recorded in its state file.
        sample_file_path: Optional path to a sample archive to upload alongside the
            dataset archive. A sample file is not required to submit a dataset.
        sample_state_path: Optional path to persist the sample upload state.
        max_workers: Number of parts uploaded concurrently, for both the dataset
            archive and the sample file. Use 1 to upload parts sequentially.
    """
    _enable_logging(enable_logging)

    submission = _ensure_submission_model(submission)

    # Fail fast on a missing sample file, before uploading the dataset archive
    if sample_file_path and not Path(sample_file_path).exists():
        raise FileNotFoundError(f"Sample file not found: `{sample_file_path}`")

    state_file, existing_upload_state = _resolve_upload_state(file_path, state_path)

    if existing_upload_state:
        submission_id = existing_upload_state.submissionId
        logger.info(
            f"Found existing upload state at `{state_file}`. Resuming submission {submission_id}."
        )
    else:
        logger.info(f"Creating submission draft for '{submission.name}'...")

        draft = create_submission_draft(submission)

        submission_payload = draft.get("submission")
        draft_id = (
            submission_payload.get("id")
            if isinstance(submission_payload, dict)
            else None
        )
        if not draft_id:
            raise RuntimeError("Draft creation did not return a submission id")
        submission_id = str(draft_id)

        logger.info(f"Draft created. Submission ID: {submission_id}")

    logger.info("Updating submission metadata...")

    update_submission(submission_id, submission)

    upload_dataset_file(
        file_path=file_path,
        submission_id=submission_id,
        state_path=state_path,
        enable_logging=enable_logging,
        part_size=part_size,
        max_workers=max_workers,
    )

    if sample_file_path:
        logger.info("Uploading sample file...")
        upload_sample_file(
            file_path=sample_file_path,
            submission_id=submission_id,
            state_path=sample_state_path,
            enable_logging=enable_logging,
            part_size=part_size,
            max_workers=max_workers,
        )

    # The uploaded file is linked to the submission automatically when the
    # multipart upload completes (the upload was started with `submissionId`).
    # Completeness of the submission is validated by the platform on submit.
    logger.info("Submitting dataset for review...")

    response = submit_submission(submission_id, submission)

    logger.info("Submission complete!")

    return response

delete_submission(submission_id, enable_logging=False)

Delete a dataset submission that has not been published, and every file uploaded for it. WARNING: The deletion cannot be undone!

Parameters:

Name	Type	Description	Default
submission_id	str	Dataset submission ID.	required
enable_logging	bool	Whether to enable SDK logging to console and a local log file.	False

Raises:

Type	Description
ValueError	If submission_id is empty.
FileNotFoundError	If the submission does not exist (404).
PermissionError	If the submission belongs to another organization or has been published (403).
ResourceRemovedError	If the submission was already deleted (410).
RuntimeError	If rate limit is exceeded (429).
HTTPError	For other non-2xx responses.

Source code in src/datacollective/submissions.py

def delete_submission(submission_id: str, enable_logging: bool = False) -> None:
    """
    Delete a dataset submission that has not been published, and every file
    uploaded for it. WARNING: The deletion cannot be undone!

    Args:
        submission_id: Dataset submission ID.
        enable_logging: Whether to enable SDK logging to console and a local log file.

    Raises:
        ValueError: If submission_id is empty.
        FileNotFoundError: If the submission does not exist (404).
        PermissionError: If the submission belongs to another organization or
            has been published (403).
        ResourceRemovedError: If the submission was already deleted (410).
        RuntimeError: If rate limit is exceeded (429).
        requests.HTTPError: For other non-2xx responses.
    """
    if not submission_id or not submission_id.strip():
        raise ValueError("`submission_id` must be a non-empty string")

    _enable_logging(enable_logging)

    url = f"{_get_api_url()}/submissions/{submission_id}"
    logger.info(f"Deleting submission {submission_id}...")
    try:
        # 204 No Content: there is no body to return.
        _send_api_request("DELETE", url)
    except Exception as exc:
        logger.error(f"Failed to delete submission {submission_id}: {exc}")
        raise
    logger.info(f"Submission {submission_id} deleted.")

submit_submission(submission_id, submission)

Submit a dataset submission for review.

Parameters:

Name	Type	Description	Default
submission_id	str	Dataset submission ID.	required
submission	DatasetSubmission	Dataset submission model with agreeToSubmit=True.	required

Returns:

Type	Description
dict[str, Any]	The full API response dict (contains a submission key with
dict[str, Any]	the submission whose status should be "submitted").

Source code in src/datacollective/submissions.py

def submit_submission(
    submission_id: str,
    submission: DatasetSubmission,
) -> dict[str, Any]:
    """
    Submit a dataset submission for review.

    Args:
        submission_id: Dataset submission ID.
        submission: Dataset submission model with `agreeToSubmit=True`.

    Returns:
        The full API response dict (contains a ``submission`` key with
        the submission whose status should be ``"submitted"``).
    """
    submission = _ensure_submission_model(submission)

    if submission.agreeToSubmit is not True:
        raise ValueError("`agreeToSubmit` must be True to submit a dataset")

    payload = _payload_for_fields(submission, SUBMIT_FIELDS)
    url = f"{_get_api_url()}/submissions/{submission_id}"
    resp = _send_api_request("POST", url, json_body=payload)
    return dict(resp.json())

update_submission(submission_id, submission)

Update metadata on an existing dataset submission.

Parameters:

Name	Type	Description	Default
submission_id	str	Dataset submission ID.	required
submission	DatasetSubmission	Dataset submission model containing update fields.	required

Returns:

Type	Description
dict[str, Any]	The full API response dict (contains a submission key).

Source code in src/datacollective/submissions.py

def update_submission(
    submission_id: str,
    submission: DatasetSubmission,
) -> dict[str, Any]:
    """
    Update metadata on an existing dataset submission.

    Args:
        submission_id: Dataset submission ID.
        submission: Dataset submission model containing update fields.

    Returns:
        The full API response dict (contains a ``submission`` key).
    """
    submission = _ensure_submission_model(submission)

    payload = _payload_for_fields(submission, UPDATE_FIELDS)
    if not payload:
        raise ValueError("`submission` must include at least one updatable field")

    url = f"{_get_api_url()}/submissions/{submission_id}"
    resp = _send_api_request("PATCH", url, json_body=payload)
    return dict(resp.json())

datacollective.upload

upload_dataset_file(file_path, submission_id, state_path=None, show_progress=True, enable_logging=False, part_size=DEFAULT_PART_SIZE, max_workers=DEFAULT_MAX_WORKERS)

Upload a dataset file using multipart uploads with resumable state.

Uploads use the application/gzip MIME type. Pass the submission ID of the target dataset submission. This works for both draft submissions and for uploading a new .tar.gz version to an already approved dataset submission.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the dataset archive on disk.	required
submission_id	str	Dataset submission ID (not the dataset ID).	required
state_path	str | None	Optional path to persist upload state. Defaults to <filename>.mdc-upload.json alongside the archive.	None
enable_logging	bool	Whether to enable detailed logging during the upload.	False
show_progress	bool	Whether to show a progress bar during upload.	True
part_size	int	Multipart part size in bytes. Ignored when resuming an existing upload, which keeps the part size recorded in its state file.	DEFAULT_PART_SIZE
max_workers	int	Number of parts uploaded concurrently. The file is still read and hashed in order; about max_workers + 1 parts are held in memory. Use 1 to upload parts one at a time.	DEFAULT_MAX_WORKERS

Source code in src/datacollective/upload.py

def upload_dataset_file(
    file_path: str,
    submission_id: str,
    state_path: str | None = None,
    show_progress: bool = True,
    enable_logging: bool = False,
    part_size: int = DEFAULT_PART_SIZE,
    max_workers: int = DEFAULT_MAX_WORKERS,
) -> UploadState:
    """
    Upload a dataset file using multipart uploads with resumable state.

    Uploads use the `application/gzip` MIME type.
    Pass the submission ID of the target dataset submission. This works for
    both draft submissions and for uploading a new `.tar.gz` version to an
    already approved dataset submission.

    Args:
        file_path: Path to the dataset archive on disk.
        submission_id: Dataset submission ID (not the dataset ID).
        state_path: Optional path to persist upload state. Defaults to
            `<filename>.mdc-upload.json` alongside the archive.
        enable_logging: Whether to enable detailed logging during the upload.
        show_progress: Whether to show a progress bar during upload.
        part_size: Multipart part size in bytes. Ignored when resuming an
            existing upload, which keeps the part size recorded in its state file.
        max_workers: Number of parts uploaded concurrently. The file is still
            read and hashed in order; about `max_workers + 1` parts are held in
            memory. Use 1 to upload parts one at a time.
    """
    return _upload_file(
        file_path=file_path,
        submission_id=submission_id,
        state_path=state_path,
        show_progress=show_progress,
        enable_logging=enable_logging,
        part_size=part_size,
        max_workers=max_workers,
        is_sample=False,
    )

upload_sample_file(file_path, submission_id, state_path=None, show_progress=True, enable_logging=False, part_size=DEFAULT_PART_SIZE, max_workers=DEFAULT_MAX_WORKERS)

Upload an optional sample file for a dataset submission.

A sample file is a small, representative excerpt of the dataset that users can inspect without downloading the full archive. It is uploaded exactly like the dataset archive (resumable multipart upload, application/gzip MIME type) but through the submission's sample endpoints, and it does not replace the dataset file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the sample archive on disk.	required
submission_id	str	Dataset submission ID (not the dataset ID).	required
state_path	str | None	Optional path to persist upload state. Defaults to <filename>.mdc-sample-upload.json alongside the archive.	None
enable_logging	bool	Whether to enable detailed logging during the upload.	False
show_progress	bool	Whether to show a progress bar during upload.	True
part_size	int	Multipart part size in bytes. Ignored when resuming an existing upload, which keeps the part size recorded in its state file.	DEFAULT_PART_SIZE
max_workers	int	Number of parts uploaded concurrently. The file is still read and hashed in order; about max_workers + 1 parts are held in memory. Use 1 to upload parts one at a time.	DEFAULT_MAX_WORKERS

Source code in src/datacollective/upload.py

def upload_sample_file(
    file_path: str,
    submission_id: str,
    state_path: str | None = None,
    show_progress: bool = True,
    enable_logging: bool = False,
    part_size: int = DEFAULT_PART_SIZE,
    max_workers: int = DEFAULT_MAX_WORKERS,
) -> UploadState:
    """
    Upload an **optional** sample file for a dataset submission.

    A sample file is a small, representative excerpt of the dataset that
    users can inspect without downloading the full archive. It is uploaded
    exactly like the dataset archive (resumable multipart upload,
    `application/gzip` MIME type) but through the submission's sample endpoints,
    and it does not replace the dataset file.

    Args:
        file_path: Path to the sample archive on disk.
        submission_id: Dataset submission ID (not the dataset ID).
        state_path: Optional path to persist upload state. Defaults to
            `<filename>.mdc-sample-upload.json` alongside the archive.
        enable_logging: Whether to enable detailed logging during the upload.
        show_progress: Whether to show a progress bar during upload.
        part_size: Multipart part size in bytes. Ignored when resuming an
            existing upload, which keeps the part size recorded in its state file.
        max_workers: Number of parts uploaded concurrently. The file is still
            read and hashed in order; about `max_workers + 1` parts are held in
            memory. Use 1 to upload parts one at a time.
    """
    return _upload_file(
        file_path=file_path,
        submission_id=submission_id,
        state_path=state_path,
        show_progress=show_progress,
        enable_logging=enable_logging,
        part_size=part_size,
        max_workers=max_workers,
        is_sample=True,
    )

datacollective.models

Dataset

Bases: BaseModel

Dataset fields shared by the platform's dataset and dataset-submission API payloads. Only fields present in both payloads belong here.

DatasetDetails inherits this class and is tolerant to new fields that are not declared here in order to prevent breaking changes when the API returns new fields. DatasetSubmission inherits this class, adds the submission-only datasheet fields and overrides the enum-like fields with strict types for validation.

Note: Fields are camelCase to match the API payloads.

Source code in src/datacollective/models.py

class Dataset(BaseModel):
    """
    Dataset fields shared by the platform's dataset and dataset-submission
    API payloads. Only fields present in *both* payloads belong here.

    DatasetDetails inherits this class and is tolerant to new fields that are
    not declared here in order to prevent breaking changes when the API returns new fields.
    DatasetSubmission inherits this class, adds the submission-only datasheet
    fields and overrides the enum-like fields with strict types for validation.

    Note: Fields are camelCase to match the API payloads.
    """

    name: str | None = Field(None, description="Name of the dataset.")
    shortDescription: str | None = Field(
        None, description="Brief description of the dataset."
    )
    longDescription: str | None = Field(
        None, description="Detailed description of the dataset."
    )
    locale: str | None = Field(
        None, description="Language/locale code (e.g., `en-US`, `de-DE`)."
    )
    task: str | None = Field(None, description="ML task type.")
    format: str | None = Field(None, description="File format (e.g., `TSV`, `WAV`).")
    licenseAbbreviation: str | None = Field(
        None, description="Abbreviated license name."
    )
    license: str | None = Field(
        None,
        description="Full license name for custom licenses.",
    )
    isPaid: bool | None = Field(
        None,
        description="Whether the dataset is compensated, i.e. has a price. Defaults to `False` on the platform when left unset.",
    )
    basePriceCents: int | None = Field(
        None,
        description=(
            "Price of the dataset in USD cents (e.g. `100_000` is $1,000.00). Required when "
            "`isPaid` is True. The platform validates that the price is within an acceptable "
            "range and rejects the submission otherwise."
        ),
    )
    # Defined by the API and not user-editable
    id: str | None = Field(
        None, description="Unique identifier as returned by the API."
    )
    slug: str | None = Field(
        None,
        description="URL-friendly slug generated from the name. Determined by the API.",
    )
    createdAt: str | None = Field(
        None,
        description="Timestamp when the record was created. Set by the API upon creation.",
    )

DatasetDetails

Bases: Dataset

Dataset details as returned by the MDC API (read model).

Returned by get_dataset_details and, one per entry, by list_datasets: the platform serves the same dataset payload from both endpoints.

Tolerant of platform schema changes by design: fields the API adds are kept as extra attributes, fields the API removes simply read as None, and enum-like fields (task, visibility) are plain strings so new platform values don't fail validation. Only id is required.

Dict-style access (details["id"], details.get("checksum")) is supported for backward compatibility with the previous dict return type.

Source code in src/datacollective/models.py

class DatasetDetails(Dataset):
    """
    Dataset details as returned by the MDC API (read model).

    Returned by `get_dataset_details` and, one per entry, by `list_datasets`:
    the platform serves the same dataset payload from both endpoints.

    Tolerant of platform schema changes by design: fields the API adds are
    kept as extra attributes, fields the API removes simply read as None,
    and enum-like fields (`task`, `visibility`) are plain strings so new
    platform values don't fail validation. Only `id` is required.

    Dict-style access (`details["id"]`, `details.get("checksum")`) is
    supported for backward compatibility with the previous dict return type.
    """

    model_config = ConfigDict(extra="allow")

    id: str = Field(..., description="Unique identifier of the dataset.")
    filename: str | None = Field(
        None, description="Archive filename of the current dataset file version."
    )
    checksum: str | None = Field(
        None, description="Checksum of the current dataset file version."
    )
    sizeBytes: int | float | None = Field(
        None, description="Size of the dataset archive in bytes."
    )
    organization: DatasetOrganization | None = Field(
        None, description="Organization that published the dataset."
    )
    pricing: DatasetPricing | None = Field(
        None,
        description="Pricing information. Prefer this over the top-level `isPaid`/`basePriceCents`, which the API has deprecated.",
    )
    datasetUrl: str | None = Field(
        None, description="URL of the dataset page on the MDC platform."
    )
    submissionId: str | None = Field(
        None,
        description="Identifier of the underlying submission. Only present for datasets owned by the caller.",
    )

    def __contains__(self, key: object) -> bool:
        # Mirrors the previous dict semantics: only keys the API actually
        # returned are "present", even though declared fields always exist
        # as attributes (defaulting to None).
        return isinstance(key, str) and key in self.model_fields_set

    def __getitem__(self, key: str) -> Any:
        if key not in self:
            raise KeyError(key)
        return getattr(self, key)

    def get(self, key: str, default: Any = None) -> Any:
        try:
            return self[key]
        except KeyError:
            return default

DatasetFilters

Bases: BaseModel

The values list_datasets can currently be narrowed by, as returned by list_dataset_filters.

Every value except a task is drawn from the published catalog, so it appears only while some dataset carries it; the task vocabulary is fixed.

Source code in src/datacollective/models.py

class DatasetFilters(BaseModel):
    """
    The values `list_datasets` can currently be narrowed by, as returned by
    `list_dataset_filters`.

    Every value except a task is drawn from the published catalog, so it appears
    only while some dataset carries it; the task vocabulary is fixed.
    """

    model_config = ConfigDict(extra="allow")

    tasks: list[str] = Field(
        default_factory=list, description="Available `task` values."
    )
    locales: list[str] = Field(
        default_factory=list, description="Available `locale` values."
    )
    licenses: list[str] = Field(
        default_factory=list, description="Available `license` (abbreviation) values."
    )
    formats: list[str] = Field(
        default_factory=list, description="Available `format` values."
    )

DatasetList

Bases: BaseModel

One page of the public dataset catalog, as returned by list_datasets.

Supports len(), iteration and indexing over items for convenience.

Source code in src/datacollective/models.py

class DatasetList(BaseModel):
    """
    One page of the public dataset catalog, as returned by `list_datasets`.

    Supports `len()`, iteration and indexing over `items` for convenience.
    """

    model_config = ConfigDict(extra="allow")

    items: list[DatasetDetails] = Field(
        default_factory=list, description="Datasets on this page."
    )
    total: int = Field(
        ...,
        description="Number of matching datasets across the whole result set, not just this page.",
    )

    def __len__(self) -> int:
        return len(self.items)

    def __iter__(self) -> Iterator[DatasetDetails]:  # type: ignore[override]
        return iter(self.items)

    def __getitem__(self, index: int) -> DatasetDetails:
        return self.items[index]

DatasetOrganization

Bases: BaseModel

Organization that published a dataset.

Source code in src/datacollective/models.py

class DatasetOrganization(BaseModel):
    """Organization that published a dataset."""

    model_config = ConfigDict(extra="allow")

    name: str | None = Field(None, description="Display name of the organization.")
    slug: str | None = Field(None, description="URL-friendly slug of the organization.")

DatasetPricing

Bases: BaseModel

Pricing of a dataset. All price fields are None for free datasets.

Source code in src/datacollective/models.py

class DatasetPricing(BaseModel):
    """Pricing of a dataset. All price fields are `None` for free datasets."""

    model_config = ConfigDict(extra="allow")

    isPaid: bool | None = Field(
        None, description="Whether the dataset is compensated (has a price)."
    )
    basePriceCents: int | None = Field(
        None, description="Base price set by the provider, in USD cents."
    )
    currency: str | None = Field(None, description="Currency code (`usd`).")
    platformFeeRate: str | None = Field(
        None,
        description="Platform fee rate applied on top of the base price (e.g. `0.05`).",
    )
    platformFeeCents: int | None = Field(None, description="Platform fee in USD cents.")
    totalPriceCents: int | None = Field(
        None,
        description="Total price paid by the buyer (base + platform fee), in USD cents.",
    )

DatasetSubmission

Bases: NonEmptyStrModel, Dataset

DatasetSubmission schema aligned with the DB representation used for draft creation, metadata updates, and final submission.

Shared datasheet fields come from Dataset. This model adds the fields that only exist on submissions and overrides the enum-like ones with strict types so user input is validated before it is sent to the API.

Source code in src/datacollective/models.py

class DatasetSubmission(NonEmptyStrModel, Dataset):
    """
    DatasetSubmission schema aligned with the DB representation used
    for draft creation, metadata updates, and final submission.

    Shared datasheet fields come from Dataset. This model adds the fields
    that only exist on submissions and overrides the enum-like ones with
    strict types so user input is validated before it is sent to the API.
    """

    task: Task | None = Field(
        None,
        description="ML task type — must be one of the Task enum values listed in api.md.",
    )
    licenseAbbreviation: License | str | None = Field(
        None,
        description="Either one of the predefined License enum values or, optionally, a custom abbreviated license name.",
    )
    visibility: Visibility | None = Field(
        None,
        description="Dataset visibility: `public`, `private`, or `restricted`.",
    )
    # Submission-specific datasheet fields defined by the user
    licenseUrl: str | None = Field(
        None,
        description="Optional URL to the license text for custom licenses.",
    )
    other: str | None = Field(None, description="The datasheet of the dataset.")
    restrictions: str | None = Field(
        None, description="Any restrictions on dataset use. Supports Markdown."
    )
    forbiddenUsage: str | None = Field(
        None, description="Explicitly forbidden use cases. Supports Markdown."
    )
    additionalConditions: str | None = Field(
        None, description="Additional conditions for use. Supports Markdown."
    )
    pointOfContactFullName: str | None = Field(
        None, description="Primary contact name."
    )
    pointOfContactEmail: str | None = Field(None, description="Primary contact email.")
    fundedByFullName: str | None = Field(None, description="Funder's name.")
    fundedByEmail: str | None = Field(None, description="Funder's email.")
    legalContactFullName: str | None = Field(None, description="Legal contact name.")
    legalContactEmail: str | None = Field(None, description="Legal contact email.")
    intendedUsage: str | None = Field(None, description="Intended use of the dataset.")
    ethicalReviewProcess: str | None = Field(
        None, description="Description of ethical review conducted."
    )
    showContactInfo: bool | None = Field(
        None,
        description="Whether to publicly display the dataset contact information.",
    )
    showComplianceAttributes: bool | None = Field(
        None,
        description="Whether to publicly display the dataset's compliance attributes.",
    )
    # Submission-specific fields defined by the user
    createdByFullName: str | None = Field(None, description="Creator's name.")
    createdByEmail: str | None = Field(None, description="Creator's email.")
    exclusivityOptOut: bool | None = Field(
        None,
        description="True if dataset is non-exclusive; False if hosted exclusively on Mozilla Data Collective (see https://mozilladatacollective.com/terms/providers#appendix-1).",
    )
    agreeToSubmit: bool | None = Field(
        None,
        description="You confirm that you have the right to submit this dataset and that all information provided in the datasheet is accurate. Required to be True to complete the submission process",
    )
    autoApproveAccessRequests: bool | None = Field(
        None,
        description=(
            "Only applies to `restricted` datasets. When True, the platform grants every "
            "access request automatically as soon as it is made, instead of leaving it "
            "pending for you to review; the requester's email is shared with you and you "
            "can still revoke access later. Defaults to False on the platform when left unset."
        ),
    )
    # Submission-specific fields defined by the API and not user-editable
    organizationId: str | None = Field(
        None,
        description="Identifier for the organization that owns the dataset.",
    )
    createdBy: str | None = Field(
        None, description="Identifier for the user who created the submission."
    )
    status: str | None = Field(
        None,
        description="Current status of the submission (e.g., 'draft', 'submitted'). Determined by the API.",
    )
    fileUploadId: str | None = Field(
        None,
        description="Identifier for the associated file upload, if any. Generated by the API when a file is uploaded.",
    )
    sampleFileReferenceId: str | None = Field(
        None,
        description="Identifier for the associated sample file, if any. Generated by the API when a sample file is uploaded.",
    )
    submittedAt: str | None = Field(
        None,
        description="Timestamp when the submission was finalized and submitted. Set by the API upon submission.",
    )

    updatedAt: str | None = Field(
        None,
        description="Timestamp when the record was last updated. Updated by the API on changes.",
    )

    @computed_field(  # type: ignore[prop-decorator]
        description="Currency for `basePriceCents`. Always `usd`, the only currency the platform currently supports."
    )
    @property
    def currency(self) -> str | None:
        return "usd" if self.isPaid else None

License

Bases: str, Enum

List of pre-defined dataset licenses.

Source code in src/datacollective/models.py

class License(str, Enum):
    """List of pre-defined dataset licenses."""

    APACHE_2_0 = "Apache-2.0"
    BSD_3_CLAUSE = "BSD-3-Clause"
    CC_BY_4_0 = "CC-BY-4.0"
    CC_BY_ND_4_0 = "CC-BY-ND-4.0"
    CC_BY_NC_4_0 = "CC-BY-NC-4.0"
    CC_BY_NC_SA_4_0 = "CC-BY-NC-SA-4.0"
    CC_BY_SA_4_0 = "CC-BY-SA-4.0"
    CC_SA_1_0 = "CC-SA-1.0"
    CC0_1_0 = "CC0-1.0"
    EUPL_1_2 = "EUPL-1.2"
    AGPL_3_0 = "AGPL-3.0"
    GFDL_1_3 = "GFDL-1.3"
    GPL_3_0 = "GPL-3.0"
    LGPLLR = "LGPLLR"
    MIT = "MIT"
    MPL_2_0 = "MPL-2.0"
    NLOD_2_0 = "NLOD-2.0"
    NOODL_1_0 = "NOODL-1.0"
    ODC_BY_1_0 = "ODC-By-1.0"
    ODBL_1_0 = "ODbL-1.0"
    OGL_CANADA_2_0 = "OGL-Canada-2.0"
    OGL_UK_3_0 = "OGL-UK-3.0"
    OPUBL_1_0 = "OPUBL-1.0"
    OGDL_TAIWAN_1_0 = "OGDL-Taiwan-1.0"
    UNLICENSE = "Unlicense"

NonEmptyStrModel

Bases: BaseModel

Base model that trims string fields and rejects empty values.

Source code in src/datacollective/models.py

class NonEmptyStrModel(BaseModel):
    """Base model that trims string fields and rejects empty values."""

    model_config = ConfigDict(extra="forbid")
    _allow_empty_trimmed_strings: ClassVar[frozenset[str]] = frozenset(
        {"licenseAbbreviation"}
    )

    @field_validator("*", mode="before")
    @classmethod
    def _non_empty_strings(cls, value: Any, info: Any) -> Any:
        if value is None:
            return value
        if isinstance(value, Enum):
            return value
        if isinstance(value, str):
            trimmed = value.strip()
            if not trimmed:
                if info.field_name in cls._allow_empty_trimmed_strings:
                    return trimmed
                raise ValueError(f"`{info.field_name}` must be a non-empty string")
            return trimmed
        return value

Task

Bases: str, Enum

Valid ML task types for a dataset submission.

Source code in src/datacollective/models.py

class Task(str, Enum):
    """Valid ML task types for a dataset submission."""

    NA = "N/A"
    NLP = "NLP"
    ASR = "ASR"
    LID = "LID"
    TTS = "TTS"
    MT = "MT"
    LM = "LM"
    LLM = "LLM"
    NLU = "NLU"
    NLG = "NLG"
    CALL = "CALL"
    RAG = "RAG"
    CV = "CV"
    ML = "ML"
    OTH = "OTH"

UploadPart

Bases: BaseModel

A single multipart upload part.

Source code in src/datacollective/models.py

class UploadPart(BaseModel):
    """A single multipart upload part."""

    model_config = ConfigDict(extra="forbid")

    partNumber: int = Field(..., ge=1)
    etag: str

Visibility

Bases: str, Enum

Dataset visibility levels.

• PUBLIC: visible to everyone, downloadable by everyone.
• RESTRICTED: visible to everyone, downloadable by your organization and approved requesters (see autoApproveAccessRequests).
• PRIVATE: visible only to your organization, downloadable by your organization via the SDK.

Source code in src/datacollective/models.py

class Visibility(str, Enum):
    """
    Dataset visibility levels.

    - ``PUBLIC``: visible to everyone, downloadable by everyone.
    - ``RESTRICTED``: visible to everyone, downloadable by your organization and
      approved requesters (see ``autoApproveAccessRequests``).
    - ``PRIVATE``: visible only to your organization, downloadable by your
      organization via the SDK.
    """

    PUBLIC = "public"
    PRIVATE = "private"
    RESTRICTED = "restricted"

datacollective.errors

AuthenticationError

Bases: RuntimeError

Raised when the MDC API responds with HTTP 401.

Source code in src/datacollective/errors.py

class AuthenticationError(RuntimeError):
    """Raised when the MDC API responds with HTTP 401."""

DataLoadWarning

Bases: UserWarning

Emitted when values are lost or unresolved while loading a dataset (e.g. unresolvable file paths, unparseable numeric cells, skipped sidecar files).

Issued via :mod:warnings, so it is shown even when package logging is disabled (enable_logging=False).

Source code in src/datacollective/errors.py

class DataLoadWarning(UserWarning):
    """Emitted when values are lost or unresolved while loading a dataset
    (e.g. unresolvable file paths, unparseable numeric cells, skipped
    sidecar files).

    Issued via :mod:`warnings`, so it is shown even when package logging is
    disabled (``enable_logging=False``).
    """

DownloadError

Bases: Exception

Exception raised when a download fails.

Source code in src/datacollective/errors.py

class DownloadError(Exception):
    """Exception raised when a download fails."""

    def __init__(
        self,
        session_bytes: int,
        total_downloaded_bytes: int,
        total_archive_bytes: int,
        checksum: str | None,
    ):
        remaining_bytes = total_archive_bytes - total_downloaded_bytes
        self.remaining_percentage = round(
            (
                (remaining_bytes / total_archive_bytes) * 100
                if total_archive_bytes > 0
                else 0
            ),
            2,
        )
        self.remaining_bytes = _format_bytes(remaining_bytes)
        self.session_bytes = _format_bytes(session_bytes)
        self.total_downloaded_bytes = _format_bytes(total_downloaded_bytes)
        self.total_archive_bytes = _format_bytes(total_archive_bytes)
        self.checksum = checksum
        super().__init__()

    def __str__(self) -> str:
        if self.checksum:
            return f"""Download failed with {self.session_bytes} bytes written in this session.
            You can try downloading again to resume the session.
            Dataset Download Details:
            - Bytes downloaded in this session: {self.session_bytes}
            - Bytes downloaded in total: {self.total_downloaded_bytes}
            - Bytes remaining: {self.remaining_bytes} ({self.remaining_percentage}% remaining)
            - Total archive size: {self.total_archive_bytes}
            """
        return "Download failed. Unfortunately this dataset does not support resuming downloads — please try again."

MissingDependencyError

Bases: ImportError

Raised when an optional dependency required for a feature is not installed.

Source code in src/datacollective/errors.py

class MissingDependencyError(ImportError):
    """Raised when an optional dependency required for a feature is not installed."""

RateLimitError

Bases: RuntimeError

Raised when the MDC API responds with HTTP 429.

Source code in src/datacollective/errors.py

class RateLimitError(RuntimeError):
    """Raised when the MDC API responds with HTTP 429."""

    def __init__(
        self,
        *,
        response: requests.Response | None = None,
        hint: str | None = None,
    ) -> None:
        self.response = response
        message = "Rate limit exceeded. Please try again later."
        if hint:
            message = f"{message} {hint}"
        super().__init__(message)

ResourceRemovedError

Bases: FileNotFoundError

Raised when the MDC API responds with HTTP 410.

The dataset is no longer available, or the submission was deleted. Subclasses :class:FileNotFoundError so callers handling 404 also handle it.

Source code in src/datacollective/errors.py

class ResourceRemovedError(FileNotFoundError):
    """Raised when the MDC API responds with HTTP 410.

    The dataset is no longer available, or the submission was deleted. Subclasses
    :class:`FileNotFoundError` so callers handling 404 also handle it.
    """

SchemaValidationWarning

Bases: UserWarning

Emitted when a schema contains unknown keys (kept under extra).

Issued via :mod:warnings, so it is shown even when package logging is disabled (enable_logging=False).

Source code in src/datacollective/errors.py

class SchemaValidationWarning(UserWarning):
    """Emitted when a schema contains unknown keys (kept under ``extra``).

    Issued via :mod:`warnings`, so it is shown even when package logging is
    disabled (``enable_logging=False``).
    """

TaskValidationWarning

Bases: UserWarning

Emitted when a loaded dataset does not satisfy its task's column contract.

Issued via :mod:warnings, so it is shown even when package logging is disabled (enable_logging=False).

Source code in src/datacollective/errors.py

class TaskValidationWarning(UserWarning):
    """Emitted when a loaded dataset does not satisfy its task's column contract.

    Issued via :mod:`warnings`, so it is shown even when package logging is
    disabled (``enable_logging=False``).
    """

datacollective.archive_utils

datacollective.hf_utils

datacollective.schema

ColumnMapping

Bases: BaseModel

A single column mapping entry inside a schema.

Used by index-based tasks to describe how columns in the index file map to logical fields and their data types.

Unknown keys and unknown dtype values are rejected at parse time.

Source code in src/datacollective/schema.py

class ColumnMapping(BaseModel):
    """
    A single column mapping entry inside a schema.

    Used by index-based tasks to describe how columns in the
    index file map to logical fields and their data types.

    Unknown keys and unknown ``dtype`` values are rejected at parse time.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    source_column: str | int = Field(
        description="column name (str) or positional index (int) for headerless files"
    )
    dtype: Literal[
        "string", "file_path", "file_content", "category", "int", "float"
    ] = "string"
    optional: bool = False
    path_match_strategy: Literal["direct", "exact", "contains"] = "direct"
    file_extension: str | None = Field(
        default=None,
        description=(
            'optional extension used when resolving file_path columns (e.g. ".wav")'
        ),
    )
    path_template: str | None = Field(
        default=None,
        description=(
            "optional template used to construct file_path values from one or more "
            "metadata columns. Supports relative paths and ${value}, e.g. "
            '"${Speaker ID}_khm_${Sentence ID}.wav" or "${Split}/${value}.wav"'
        ),
    )

DatasetSchema

Bases: BaseModel

Task-agnostic representation of a dataset schema, as defined by a schema.yaml file.

Every schema must have dataset_id and, to be loadable, a root_strategy. The remaining fields depend on the strategy; the loader registered for that strategy decides which fields are required at load time.

task is optional. When set to a task with a known contract (e.g. ASR, TTS), the loaded DataFrame is validated to contain the task's required logical columns.

Source code in src/datacollective/schema.py

class DatasetSchema(BaseModel):
    """
    Task-agnostic representation of a dataset schema, as defined by a ``schema.yaml`` file.

    Every schema **must** have ``dataset_id`` and, to be loadable, a
    ``root_strategy``.  The remaining fields depend on the strategy; the
    loader registered for that strategy decides which fields are required
    at load time.

    ``task`` is optional.  When set to a task with a known contract (e.g. ASR,
    TTS), the loaded DataFrame is validated to contain the task's required
    logical columns.
    """

    model_config = ConfigDict(frozen=False)

    dataset_id: str = Field(
        description="Unique identifier for the dataset in the registry"
    )
    task: str | None = Field(
        default=None,
        description=(
            "Optional task as defined in the MDC Platform e.g. ASR, TTS etc. "
            "When set to a task with a known contract, the loaded dataset is "
            "validated against it."
        ),
    )

    # --- Index-based strategy (ASR / TTS) ---
    format: str | None = Field(
        default=None,
        description=(
            'optional format hint (e.g. "csv", "tsv", "pipe"); '
            "inferred from the index file when omitted"
        ),
    )
    index_file: str | None = Field(default=None, description='e.g. "train.csv"')
    base_audio_path: str | list[str] | None = Field(
        default=None,
        description=(
            'e.g. "clips/" or ["clips/", "wavs/"]". Entries may also use '
            'metadata placeholders such as "${Split}/clips/".'
        ),
    )
    columns: dict[str, ColumnMapping] = Field(
        default_factory=dict, description="Mapping of index columns to logical fields"
    )
    separator: str | None = Field(
        default=None, description='explicit separator override (e.g. "|")'
    )
    has_header: bool = Field(
        default=True, description="whether the index file has a header row"
    )
    encoding: str = Field(
        default="utf-8", description='file encoding (e.g. "utf-8-sig" for BOM)'
    )
    quoting: Literal["minimal", "none"] | None = Field(
        default=None,
        description=(
            'how \'"\' is treated in delimited files: "minimal" parses quoted '
            'fields (CSV-style), "none" reads \'"\' as a literal character. '
            'When omitted, tab-separated files use "none" and all others '
            '"minimal"'
        ),
    )
    strict: bool = Field(
        default=False,
        description=(
            "Disable archive heuristics for deterministic loading: "
            "'index_file' must exist at its literal path relative to the "
            "dataset root (no recursive search) and source column names "
            "must match exactly (no fuzzy matching)."
        ),
    )

    # --- Loading strategy ---
    root_strategy: str | None = Field(
        default=None,
        description=(
            'Loading strategy; required to load a dataset: "index" | "glob" | '
            '"paired_glob" | "multi_split" | "multi_sections"'
        ),
    )
    file_pattern: str | None = Field(default=None, description='e.g. "**/*.txt"')
    audio_extension: str | None = Field(
        default=None, description='for paired-file TTS: e.g. ".webm"'
    )
    record_path: str | None = Field(
        default=None,
        description=(
            "for JSON sidecar files: top-level key holding the list of records "
            '(e.g. "transcriptions"); one DataFrame row per record'
        ),
    )

    # --- Multi-split strategy (e.g. Common Voice) ---
    splits: list[str] | None = Field(
        default=None, description='split names to load, e.g. ["train", "dev", "test"]'
    )
    splits_file_pattern: str | None = Field(
        default=None, description='glob pattern for split files, e.g. "**/*.tsv"'
    )
    # --- Multi-section strategy
    sections: list[str] | None = None
    section_root: str | None = None

    # --- Inner archive extraction ---
    extract_files: list[str] | None = Field(
        default=None,
        description=(
            "List of archive paths (relative to dataset root) that must be "
            "extracted before loading, e.g. ['Train.tar.gz', 'Dev.tar.gz']"
        ),
    )

    # --- Schema versioning ---
    checksum: str | None = Field(
        default=None, description="archive checksum for cache validation"
    )

    # --- Catch-all for future / unknown keys ---
    extra: dict[str, Any] = Field(
        default_factory=dict, description="Catch-all for future / unknown keys"
    )

    @model_validator(mode="before")
    @classmethod
    def _route_unknown_keys(cls, data: Any) -> Any:
        """Route unknown top-level keys into ``extra``, warning about each.

        Near-miss typos get a "did you mean" hint. The keys are preserved
        under ``extra`` (not dropped) so that schemas written for newer SDK
        versions keep round-tripping.
        """
        if not isinstance(data, dict):
            return data
        if not data.get("dataset_id"):
            raise ValueError("schema.yaml must contain 'dataset_id'")

        known = set(cls.model_fields)
        unknown = [key for key in data if key not in known]
        if not unknown:
            return data

        for key in unknown:
            suggestion = get_close_matches(key, known - {"extra"}, n=1)
            message = f"Unknown schema key '{key}'"
            if suggestion:
                message += f" — did you mean '{suggestion[0]}'?"
            message += " The key is ignored by this SDK version (kept under 'extra')."
            warnings.warn(message, SchemaValidationWarning, stacklevel=2)

        cleaned = {key: value for key, value in data.items() if key in known}
        cleaned["extra"] = dict(data.get("extra") or {}) | {
            key: data[key] for key in unknown
        }
        return cleaned

    @field_validator("task", mode="before")
    @classmethod
    def _normalize_task(cls, value: Any) -> str | None:
        return str(value).upper() if value else None

    @field_validator("root_strategy", mode="before")
    @classmethod
    def _validate_root_strategy(cls, value: Any) -> Any:
        if value is None:
            return value
        try:
            Strategy(value)
        except ValueError:
            supported = ", ".join(member.value for member in Strategy)
            raise ValueError(
                f"Unknown root_strategy '{value}'. Supported strategies: {supported}"
            ) from None
        return value

    def to_yaml_dict(self) -> dict[str, Any]:
        """
        Serialise the schema to a plain dict suitable for YAML output.

        Excludes fields that are at their default values so that the
        generated ``schema.yaml`` stays compact and readable.  The
        ``extra`` dict is merged into the top level.
        """
        data = self.model_dump(exclude_defaults=True, exclude={"extra"})
        # Merge extra keys into the top level
        if self.extra:
            data.update(self.extra)
        return data

to_yaml_dict()

Serialise the schema to a plain dict suitable for YAML output.

Excludes fields that are at their default values so that the generated schema.yaml stays compact and readable. The extra dict is merged into the top level.

Source code in src/datacollective/schema.py

def to_yaml_dict(self) -> dict[str, Any]:
    """
    Serialise the schema to a plain dict suitable for YAML output.

    Excludes fields that are at their default values so that the
    generated ``schema.yaml`` stays compact and readable.  The
    ``extra`` dict is merged into the top level.
    """
    data = self.model_dump(exclude_defaults=True, exclude={"extra"})
    # Merge extra keys into the top level
    if self.extra:
        data.update(self.extra)
    return data

Strategy

Bases: StrEnum

Loading strategies recognised by schema loaders.

The values are the valid root_strategy schema field entries; the registry maps each member to its loader class.

Source code in src/datacollective/schema.py

class Strategy(StrEnum):
    """Loading strategies recognised by schema loaders.

    The values are the valid ``root_strategy`` schema field entries; the
    registry maps each member to its loader class.
    """

    INDEX = "index"
    MULTI_SPLIT = "multi_split"
    MULTI_SECTIONS = "multi_sections"
    PAIRED_GLOB = "paired_glob"
    GLOB = "glob"

datacollective.schema_loaders.base

BaseSchemaLoader

Bases: ABC

Interface that every strategy loader must implement.

Parameters:

Name	Type	Description	Default
schema	DatasetSchema	The parsed schema for the dataset.	required
extract_dir	Path	The directory where the dataset files have been extracted.	required

Source code in src/datacollective/schema_loaders/base.py

class BaseSchemaLoader(abc.ABC):
    """
    Interface that every strategy loader must implement.

    Args:
        schema (DatasetSchema): The parsed schema for the dataset.
        extract_dir (Path): The directory where the dataset files have been extracted.
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        self.schema = schema
        self.extract_dir = extract_dir.expanduser().resolve()
        self._resolved_index_file: Path | None = None
        self._dataset_root: Path | None = None
        self._audio_file_cache: dict[
            tuple[tuple[str, ...], str | None], list[Path]
        ] = {}

    @abc.abstractmethod
    def load(self) -> pd.DataFrame:
        """Load the dataset into a pandas DataFrame according to ``self.schema``."""
        ...

    def _load_index_file(self) -> pd.DataFrame:
        """Locate the index file and read it into a raw `~pandas.DataFrame`.

        Resolves the separator from ``schema.separator`` (explicit override) or
        ``schema.format`` via `FORMAT_SEP`, then delegates the file
        lookup to `_resolve_index_file`.

        Used by index-based strategies so that each loader only needs to call
        `_apply_column_mappings` on the result.

        Returns:
            A raw (unmapped) DataFrame exactly as read from the index file.
        """
        index_path = self._resolve_index_file()
        return self._read_delimited_file(index_path)

    def _resolve_index_file(self) -> Path:
        """Find the index file inside the extracted directory.

        Resolution is deterministic: the literal path relative to the dataset
        root wins when it exists. Otherwise (non-strict schemas only) the tree
        is searched recursively and the shallowest match is used; multiple
        matches at the same depth are an error rather than an arbitrary pick.

        Used by index-based loaders.

        Raises:
            FileNotFoundError: If no matching file is found (in strict mode,
                if the literal relative path does not exist).
            ValueError: If the recursive search is ambiguous.
        """
        if self._resolved_index_file is not None:
            return self._resolved_index_file

        assert self.schema.index_file is not None
        literal = self.extract_dir / self.schema.index_file
        if literal.is_file():
            resolved = literal
        elif self.schema.strict:
            raise FileNotFoundError(
                f"Index file '{self.schema.index_file}' not found at "
                f"'{literal}' (strict schema: no recursive search)"
            )
        else:
            candidates = list(self.extract_dir.rglob(self.schema.index_file))
            if not candidates:
                raise FileNotFoundError(
                    f"Index file '{self.schema.index_file}' not found "
                    f"under '{self.extract_dir}'"
                )
            # Prefer the shallowest match; equal-depth ties are ambiguous
            candidates.sort(key=lambda p: (len(p.parts), str(p)))
            min_depth = len(candidates[0].parts)
            ties = [c for c in candidates if len(c.parts) == min_depth]
            if len(ties) > 1:
                raise ValueError(
                    f"Ambiguous index_file '{self.schema.index_file}': "
                    f"{len(ties)} matches at the same depth under "
                    f"'{self.extract_dir}': {[str(t) for t in ties[:5]]}. "
                    "Set 'index_file' to an explicit path relative to the "
                    "dataset root."
                )
            resolved = candidates[0]

        self._resolved_index_file = resolved
        self._dataset_root = self._derive_dataset_root(resolved, self.schema.index_file)
        return self._resolved_index_file

    def _apply_column_mappings(self, raw_df: pd.DataFrame) -> pd.DataFrame:
        """Select and rename columns according to the schema, applying dtype conversions.

        Used by index-based loaders.

        Raises:
            KeyError: If a required column is not found in *raw_df*.
        """
        result_cols: dict[str, pd.Series] = {}

        for logical_name, col_map in self.schema.columns.items():
            source = col_map.source_column
            resolved_source = self._resolve_column(raw_df.columns, source)

            if resolved_source is None:
                if col_map.optional:
                    logger.debug(f"Optional column '{source}' not found — skipping.")
                    continue
                raise KeyError(
                    f"Required column '{source}' not found in index file. "
                    f"Available columns: {list(raw_df.columns)}"
                )

            series = raw_df[resolved_source]

            if col_map.dtype == "file_path":
                misses: list[str] = []
                series = raw_df.apply(
                    lambda row, _col_map=col_map, _source=resolved_source: (
                        self._resolve_file_path(row[_source], _col_map, row, misses)
                    ),
                    axis=1,
                )
                self._warn_unresolved_files(
                    logical_name,
                    misses,
                    len(raw_df),
                    "the constructed paths are kept as-is",
                )
            elif col_map.dtype == "file_content":
                misses = []
                series = raw_df.apply(
                    lambda row, _col_map=col_map, _source=resolved_source: (
                        self._load_file_content(row[_source], _col_map, row, misses)
                    ),
                    axis=1,
                )
                self._warn_unresolved_files(
                    logical_name, misses, len(raw_df), "their values are set to missing"
                )
            elif col_map.dtype == "category":
                series = series.astype("category")
            elif col_map.dtype in ("int", "float"):
                numeric = pd.to_numeric(series, errors="coerce")
                coerced = series.notna() & numeric.isna()
                if coerced.any():
                    examples = ", ".join(
                        repr(value) for value in series[coerced].unique()[:3]
                    )
                    warnings.warn(
                        f"Column '{logical_name}': {int(coerced.sum())} of "
                        f"{len(series)} values could not be parsed as "
                        f"{col_map.dtype} and were set to missing "
                        f"(e.g. {examples}).",
                        DataLoadWarning,
                        stacklevel=2,
                    )
                series = numeric.astype("Int64") if col_map.dtype == "int" else numeric
            else:
                # default: treat as string, preserving missing values
                # (a plain astype(str) would turn NaN into the string "nan")
                series = series.where(series.isna(), series.astype(str))

            result_cols[logical_name] = series

        return pd.DataFrame(result_cols)

    def _warn_unresolved_files(
        self, logical_name: str, misses: list[str], total: int, consequence: str
    ) -> None:
        if not misses:
            return
        examples = ", ".join(repr(miss) for miss in misses[:3])
        warnings.warn(
            f"Column '{logical_name}': {len(misses)} of {total} values did not "
            f"resolve to an existing file ({consequence}). "
            f"Examples: {examples}.",
            DataLoadWarning,
            stacklevel=3,
        )

    def _read_delimited_file(self, file_path: Path) -> pd.DataFrame:
        sep = self._resolve_separator(file_path)
        # TSVs (e.g. Common Voice) don't quote fields; a field starting with
        # '"' would otherwise swallow every row up to the next '"'.
        # See https://github.com/Mozilla-Data-Collective/datacollective-python/issues/141 for more details
        quoting = self.schema.quoting or ("none" if sep == "\t" else "minimal")

        logger.debug(f"Reading delimited file: {file_path} (sep={sep!r})")
        df = pd.read_csv(
            file_path,
            sep=sep,
            # Separator sniffing (sep=None) requires the python engine
            engine="python" if sep is None else None,
            header="infer" if self.schema.has_header else None,
            encoding=self.schema.encoding,
            skipinitialspace=True,
            quoting=csv.QUOTE_NONE if quoting == "none" else csv.QUOTE_MINIMAL,
        )

        if quoting == "minimal":
            self._warn_if_rows_merged(file_path, df, sep)
        elif self.schema.quoting is None:
            self._warn_if_fields_look_quoted(file_path, df)
        return self._normalize_dataframe_columns(df)

    def _warn_if_rows_merged(
        self, file_path: Path, df: pd.DataFrame, sep: str | None
    ) -> None:
        """Warn when quoted fields absorbed lines that look like separate rows.

        Legitimate multi-line quoted values trigger this too, which is why it
        is a warning and not an error.
        """
        # pandas skips lines of only spaces/tabs, unless a tab is the separator
        blank = " \r\n" if sep == "\t" else " \t\r\n"
        with open(file_path, encoding=self.schema.encoding, errors="replace") as f:
            data_lines = sum(1 for line in f if line.strip(blank))
        data_lines -= self.schema.has_header
        merged = data_lines - len(df)
        if merged <= 0:
            return
        warnings.warn(
            f"'{file_path.name}': {data_lines} data lines were parsed into "
            f"{len(df)} rows, so {merged} lines were read as part of quoted "
            "fields. This is expected if the file has quoted values spanning "
            "several lines. If the file does not quote its fields (a value "
            "merely starts with '\"'), rows were lost: set `quoting: none` in "
            "the schema.",
            DataLoadWarning,
            stacklevel=3,
        )

    def _warn_if_fields_look_quoted(self, file_path: Path, df: pd.DataFrame) -> None:
        """Warn when a TSV read without quoting has CSV-style quoted values.

        Writers such as ``pandas.DataFrame.to_csv(sep="\\t")`` wrap values that
        contain '"' in quotes and double the inner ones: ``say "hi" now``
        becomes ``"say ""hi"" now"``.
        Read with quoting disabled, those values keep the quote characters.
        """
        for column in df.columns:
            values = df[column]
            # pandas < 3 reads text as object, pandas >= 3 as StringDtype
            if values.dtype != object and not isinstance(values.dtype, pd.StringDtype):
                continue
            # A '""' pair inside the wrapping quotes is what a quoting writer
            # leaves behind; a sentence that merely starts and ends with '"'
            # has none. The substring test is a cheap pre-filter.
            candidates = values[values.str.contains('""', regex=False, na=False)]
            quoted = candidates[candidates.str.fullmatch(r'".*"".*"')]
            if quoted.empty:
                continue
            warnings.warn(
                f"'{file_path.name}': column '{column}' has {len(quoted)} values "
                f"that look CSV-quoted (e.g. {quoted.iloc[0]!r}). Tab-separated "
                "files are read without quoting, so the quote characters were "
                "kept. If the file was written with quoting (e.g. by pandas), "
                "set `quoting: minimal` in the schema; set `quoting: none` to "
                "silence this warning.",
                DataLoadWarning,
                stacklevel=3,
            )
            return

    def _resolve_separator(self, file_path: Path | None = None) -> str | None:
        if self.schema.separator:
            return self.schema.separator
        if self.schema.format:
            return FORMAT_SEP.get(self.schema.format.casefold())
        index_file_path = (
            Path(self.schema.index_file) if self.schema.index_file else None
        )
        for candidate in (file_path, index_file_path):
            if not candidate:
                continue
            suffix = candidate.suffix.casefold()
            if suffix in SUFFIX_SEP:
                return SUFFIX_SEP[suffix]
        return None

    def _normalize_dataframe_columns(self, raw_df: pd.DataFrame) -> pd.DataFrame:
        if raw_df.empty and not len(raw_df.columns):
            return raw_df

        normalized_columns: list[str | int] = []
        for column in raw_df.columns:
            if isinstance(column, str):
                normalized_columns.append(column.replace("\ufeff", "").strip())
            else:
                normalized_columns.append(column)

        result = raw_df.copy()
        result.columns = normalized_columns
        return result

    def _resolve_column(
        self, columns: pd.Index, source: str | int, kind: str = "index"
    ) -> str | int | None:
        """Find *source* among *columns*, tolerating whitespace/case/BOM
        differences unless the schema is strict."""
        if source in columns:
            return source
        if isinstance(source, int) or self.schema.strict:
            # Strict schemas require exact column names — no fuzzy matching
            return None

        stripped_source = source.strip()
        if stripped_source in columns:
            return stripped_source

        normalized_source = self._normalize_column_key(stripped_source)
        matches = [
            column
            for column in columns
            if isinstance(column, str)
            and self._normalize_column_key(column) == normalized_source
        ]
        if len(matches) > 1:
            raise KeyError(
                f"Column '{source}' matched multiple {kind} columns after normalization: {matches}"
            )
        return matches[0] if matches else None

    def _normalize_column_key(self, column: str) -> str:
        cleaned = column.replace("\ufeff", "").strip()
        return " ".join(cleaned.split()).casefold()

    def _resolve_file_path(
        self,
        value: Any,
        col_map: ColumnMapping,
        row: pd.Series | None = None,
        misses: list[str] | None = None,
    ) -> Any:
        """Resolve *value* to an existing file path.

        With the default ``direct`` strategy, a value that resolves to no
        existing file is returned as the first constructed candidate path
        and recorded in *misses* (when given) so the caller can warn.

        Raises:
            FileNotFoundError: If ``path_match_strategy`` is ``exact`` or
                ``contains`` and the search finds no file.
            ValueError: If the ``exact`` or ``contains`` search matches more
                than one file.
        """
        if pd.isna(value):
            return value

        source_value = str(value).strip()
        raw_value = source_value
        if row is not None and col_map.path_template:
            raw_value = self._render_path_template(
                source_value, row, col_map.path_template
            )
        if not raw_value:
            return raw_value

        direct_candidates = self._build_direct_file_candidates(
            raw_value,
            col_map.file_extension,
            row=row,
            template_value=source_value,
        )
        for candidate in direct_candidates:
            if candidate.exists():
                return str(candidate)

        if col_map.path_match_strategy != "direct":
            matched_path = self._search_audio_file(
                raw_value,
                col_map,
                row=row,
                template_value=source_value,
            )
            if matched_path is not None:
                return str(matched_path)
            raise FileNotFoundError(
                f"Could not resolve file_path value '{raw_value}' using "
                f"path_match_strategy='{col_map.path_match_strategy}' "
                f"under base_audio_path={self.schema.base_audio_path!r}"
            )

        if misses is not None:
            misses.append(raw_value)
        if direct_candidates:
            return str(direct_candidates[0])
        return raw_value

    def _load_file_content(
        self,
        value: Any,
        col_map: ColumnMapping,
        row: pd.Series | None = None,
        misses: list[str] | None = None,
    ) -> Any:
        """Resolve a file path (like ``file_path`` dtype) and return its text content.

        With the default ``direct`` strategy, a value that does not resolve to
        an existing file makes the cell missing (``None``) and is recorded in
        *misses* (when given) so the caller can warn — a content column must
        never silently contain a path instead of the file's text.

        Raises:
            FileNotFoundError: If ``path_match_strategy`` is ``exact`` or
                ``contains`` and the search finds no file.
            ValueError: If the ``exact`` or ``contains`` search matches more
                than one file.
        """
        if pd.isna(value):  # if missing value, skip loading
            return value

        # Remove whitespaces in the path
        raw = str(value).strip()
        parts = Path(raw).parts
        if parts:
            raw = str(Path(*[p.strip() for p in parts]))

        resolved = self._resolve_file_path(raw, col_map, row)
        path = Path(resolved)
        if path.is_file():
            return path.read_text(encoding=self.schema.encoding).strip()
        if misses is not None:
            misses.append(raw)
        return None

    def _build_direct_file_candidates(
        self,
        raw_value: str,
        file_extension: str | None,
        row: pd.Series | None = None,
        template_value: str | None = None,
    ) -> list[Path]:
        relative_candidates = [Path(raw_value)]
        normalized_extension = self._normalize_extension(file_extension)
        if normalized_extension is not None and not Path(raw_value).suffix:
            relative_candidates.append(
                Path(raw_value).with_suffix(normalized_extension)
            )

        candidates: list[Path] = []
        seen: set[str] = set()
        for relative_candidate in relative_candidates:
            if relative_candidate.is_absolute():
                path_candidates = [relative_candidate]
            else:
                path_candidates = [
                    root / relative_candidate
                    for root in self._get_audio_search_roots(
                        row=row, template_value=template_value or raw_value
                    )
                ]
                dataset_root = self._get_dataset_root()
                path_candidates.append(dataset_root / relative_candidate)
                if dataset_root != self.extract_dir:
                    path_candidates.append(self.extract_dir / relative_candidate)

            for candidate in path_candidates:
                key = str(candidate)
                if key in seen:
                    continue
                seen.add(key)
                candidates.append(candidate)

        return candidates

    def _get_audio_search_roots(
        self,
        row: pd.Series | None = None,
        template_value: str | None = None,
    ) -> list[Path]:
        """Resolve ``base_audio_path`` into deduplicated search roots.

        Empty entries (including template renders that come out empty) fall
        back to the dataset root; relative entries are anchored at it.
        """
        raw_paths = self.schema.base_audio_path
        if not isinstance(raw_paths, list):
            raw_paths = [raw_paths] if raw_paths else []

        roots: list[Path] = []
        for raw_path in raw_paths:
            root = self._resolve_audio_root(raw_path, row, template_value)
            if root not in roots:
                roots.append(root)
        return roots or [self._get_dataset_root()]

    def _resolve_audio_root(
        self,
        raw_path: str,
        row: pd.Series | None,
        template_value: str | None,
    ) -> Path:
        dataset_root = self._get_dataset_root()
        if not raw_path:
            return dataset_root

        if row is not None and "${" in raw_path:
            raw_path = self._render_path_template(
                template_value or "",
                row,
                raw_path,
                template_name="base_audio_path",
            )
            if not raw_path:
                return dataset_root

        path = Path(raw_path)
        return path if path.is_absolute() else dataset_root / path

    def _search_audio_file(
        self,
        raw_value: str,
        col_map: ColumnMapping,
        row: pd.Series | None = None,
        template_value: str | None = None,
    ) -> Path | None:
        search_roots = self._get_audio_search_roots(
            row=row, template_value=template_value or raw_value
        )
        search_files = self._get_searchable_audio_files(
            search_roots, col_map.file_extension
        )

        if col_map.path_match_strategy == "exact":
            matches = self._find_exact_matches(
                raw_value, col_map.file_extension, search_files, search_roots
            )
        else:  # "contains"
            matches = self._find_contains_matches(raw_value, search_files, search_roots)

        if len(matches) > 1:
            raise ValueError(
                f"Ambiguous file_path value '{raw_value}' using "
                f"path_match_strategy='{col_map.path_match_strategy}'. "
                f"Matches: {[str(match) for match in matches[:5]]}"
            )
        return matches[0] if matches else None

    def _find_exact_matches(
        self,
        raw_value: str,
        file_extension: str | None,
        search_files: list[Path],
        search_roots: list[Path],
    ) -> list[Path]:
        """Candidates whose name — or, for extension-less values, stem or
        extension-completed name — equals the value, or whose path relative to
        a search root equals it (case-insensitive)."""
        raw_path = Path(raw_value)
        extension = self._normalize_extension(file_extension)

        expected_names = {raw_path.name}
        expected_relatives = {raw_path.as_posix().casefold()}
        match_stem = not raw_path.suffix
        if match_stem and extension is not None:
            expected_names.add(f"{raw_path.name}{extension}")
            expected_relatives.add(
                f"{raw_path.as_posix().casefold()}{extension.casefold()}"
            )

        matches: list[Path] = []
        for candidate in search_files:
            is_match = (
                candidate.name in expected_names
                or (match_stem and candidate.stem == raw_path.name)
                or not expected_relatives.isdisjoint(
                    self._candidate_relative_paths(candidate, search_roots)
                )
            )
            if is_match and candidate not in matches:
                matches.append(candidate)
        return matches

    def _find_contains_matches(
        self,
        raw_value: str,
        search_files: list[Path],
        search_roots: list[Path],
    ) -> list[Path]:
        """Candidates whose name, stem, or path relative to a search root
        contains the value as a substring (case-insensitive)."""
        needle = raw_value.casefold()

        matches: list[Path] = []
        for candidate in search_files:
            haystacks = [candidate.name.casefold(), candidate.stem.casefold()]
            haystacks.extend(self._candidate_relative_paths(candidate, search_roots))
            if (
                any(needle in haystack for haystack in haystacks)
                and candidate not in matches
            ):
                matches.append(candidate)
        return matches

    def _candidate_relative_paths(
        self, candidate: Path, search_roots: list[Path]
    ) -> list[str]:
        relative_paths: list[str] = []
        for root in search_roots:
            try:
                relative_paths.append(candidate.relative_to(root).as_posix().casefold())
            except ValueError:
                continue
        return relative_paths

    def _get_searchable_audio_files(
        self, search_roots: list[Path], file_extension: str | None
    ) -> list[Path]:
        """List candidate files under *search_roots* (shallowest first per
        root), cached per (roots, extension) pair."""
        extension = self._normalize_extension(file_extension)
        cache_key = (tuple(str(root) for root in search_roots), extension)
        if cache_key not in self._audio_file_cache:
            self._audio_file_cache[cache_key] = [
                path
                for root in search_roots
                for path in self._list_searchable_files(root, extension)
            ]
        return self._audio_file_cache[cache_key]

    def _list_searchable_files(self, root: Path, extension: str | None) -> list[Path]:
        if root.is_file():
            return [root] if self._is_searchable_audio_file(root, extension) else []
        if not root.exists():
            return []

        files = [
            path
            for path in root.rglob("*")
            if self._is_searchable_audio_file(path, extension)
        ]
        files.sort(key=lambda path: (len(path.relative_to(root).parts), str(path)))
        return files

    def _matches_extension(self, path: Path, extension: str | None) -> bool:
        if extension is None:
            return True
        return path.suffix.casefold() == extension.casefold()

    def _is_searchable_audio_file(self, path: Path, extension: str | None) -> bool:
        return (
            path.is_file()
            and not path.name.startswith("._")
            and self._matches_extension(path, extension)
        )

    def _normalize_extension(self, extension: str | None) -> str | None:
        if extension is None or extension == "":
            return None
        return extension if extension.startswith(".") else f".{extension}"

    def _get_dataset_root(self) -> Path:
        return self._dataset_root or self.extract_dir

    def _derive_dataset_root(
        self, resolved_path: Path, relative_path: str | None
    ) -> Path:
        if not relative_path:
            return resolved_path.parent

        relative = Path(relative_path)
        if relative.is_absolute():
            return relative.parent

        num_parts = len(relative.parts)
        if num_parts <= 1:
            return resolved_path.parent

        return resolved_path.parents[num_parts - 1]

    def _render_path_template(
        self,
        raw_value: str,
        row: pd.Series,
        template: str,
        template_name: str = "path_template",
    ) -> str:
        def replace(match: re.Match[str]) -> str:
            placeholder = match.group(1).strip()
            if placeholder == "value":
                return raw_value

            row_key = self._resolve_column(row.index, placeholder, kind="row")
            if row_key is None:
                raise KeyError(
                    f"Could not render {template_name} placeholder '{placeholder}'. "
                    f"Available columns: {list(row.index)}"
                )

            cell_value = row[row_key]
            if pd.isna(cell_value):
                return ""
            return str(cell_value).strip()

        return re.sub(r"\$\{([^}]+)\}", replace, template)

load() abstractmethod

Load the dataset into a pandas DataFrame according to self.schema.

Source code in src/datacollective/schema_loaders/base.py

@abc.abstractmethod
def load(self) -> pd.DataFrame:
    """Load the dataset into a pandas DataFrame according to ``self.schema``."""
    ...

Strategy

Bases: StrEnum

Loading strategies recognised by schema loaders.

The values are the valid root_strategy schema field entries; the registry maps each member to its loader class.

Source code in src/datacollective/schema.py

class Strategy(StrEnum):
    """Loading strategies recognised by schema loaders.

    The values are the valid ``root_strategy`` schema field entries; the
    registry maps each member to its loader class.
    """

    INDEX = "index"
    MULTI_SPLIT = "multi_split"
    MULTI_SECTIONS = "multi_sections"
    PAIRED_GLOB = "paired_glob"
    GLOB = "glob"

datacollective.schema_loaders.registry

datacollective.schema_loaders.cache_schema

datacollective.schema_loaders.contracts

datacollective.schema_loaders.strategies.index

IndexLoader

Bases: BaseSchemaLoader

Load a dataset from a single delimited index file (the default strategy).

An index file (e.g. CSV/TSV) holds one row per sample. When the schema declares column mappings they are applied (renaming, dtype conversion, file-path resolution); otherwise the raw DataFrame is returned as-is.

Source code in src/datacollective/schema_loaders/strategies/index.py

class IndexLoader(BaseSchemaLoader):
    """Load a dataset from a single delimited index file (the default strategy).

    An index file (e.g. CSV/TSV) holds one row per sample.  When the schema
    declares column mappings they are applied (renaming, dtype conversion,
    file-path resolution); otherwise the raw DataFrame is returned as-is.
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        super().__init__(schema, extract_dir)
        if not schema.index_file:
            raise ValueError("index strategy schema must specify 'index_file'")

    def load(self) -> pd.DataFrame:
        raw_df = self._load_index_file()
        if not self.schema.columns:
            # No column mapping -> return the raw dataframe as-is
            return raw_df
        return self._apply_column_mappings(raw_df)

datacollective.schema_loaders.strategies.multi_split

MultiSplitLoader

Bases: BaseSchemaLoader

Load a dataset spread across one delimited file per split.

All split files whose stems match the splits list are read, a split column is added to each, column mappings are applied when declared, and the parts are concatenated.

Source code in src/datacollective/schema_loaders/strategies/multi_split.py

class MultiSplitLoader(BaseSchemaLoader):
    """Load a dataset spread across one delimited file per split.

    All split files whose stems match the ``splits`` list are read, a
    ``split`` column is added to each, column mappings are applied when
    declared, and the parts are concatenated.
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        super().__init__(schema, extract_dir)
        if not schema.splits:
            raise ValueError(
                "multi_split schema must specify 'splits' (list of split names)"
            )

    def load(self) -> pd.DataFrame:
        assert self.schema.splits is not None

        pattern = self.schema.splits_file_pattern or "**/*.tsv"
        allowed_splits = set(self.schema.splits)

        split_files: dict[str, Path] = {}
        for path in self.extract_dir.rglob(pattern):
            if path.stem in allowed_splits:
                # Prefer the shallowest match per split name
                if path.stem not in split_files or len(path.parts) < len(
                    split_files[path.stem].parts
                ):
                    split_files[path.stem] = path

        if not split_files:
            raise RuntimeError(
                f"No split files matching pattern '{pattern}' with stems in "
                f"{sorted(allowed_splits)} found under '{self.extract_dir}'"
            )

        frames: list[pd.DataFrame] = []

        for split_name, file_path in sorted(split_files.items()):
            logger.debug(f"Reading split '{split_name}' from {file_path}")
            df = self._read_delimited_file(file_path)
            # Set before mapping too, so column mappings can source from `split`
            df["split"] = split_name
            if self.schema.columns:
                df = self._apply_column_mappings(df)
            df["split"] = split_name
            frames.append(df)

        return pd.concat(frames, ignore_index=True)

datacollective.schema_loaders.strategies.multi_sections

MultiSectionsLoader

Bases: BaseSchemaLoader

Load a dataset organised as one index file per section directory.

Each section directory under section_root holds its own index file. A section column (the directory name) is added to each part, column mappings are applied when declared, and the parts are concatenated.

Source code in src/datacollective/schema_loaders/strategies/multi_sections.py

class MultiSectionsLoader(BaseSchemaLoader):
    """Load a dataset organised as one index file per section directory.

    Each section directory under ``section_root`` holds its own index file.
    A ``section`` column (the directory name) is added to each part, column
    mappings are applied when declared, and the parts are concatenated.
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        super().__init__(schema, extract_dir)
        if not schema.sections:
            raise ValueError(
                "multi_sections schema must specify 'sections' (list of section names)"
            )
        if not schema.section_root:
            raise ValueError("multi_sections schema must specify 'section_root'")
        if not schema.index_file:
            raise ValueError("multi_sections schema must specify 'index_file'")

    def load(self) -> pd.DataFrame:
        parts: list[pd.DataFrame] = []
        for section_name, section_path in self._resolve_sections():
            section_df = self._read_delimited_file(section_path)

            # Anchor relative paths (``base_audio_path``, ``file_path`` values)
            # at the section directory, mirroring how the index strategy
            # anchors them at the directory of the resolved index file.
            self._dataset_root = self._derive_dataset_root(
                section_path, self.schema.index_file
            )
            if self.schema.columns:
                section_df = self._apply_column_mappings(section_df)
            section_df["section"] = section_name
            parts.append(section_df)

        self._dataset_root = None
        return pd.concat(parts, ignore_index=True)

    def _resolve_sections(self) -> list[tuple[str, Path]]:
        """
        Get the ``(section name, index file path)`` pair for each declared
        section, i.e. each subdirectory that includes an index file.
        """
        assert self.schema.sections is not None
        assert self.schema.index_file is not None
        assert self.schema.section_root is not None

        section_paths = []
        for section in self.schema.sections:
            section_path = (
                self.extract_dir
                / Path(self.schema.section_root)
                / Path(section)
                / self.schema.index_file
            )
            if not section_path.exists():
                raise FileNotFoundError(f"Index file '{section_path}' not found ")
            section_paths.append((section, section_path))

        return section_paths

datacollective.schema_loaders.strategies.paired_glob

PairedGlobLoader

Bases: BaseSchemaLoader

Load a dataset where each audio file is paired with a sidecar file.

Two variants exist, selected by schema.format:

• format: "json": each audio file has a JSON sidecar (matched via file_pattern); column mappings are required and are applied to the normalised JSON records.
• otherwise: each audio file has a matching text sidecar containing the transcription; requires file_pattern and audio_extension. Column mappings, when declared, are applied over the derived audio_path / transcription / split sources.

Source code in src/datacollective/schema_loaders/strategies/paired_glob.py

class PairedGlobLoader(BaseSchemaLoader):
    """Load a dataset where each audio file is paired with a sidecar file.

    Two variants exist, selected by ``schema.format``:

    - ``format: "json"``: each audio file has a JSON sidecar (matched via
      ``file_pattern``); column mappings are required and are applied to the
      normalised JSON records.
    - otherwise: each audio file has a matching text sidecar containing the
      transcription; requires ``file_pattern`` and ``audio_extension``.
      Column mappings, when declared, are applied over the derived
      ``audio_path`` / ``transcription`` / ``split`` sources.
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        super().__init__(schema, extract_dir)
        if not schema.file_pattern:
            raise ValueError("paired_glob schema must specify 'file_pattern'")
        if self._is_json_variant():
            if not schema.columns:
                raise ValueError(
                    "paired_glob schema with 'format: json' must specify column "
                    "mappings (e.g. for audio and transcription)"
                )
        elif not schema.audio_extension:
            raise ValueError(
                "paired_glob schema must specify 'audio_extension' "
                "(or 'format: json' for JSON sidecar files)"
            )

    def _is_json_variant(self) -> bool:
        return (self.schema.format or "").casefold() == "json"

    def load(self) -> pd.DataFrame:
        if self._is_json_variant():
            return self._load_json_sidecars()
        return self._load_text_sidecars()

    def _load_json_sidecars(self) -> pd.DataFrame:
        """
        Load a dataset where each audio file is paired with a JSON sidecar
        (matched via ``file_pattern``) instead of a central index file.

        When ``record_path`` is set, the JSON key it names must hold a list of
        records (e.g. time-aligned utterances) and each record becomes one row;
        the remaining top-level keys are flattened with dot notation
        (audio.filename, ...) and repeated on
        every row of that file.  Without ``record_path`` each JSON file yields
        a single row.  Column mappings are then applied as for index files, so
        ``file_path`` columns (typically sourced from a filename field inside
        the JSON) resolve through the usual audio-path machinery.
        """
        assert self.schema.file_pattern is not None

        json_files = sorted(self.extract_dir.rglob(self.schema.file_pattern))
        json_files = [p for p in json_files if not p.name.startswith("._")]
        if not json_files:
            raise FileNotFoundError(
                f"No files matching '{self.schema.file_pattern}' "
                f"found under '{self.extract_dir}'"
            )

        logger.debug(
            f"Found {len(json_files)} JSON files matching '{self.schema.file_pattern}'"
        )

        record_path = self.schema.record_path
        frames: list[pd.DataFrame] = []
        for path in json_files:
            data = json.loads(path.read_text(encoding=self.schema.encoding))
            if record_path:
                if record_path not in data:
                    raise KeyError(
                        f"record_path '{record_path}' not found in '{path}'. "
                        f"Available keys: {list(data)}"
                    )
                frame = pd.json_normalize(data, record_path=record_path)
                meta = pd.json_normalize(
                    {key: value for key, value in data.items() if key != record_path}
                )
                for column in meta.columns:
                    frame[column] = meta[column].iloc[0]
            else:
                frame = pd.json_normalize(data)
            frames.append(frame)

        raw_df = pd.concat(frames, ignore_index=True)
        return self._apply_column_mappings(raw_df)

    def _load_text_sidecars(self) -> pd.DataFrame:
        """
        Load a dataset where each audio file has a matching text file (e.g.
        ``.txt``) containing the transcription. The loader searches recursively
        for all text files matching the specified `file_pattern`, reads their
        contents, and pairs them with the corresponding audio files based on
        the same filename stem. The parent directory name of each text/audio
        pair is captured as a `split` column in the resulting DataFrame.

        When the schema declares ``columns``, the mappings are applied over
        the derived ``audio_path`` / ``transcription`` / ``split`` sources
        (renaming, dtype conversion, dropping); the ``split`` column is kept,
        mirroring the multi_split strategy.
        """
        assert self.schema.file_pattern is not None
        assert self.schema.audio_extension is not None

        text_files = sorted(self.extract_dir.rglob(self.schema.file_pattern))
        if not text_files:
            raise FileNotFoundError(
                f"No files matching '{self.schema.file_pattern}' "
                f"found under '{self.extract_dir}'"
            )

        logger.debug(
            f"Found {len(text_files)} text files matching '{self.schema.file_pattern}'"
        )

        audio_ext = self.schema.audio_extension
        rows: list[dict[str, str]] = []
        skipped: list[str] = []

        for txt_path in text_files:
            audio_path = txt_path.with_suffix(audio_ext)
            if not audio_path.exists():
                logger.debug(
                    f"No matching audio file for '{txt_path.name}' — skipping."
                )
                skipped.append(txt_path.name)
                continue

            transcription = txt_path.read_text(encoding=self.schema.encoding).strip()
            row: dict[str, str] = {
                "audio_path": str(audio_path),
                "transcription": transcription,
            }

            # Derive domain / split from parent directory name if present
            parent_name = txt_path.parent.name
            if parent_name:
                row["split"] = parent_name

            rows.append(row)

        if not rows:
            raise FileNotFoundError(
                f"No paired (text + {audio_ext}) files found under '{self.extract_dir}'"
            )

        if skipped:
            examples = ", ".join(repr(name) for name in skipped[:3])
            warnings.warn(
                f"{len(skipped)} of {len(text_files)} files matching "
                f"'{self.schema.file_pattern}' had no paired '{audio_ext}' "
                f"audio file and were skipped (e.g. {examples}). Check "
                "'audio_extension' if this is unexpected.",
                DataLoadWarning,
                stacklevel=2,
            )

        raw_df = pd.DataFrame(rows)
        if not self.schema.columns:
            return raw_df

        mapped = self._apply_column_mappings(raw_df)
        if "split" in raw_df.columns:
            mapped["split"] = raw_df["split"]
        return mapped

datacollective.schema_loaders.strategies.glob

GlobLoader

Bases: BaseSchemaLoader

Load a directory-structured dataset by globbing for files.

Metadata (e.g. speaker ID, language) is derived from each matched file's path rather than from an index file or sidecar pairing.

When the schema declares columns, each mapping's source_column names a path-derived source instead of a file column:

• path: absolute path to the matched file
• name: file name (with extension)
• stem: file name without extension
• parent: parent directory name (same as parents[0])
• parents[N]: name of the Nth ancestor directory (0 = parent); empty string when the path is not that deep
• content: text content of the matched file (read with schema.encoding, stripped)

Without columns the loader falls back to its default output: audio_path (absolute path), language (parent directory name) and speaker_id (grandparent directory name).

Source code in src/datacollective/schema_loaders/strategies/glob.py

class GlobLoader(BaseSchemaLoader):
    """Load a directory-structured dataset by globbing for files.

    Metadata (e.g. speaker ID, language) is derived from each matched file's
    path rather than from an index file or sidecar pairing.

    When the schema declares ``columns``, each mapping's ``source_column``
    names a path-derived source instead of a file column:

    - ``path``: absolute path to the matched file
    - ``name``: file name (with extension)
    - ``stem``: file name without extension
    - ``parent``: parent directory name (same as ``parents[0]``)
    - ``parents[N]``: name of the Nth ancestor directory (0 = parent);
      empty string when the path is not that deep
    - ``content``: text content of the matched file (read with
      ``schema.encoding``, stripped)

    Without ``columns`` the loader falls back to its default output:
    ``audio_path`` (absolute path), ``language`` (parent directory name) and
    ``speaker_id`` (grandparent directory name).
    """

    def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
        super().__init__(schema, extract_dir)
        if not schema.file_pattern:
            raise ValueError("glob schema must specify 'file_pattern'")
        for logical_name, col_map in schema.columns.items():
            if not self._is_valid_source(col_map.source_column):
                raise ValueError(
                    f"glob schema column '{logical_name}' has unsupported "
                    f"source_column {col_map.source_column!r}. Supported "
                    f"path-derived sources: {', '.join(GLOB_SOURCES)}"
                )

    def load(self) -> pd.DataFrame:
        """Glob for files and derive metadata from each matched path.

        When ``splits`` is set, each split name is treated as a subdirectory
        under ``extract_dir`` and a ``split`` column is added.  Otherwise
        the glob runs from ``extract_dir`` directly.
        """
        if self.schema.splits:
            return self._load_glob_splits()

        return self._glob_directory(self.extract_dir)

    def _load_glob_splits(self) -> pd.DataFrame:
        assert self.schema.splits is not None

        frames: list[pd.DataFrame] = []
        for split_name in self.schema.splits:
            split_dir = self.extract_dir / split_name
            if not split_dir.is_dir():
                raise FileNotFoundError(
                    f"Split directory '{split_name}' not found at '{split_dir}'"
                )
            df = self._glob_directory(split_dir)
            df["split"] = split_name
            frames.append(df)

        return pd.concat(frames, ignore_index=True)

    def _glob_directory(self, root: Path) -> pd.DataFrame:
        assert self.schema.file_pattern is not None

        matched = sorted(root.rglob(self.schema.file_pattern))
        matched = [p for p in matched if not p.name.startswith("._")]

        if not matched:
            raise FileNotFoundError(
                f"No files matching '{self.schema.file_pattern}' found under '{root}'"
            )

        logger.debug(f"Found {len(matched)} files under '{root.name}'")

        if not self.schema.columns:
            # Default output when no mapping is declared
            rows: list[dict[str, str]] = []
            for path in matched:
                rows.append(
                    {
                        "audio_path": str(path),
                        "language": path.parent.name,
                        "speaker_id": path.parent.parent.name,
                    }
                )
            return pd.DataFrame(rows)

        raw_df = self._build_path_metadata(matched)
        return self._apply_column_mappings(raw_df)

    def _build_path_metadata(self, files: list[Path]) -> pd.DataFrame:
        """Build a raw DataFrame with one column per referenced path source.

        Only the sources referenced by the schema's column mappings are
        materialised (so file contents are read only when requested).
        The result is fed through the regular `_apply_column_mappings`, which
        handles renaming, dtypes and optional columns.
        """
        sources = {
            str(col_map.source_column) for col_map in self.schema.columns.values()
        }
        return pd.DataFrame(
            {source: self._derive_source_values(files, source) for source in sources}
        )

    def _derive_source_values(self, files: list[Path], source: str) -> list[str]:
        if source == "path":
            return [str(path) for path in files]
        if source == "name":
            return [path.name for path in files]
        if source == "stem":
            return [path.stem for path in files]
        if source == "parent":
            return [path.parent.name for path in files]
        if source == "content":
            return [
                path.read_text(encoding=self.schema.encoding).strip() for path in files
            ]

        match = _PARENTS_RE.fullmatch(source)
        assert match is not None  # guaranteed by __init__ validation
        index = int(match.group(1))
        return [
            path.parents[index].name if index < len(path.parents) else ""
            for path in files
        ]

    def _is_valid_source(self, source: str | int) -> bool:
        if not isinstance(source, str):
            return False
        if source in ("path", "name", "stem", "parent", "content"):
            return True
        return _PARENTS_RE.fullmatch(source) is not None

load()

Glob for files and derive metadata from each matched path.

When splits is set, each split name is treated as a subdirectory under extract_dir and a split column is added. Otherwise the glob runs from extract_dir directly.

Source code in src/datacollective/schema_loaders/strategies/glob.py

def load(self) -> pd.DataFrame:
    """Glob for files and derive metadata from each matched path.

    When ``splits`` is set, each split name is treated as a subdirectory
    under ``extract_dir`` and a ``split`` column is added.  Otherwise
    the glob runs from ``extract_dir`` directly.
    """
    if self.schema.splits:
        return self._load_glob_splits()

    return self._glob_directory(self.extract_dir)