Skip to content

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)

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.

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.

 False
enable_logging bool

Whether to enable SDK logging to console and a local log file.

 False

Returns:

Type Description
Path

Path to the downloaded dataset archive.

Raises:

Type Description
ValueError

If dataset_id is empty.
FileNotFoundError

If the dataset does not exist (404).
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 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def download_dataset(
    dataset_id: str,
    download_directory: str | None = None,
    show_progress: bool = True,
    overwrite_existing: bool = False,
    enable_logging: 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.

    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.
        enable_logging: Whether to enable SDK logging to console and a local log file.

    Returns:
        Path to the downloaded dataset archive.

    Raises:
        ValueError: If dataset_id is empty.
        FileNotFoundError: If the dataset does not exist (404).
        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.
    """
    _enable_logging(enable_logging)
    logger.info(f"Downloading dataset {dataset_id}")

    dataset_details = get_dataset_details(dataset_id)

    archive_path = _download_dataset(
        dataset_id=dataset_details.id,
        archive_filename=_require_archive_filename(dataset_details),
        download_directory=download_directory,
        show_progress=show_progress,
        overwrite_existing=overwrite_existing,
        download_source=DOWNLOAD_SOURCE_SAVE,
    )
    return archive_path

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).
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 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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).
        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())

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.

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.
FileNotFoundError

If the dataset does not exist (404).
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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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`.

    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.
        FileNotFoundError: If the dataset does not exist (404).
        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,
    )

    base_dir = archive_path.parent
    extract_dir = _extract_archive(
        archive_path=archive_path,
        dest_dir=base_dir,
        overwrite_extracted=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 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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_dataset(
        dataset_id=dataset_id,
        download_directory=download_directory,
        show_progress=show_progress,
        overwrite_existing=overwrite_existing,
        enable_logging=enable_logging,
    )

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 at least name and longDescription.

 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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def create_submission_draft(submission: DatasetSubmission) -> dict[str, Any]:
    """
    Create a draft dataset submission.

    Args:
        submission: Dataset submission model containing at least `name`
            and `longDescription`.

    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)

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
Source code in src/datacollective/submissions.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
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,
) -> 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.
    """
    _enable_logging(enable_logging)

    submission = _ensure_submission_model(submission)

    _validate_final_submission_fields(submission, require_file_upload_id=False)

    # 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", {})
        submission_id = (
            submission_payload.get("id")
            if isinstance(submission_payload, dict)
            else None
        )
        if not submission_id:
            raise RuntimeError("Draft creation did not return a submission id")

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

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

    update_submission(submission_id, submission)

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

    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,
        )

    # The uploaded file is linked to the submission automatically when the
    # multipart upload completes (the upload was started with `submissionId`),
    # so `fileUploadId` is not sent on the metadata PATCH. We still record it on
    # the model to satisfy the local completeness check below.
    submission.fileUploadId = upload_state.fileUploadId
    _validate_final_submission_fields(submission, require_file_upload_id=True)

    logger.info("Submitting dataset for review...")

    response = submit_submission(submission_id, submission)

    logger.info("Submission complete!")

    return response

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 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
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 _should_validate_local_final_submission(submission):
        _validate_final_submission_fields(submission, require_file_upload_id=True)
    elif 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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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)

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
Source code in src/datacollective/upload.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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,
) -> 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.
    """
    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,
        is_sample=False,
    )

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

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
Source code in src/datacollective/upload.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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,
) -> 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.
    """
    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,
        is_sample=True,
    )

datacollective.models

Dataset

Bases: BaseModel

Dataset fields shared by the platform's dataset and dataset-submission API payloads.

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 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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class Dataset(BaseModel):
    """
    Dataset fields shared by the platform's dataset and dataset-submission
    API payloads.

    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 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.",
    )
    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."
    )
    forbiddenUsage: str | None = Field(
        None, description="Explicitly forbidden use cases."
    )
    additionalConditions: str | None = Field(
        None, description="Additional conditions for use."
    )
    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.",
    )
    visibility: str | None = Field(
        None,
        description="Dataset visibility (e.g., `public`, `private`, `restricted`).",
    )
    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."
    )
    organizationId: str | None = Field(
        None,
        description="Identifier for the organization that owns the dataset.",
    )
    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.",
    )
    updatedAt: str | None = Field(
        None,
        description="Timestamp when the record was last updated. Updated by the API on changes.",
    )

DatasetDetails

Bases: Dataset

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

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 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
class DatasetDetails(Dataset):
    """
    Dataset details as returned by the MDC API (read model).

    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."
    )

    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

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 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 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
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 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 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
    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.",
    )

    @model_validator(mode="after")
    def _validate_license_details(self) -> DatasetSubmission:
        has_custom_license_abbreviation = (
            self.licenseAbbreviation is not None
            and not isinstance(self.licenseAbbreviation, License)
        )
        requires_license_name = (
            has_custom_license_abbreviation or self.licenseUrl is not None
        )
        if requires_license_name and self.license is None:
            raise ValueError(
                "`license` is required when providing a custom `licenseAbbreviation` or `licenseUrl`"
            )
        return self

    @model_validator(mode="after")
    def _validate_pricing(self) -> DatasetSubmission:
        if self.isPaid and self.basePriceCents is None:
            raise ValueError(
                "`basePriceCents` is required when `isPaid` is True. The platform only "
                "accepts prices within its allowed range, in USD cents"
            )
        if self.basePriceCents is not None and not self.isPaid:
            raise ValueError(
                "`isPaid` must be True when providing `basePriceCents`, "
                "otherwise the dataset stays uncompensated and the price is ignored"
            )
        return self

    @model_validator(mode="after")
    def _validate_auto_approve_access_requests(self) -> DatasetSubmission:
        if (
            self.autoApproveAccessRequests
            and self.visibility is not None
            and self.visibility != Visibility.RESTRICTED
        ):
            raise ValueError(
                "`autoApproveAccessRequests` only applies to `restricted` datasets, "
                f"the only visibility with access requests; got `{self.visibility.value}`"
            )
        return self

    @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 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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 
16
17
18
19
20
21
22
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 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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.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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
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)'
    )
    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 
239
240
241
242
243
244
245
246
247
248
249
250
251
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 
21
22
23
24
25
26
27
28
29
30
31
32
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 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
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_source_column(raw_df, 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)
        header = "infer" if self.schema.has_header else None

        logger.debug(f"Reading delimited file: {file_path} (sep={sep!r})")
        df = self._read_csv(file_path, sep=sep, header=header)
        return self._normalize_dataframe_columns(df)

    def _read_csv(
        self, file_path: Path, sep: str | None, header: str | None
    ) -> pd.DataFrame:
        kwargs: dict[str, object] = {
            "header": header,
            "encoding": self.schema.encoding,
            "skipinitialspace": True,
        }
        if sep is None:
            kwargs["sep"] = None
            kwargs["engine"] = "python"
        else:
            kwargs["sep"] = sep
        return pd.read_csv(file_path, **kwargs)

    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_source_column(
        self, raw_df: pd.DataFrame, source: str | int
    ) -> str | int | None:
        if source in raw_df.columns:
            return source
        if isinstance(source, int):
            return source if source in raw_df.columns else None
        if self.schema.strict:
            # Strict schemas require exact column names — no fuzzy matching
            return None

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

        normalized_source = self._normalize_column_key(stripped_source)
        matches = [
            column
            for column in raw_df.columns
            if isinstance(column, str)
            and self._normalize_column_key(column) == normalized_source
        ]
        if len(matches) == 1:
            return matches[0]
        if len(matches) > 1:
            raise KeyError(
                f"Column '{source}' matched multiple index columns after normalization: {matches}"
            )
        return 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: object,
        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.
        """
        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: object,
        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.

        When the value does not resolve to an existing file, the cell becomes
        missing (``None``) and the value 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.
        """
        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_row_column(row, placeholder)
            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)

    def _resolve_row_column(
        self, row: pd.Series, source: str | int
    ) -> str | int | None:
        if source in row.index:
            return source
        if isinstance(source, int):
            return source if source in row.index else None
        if self.schema.strict:
            # Strict schemas require exact column names — no fuzzy matching
            return None

        stripped_source = source.strip()
        if stripped_source in row.index:
            return stripped_source

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

load() abstractmethod

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

Source code in src/datacollective/schema_loaders/base.py 
53
54
55
56
@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 
21
22
23
24
25
26
27
28
29
30
31
32
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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}")
            raw_df = self._read_delimited_file(file_path)
            raw_df["split"] = split_name

            if self.schema.columns:
                mapped = self._apply_column_mappings(raw_df)
                mapped["split"] = split_name
                frames.append(mapped)
            else:
                frames.append(raw_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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
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 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
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 
55
56
57
58
59
60
61
62
63
64
65
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)