Data values (import + export)

DataValuesAccessor on Dhis2Client.data_values — both directions of /api/dataValueSets. stream uploads JSON / XML / CSV / ADX without buffering the whole payload in memory; export reads a form's values back as a typed DataValueSet; and for an export too large to hold, client.stream("GET", "/api/dataValueSets.json", sink, params=...) writes the body straight to storage. For the typed-list-of-DataValue case, see Aggregate data values; the streaming accessor is the large-payload path.

When to reach for it

• Importing a CSV / JSON file that's larger than the host's free RAM.
• Pipe-style imports where the source is an AsyncIterable[bytes] (e.g. a transform step that emits a row at a time).
• Mixed-DataSet writes on DHIS2 v43 — the grouped path is the workaround for BUGS #35.
• Reading a form's values back after a write, or pulling one organisation unit's year for a report.
• Exporting a national year to a file or an object store without holding it in memory.

Worked example — stream a CSV file to DHIS2

from pathlib import Path

from dhis2w_core.client_context import open_client
from dhis2w_core.profile import profile_from_env


async with open_client(profile_from_env()) as client:
    # `stream` takes a Path (or any AsyncIterable[bytes]) + a content type.
    # The body is sent chunked; httpx2 never materialises the whole file.
    envelope = await client.data_values.stream(
        Path("./monthly-coverage-2026.csv"),
        content_type="application/csv",
    )
    count = envelope.import_count()
    if envelope.status == "OK" and count:
        print(f"imported {count.imported}  updated {count.updated}  ignored {count.ignored}")
    else:
        print(f"status={envelope.status!r}  message={envelope.message!r}")

Partly invalid imports: atomic_mode

stream(..., atomic_mode="ALL" | "OBJECT") forwards DHIS2's atomicMode switch. DHIS2 documents ALL as rejecting the whole import when any row is rejected and OBJECT as committing the rows that pass. On DHIS2 2.42 and 2.43 the switch has no observable effect on /api/dataValueSets: a payload with one valid row and one row that fails value-type validation commits the valid row under both modes and reports the other as ignored (BUGS.md #112). The parameter is forwarded so a build that honours it gets the documented behaviour; do not rely on ALL to keep a partly invalid import out.

A value equal to what the instance already holds also counts as ignored, so an unchanged re-import reads as ignored=N under either mode.

DHIS2 v42 answers any import that carries a conflict with HTTP 409, even when the other rows were committed; v43 answers 200 with the same WARNING envelope (BUGS.md #6). stream raises Dhis2ApiError on the 409, and the import summary is the exception's body:

from dhis2w_client import WebMessageResponse
from dhis2w_client.errors import Dhis2ApiError

try:
    envelope = await client.data_values.stream(body, content_type="application/json", atomic_mode="OBJECT")
except Dhis2ApiError as exc:
    if exc.status_code != 409 or not isinstance(exc.body, dict):
        raise
    envelope = WebMessageResponse.model_validate(exc.body)
count = envelope.import_count()
for conflict in envelope.conflicts():
    print(f"{conflict.object}: {conflict.value}")

Worked example — export a form's values

async with open_client(profile_from_env()) as client:
    # `data_set` / `period` / `org_unit` take one id or a sequence and repeat
    # on the wire. A date range replaces an explicit period list;
    # `children=True` includes the organisation units below `org_unit`.
    data_value_set = await client.data_values.export(
        data_set="BfMAe6Itzgt",
        org_unit="DiszpKrYNg8",
        start_date="2026-01-01",
        end_date="2026-06-30",
    )
    for value in data_value_set.dataValues or []:
        print(f"{value.dataElement} {value.period} {value.categoryOptionCombo} = {value.value}")

last_updated / last_updated_duration (2h, 1d) narrow to recently touched values; extra_params carries the rest of the endpoint's surface (idScheme, includeDeleted, dataElementGroup, ...).

Worked example — stream an export to a sink

export buffers the whole response. client.stream does not: it writes the body chunk by chunk to a pathlib.Path, to any object with .write(bytes) (sync or async), or to a callable that receives each chunk. It works for any endpoint, and client.analytics.stream_to is the Path-only convenience over it.

import io
from pathlib import Path

params = {
    "dataSet": ["BfMAe6Itzgt"],
    "orgUnit": ["ImspTQPwCqd"],
    "children": "true",
    "startDate": "2025-01-01",
    "endDate": "2025-12-31",
}

async with open_client(profile_from_env()) as client:
    # To a file. Parent directories are created; the file is never assembled in memory.
    written = await client.stream("GET", "/api/dataValueSets.json", Path("./exports/2025.json"), params=params)

    # To anything with .write(bytes): an open file, an upload stream, a BytesIO.
    buffer = io.BytesIO()
    await client.stream("GET", "/api/dataValueSets.json", buffer, params=params)

    # To a callable: count, hash, or forward each chunk. Async callables are awaited.
    async def forward(chunk: bytes) -> None:
        await queue.put(chunk)

    await client.stream("GET", "/api/dataValueSets.json", forward, params=params, chunk_size=64 * 1024)

A 4xx / 5xx raises AuthenticationError or Dhis2ApiError before anything is written; a Path sink is not created when the request fails.

Worked example — typed DataValue write (small batch)

from dhis2w_client import DataValue


values = [
    DataValue(
        dataElement="fbfJHSPpUQD",
        period="202604",
        orgUnit="ImspTQPwCqd",
        categoryOptionCombo="HllvX50cXC0",
        attributeOptionCombo="HllvX50cXC0",
        value="42",
    ),
]

async with open_client(profile_from_env()) as client:
    # `import_grouped_by_dataset` is the cross-version write path
    # (required on v43 for DEs in multiple DataSets — BUGS #35).
    # Returns `list[WebMessageResponse]` — one envelope per DataSet group.
    envelopes = await client.data_values.import_grouped_by_dataset(values)
    for env in envelopes:
        count = env.import_count()
        print(f"  status={env.status}  imported={count.imported if count else '?'}")

Related examples

• examples/client/stream_data_values.py — four streaming shapes (bytes, sync generator, Path/CSV, 1000-row file with timing).
• examples/client/aggregate_bulk_grouped.py — the grouped path against a v43 stack.
• examples/client/data_values_import_atomic.py — the same partly invalid payload under ALL and OBJECT.
• examples/client/data_values_export.py — export as a typed envelope, then client.stream to a Path, a BytesIO, and a callable.

data_values

Streaming data-value-set import — client.data_values.stream.

DHIS2's POST /api/dataValueSets accepts JSON, XML, CSV, and ADX payloads. For a 100k-row push (a typical month-end aggregate upload), buffering the whole body in Python memory before the POST is the thing to avoid:

• A 100k-row JSON payload sits at ~30-60 MB on the wire, and the Python parsed shape is 3-5x that — so ~150 MB resident just to stage the request.
• The same payload on CSV is ~8 MB; XML is in between.

client.data_values.stream(source, content_type) feeds httpx2's chunked transfer encoding directly, so the payload never sits fully in memory on the client side. The server consumes it as it arrives.

source accepts any of:

• pathlib.Path — opens the file and chunks it through.
• bytes / bytearray — single-shot for callers who already have the body assembled but want the typed WebMessageResponse envelope.
• Iterable[bytes] / AsyncIterable[bytes] — pass-through for generators that build the body on the fly (e.g. DB-row → CSV line).
• File-like with .read(size) -> bytes (sync or async) — adapted to a chunk iterator.

Supported content_type values map to the DHIS2-accepted MIME types:

• application/json (default)
• application/xml
• application/csv (also accepted: text/csv)
• application/adx+xml

Classes

DataValuesAccessor

Dhis2Client.data_values — streaming uploads to /api/dataValueSets.

Stateless wrapper over the streaming POST path. Stay here for the large import cases; use dhis2w_core.plugins.aggregate.service.push_data_values when the payload is already a small in-memory list of typed data values.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/data_values.py

class DataValuesAccessor:
    """`Dhis2Client.data_values` — streaming uploads to `/api/dataValueSets`.

    Stateless wrapper over the streaming POST path. Stay here for the large
    import cases; use `dhis2w_core.plugins.aggregate.service.push_data_values`
    when the payload is already a small in-memory list of typed data values.
    """

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client."""
        self._client = client

    async def stream(
        self,
        source: StreamSource,
        *,
        content_type: str = "application/json",
        dry_run: bool = False,
        preheat_cache: bool = True,
        import_strategy: str | None = None,
        atomic_mode: str | None = None,
        id_scheme: str | None = None,
        data_element_id_scheme: str | None = None,
        org_unit_id_scheme: str | None = None,
        skip_audit: bool = False,
        async_job: bool = False,
        chunk_size: int = _DEFAULT_CHUNK_SIZE,
    ) -> WebMessageResponse:
        """Stream `source` to `POST /api/dataValueSets` and return the typed envelope.

        `content_type` picks which DHIS2 parser handles the body (JSON / XML /
        CSV / ADX). Every param from the standard `/api/dataValueSets` surface
        is forwarded via query string:

        - `dry_run` → `dryRun=true`: validate without committing.
        - `preheat_cache=False` → `preheatCache=false`.
        - `import_strategy`: `CREATE` / `UPDATE` / `CREATE_AND_UPDATE` / `DELETE`.
        - `atomic_mode`: `ALL` rejects the whole import if any row is
          rejected; `OBJECT` commits the rows that pass and skips the rest.
          DHIS2 defaults to `ALL` when unset.
        - `id_scheme` / `data_element_id_scheme` / `org_unit_id_scheme`: pick
          the identifier scheme for the payload (`UID` / `CODE` / `NAME` / ...).
        - `skip_audit=True` → `skipAudit=true`.
        - `async_job=True` → `async=true`: DHIS2 queues the import as a job
          and the returned envelope carries `response.jobType` / `response.id`.
          Poll with `client.tasks.await_completion(envelope.task_ref())`.

        Returns a `WebMessageResponse`. For synchronous imports,
        `envelope.import_count()` gives `ImportCount.imported / updated /
        ignored / deleted`; `envelope.conflicts()` lists per-row rejections.
        Async imports return the task-ref envelope — poll it to completion
        to get the final report from DHIS2.
        """
        params: dict[str, Any] = {}
        if dry_run:
            params["dryRun"] = "true"
        if not preheat_cache:
            params["preheatCache"] = "false"
        if import_strategy is not None:
            params["importStrategy"] = import_strategy
        if atomic_mode is not None:
            params["atomicMode"] = atomic_mode
        if id_scheme is not None:
            params["idScheme"] = id_scheme
        if data_element_id_scheme is not None:
            params["dataElementIdScheme"] = data_element_id_scheme
        if org_unit_id_scheme is not None:
            params["orgUnitIdScheme"] = org_unit_id_scheme
        if skip_audit:
            params["skipAudit"] = "true"
        if async_job:
            params["async"] = "true"

        content = _coerce_stream_source(source, chunk_size=chunk_size)
        response = await self._client._request(  # noqa: SLF001 — accessor is intentionally tight with the client
            "POST",
            "/api/dataValueSets",
            params=params,
            content=content,
            extra_headers={"Content-Type": content_type},
        )
        raw = response.json() if response.content else {}
        return WebMessageResponse.model_validate(raw)

    async def export(
        self,
        *,
        data_set: str | Sequence[str] | None = None,
        period: str | Sequence[str] | None = None,
        org_unit: str | Sequence[str] | None = None,
        children: bool = False,
        start_date: str | None = None,
        end_date: str | None = None,
        last_updated: str | None = None,
        last_updated_duration: str | None = None,
        extra_params: Mapping[str, Any] | None = None,
    ) -> DataValueSet:
        """GET `/api/dataValueSets` and return the parsed `DataValueSet` envelope.

        The read side of the import handled by `stream()`. `data_set`, `period`,
        and `org_unit` each accept a single id or a sequence of ids, and repeat
        as `dataSet=`/`period=`/`orgUnit=` query params the way DHIS2 expects.

        - `children=True` -> `children=true`: include the org units below each
          `org_unit` in the export.
        - `start_date` / `end_date` (`YYYY-MM-DD`): export by date range instead
          of by `period`.
        - `last_updated` (`YYYY-MM-DD`) / `last_updated_duration` (e.g. `2h`,
          `1d`): export only values touched since then.
        - `extra_params` covers the rest of the surface (`idScheme`,
          `dataElementIdScheme`, `includeDeleted`, `dataElementGroup`,
          `orgUnitGroup`, ...). Pass a flat mapping or a list of 2-tuples.

        Buffers the whole response into a typed `DataValueSet`. For a large
        export, stream it straight to storage with
        `client.stream("GET", "/api/dataValueSets.json", sink, params=...)`
        instead of materialising every row in memory.

        Raises `Dhis2ApiError` on 4xx / 5xx.
        """
        params: dict[str, Any] = {}
        for key, value in (("dataSet", data_set), ("period", period), ("orgUnit", org_unit)):
            if value is None:
                continue
            params[key] = [value] if isinstance(value, str) else list(value)
        if children:
            params["children"] = "true"
        if start_date is not None:
            params["startDate"] = start_date
        if end_date is not None:
            params["endDate"] = end_date
        if last_updated is not None:
            params["lastUpdated"] = last_updated
        if last_updated_duration is not None:
            params["lastUpdatedDuration"] = last_updated_duration
        if extra_params:
            params.update(extra_params)
        raw = await self._client.get_raw("/api/dataValueSets", params=params)
        return DataValueSet.model_validate(raw)

    async def import_grouped_by_dataset(
        self,
        values: Sequence[DataValue],
        *,
        chunk_size: int = 1000,
        force: bool = False,
        skip_audit: bool = False,
    ) -> list[WebMessageResponse]:
        """Import typed `DataValue`s grouped by dataset — BUGS.md #35 workaround for v43.

        v43 added auto-target dataset detection on `POST /api/dataValueSets`
        (`DefaultDataEntryService.autoTargetDataSet`). When a posted DataValue's
        DataElement is referenced by 2+ DataSets, v43 aborts the entire chunk
        with `409 E8002 Data set detection failed`. v41 + v42 silently picked
        one matching dataset and imported the row.

        Workaround: pre-fetch the DataElement → DataSet membership map, group
        the input values by their DataSet (lexicographically-first DataSet id
        when a DE belongs to multiple — deterministic across runs), and POST
        each group with an explicit envelope `{"dataSet": "<id>", "dataValues":
        [...]}`. This shape is accepted by every DHIS2 major.

        Splits each per-dataset group into `chunk_size` rows per POST so the
        body stays inside httpx2's 300 s read timeout. Returns one
        `WebMessageResponse` per chunk; callers that want aggregate counts
        should walk `import_count()` across the list and sum.

        Skips values whose DataElement isn't in any DataSet (counted in the
        first response's import-count under `ignored`).
        """
        dataelement_to_dataset = await self._build_dataelement_to_dataset()
        grouped: dict[str, list[dict[str, Any]]] = {}
        for value in values:
            if value.dataElement is None:
                continue
            dataset_id = dataelement_to_dataset.get(value.dataElement)
            if dataset_id is None:
                continue
            grouped.setdefault(dataset_id, []).append(value.model_dump(by_alias=True, exclude_none=True, mode="json"))
        params: dict[str, Any] = {}
        if force:
            params["force"] = "true"
        if skip_audit:
            params["skipAudit"] = "true"
        responses: list[WebMessageResponse] = []
        for dataset_id, dumped in grouped.items():
            for start in range(0, len(dumped), chunk_size):
                chunk = dumped[start : start + chunk_size]
                raw = await self._client._request(  # noqa: SLF001
                    "POST",
                    "/api/dataValueSets",
                    params=params,
                    json={"dataSet": dataset_id, "dataValues": chunk},
                )
                body = raw.json() if raw.content else {}
                responses.append(WebMessageResponse.model_validate(body))
        return responses

    async def _build_dataelement_to_dataset(self) -> dict[str, str]:
        """Map every DE id to one of its DataSets (lexicographically-first when multiple).

        Used by `import_grouped_by_dataset` to scope each POST chunk to a
        single DataSet, avoiding v43's auto-target rejection (BUGS.md #35).
        """
        raw = await self._client.get_raw(
            "/api/dataSets",
            params={"fields": "id,dataSetElements[dataElement[id]]", "paging": "false"},
        )
        members: dict[str, list[str]] = {}
        for dataset in raw.get("dataSets") or []:
            dataset_id = dataset.get("id")
            if not isinstance(dataset_id, str):
                continue
            for entry in dataset.get("dataSetElements") or []:
                element = (entry.get("dataElement") or {}).get("id")
                if isinstance(element, str):
                    members.setdefault(element, []).append(dataset_id)
        return {element_id: sorted(dataset_ids)[0] for element_id, dataset_ids in members.items()}

Methods:

__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/data_values.py

def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client

stream(source, *, content_type='application/json', dry_run=False, preheat_cache=True, import_strategy=None, atomic_mode=None, id_scheme=None, data_element_id_scheme=None, org_unit_id_scheme=None, skip_audit=False, async_job=False, chunk_size=_DEFAULT_CHUNK_SIZE) async

Stream source to POST /api/dataValueSets and return the typed envelope.

content_type picks which DHIS2 parser handles the body (JSON / XML / CSV / ADX). Every param from the standard /api/dataValueSets surface is forwarded via query string:

• dry_run → dryRun=true: validate without committing.
• preheat_cache=False → preheatCache=false.
• import_strategy: CREATE / UPDATE / CREATE_AND_UPDATE / DELETE.
• atomic_mode: ALL rejects the whole import if any row is rejected; OBJECT commits the rows that pass and skips the rest. DHIS2 defaults to ALL when unset.
• id_scheme / data_element_id_scheme / org_unit_id_scheme: pick the identifier scheme for the payload (UID / CODE / NAME / ...).
• skip_audit=True → skipAudit=true.
• async_job=True → async=true: DHIS2 queues the import as a job and the returned envelope carries response.jobType / response.id. Poll with client.tasks.await_completion(envelope.task_ref()).

Returns a WebMessageResponse. For synchronous imports, envelope.import_count() gives ImportCount.imported / updated / ignored / deleted; envelope.conflicts() lists per-row rejections. Async imports return the task-ref envelope — poll it to completion to get the final report from DHIS2.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/data_values.py

async def stream(
    self,
    source: StreamSource,
    *,
    content_type: str = "application/json",
    dry_run: bool = False,
    preheat_cache: bool = True,
    import_strategy: str | None = None,
    atomic_mode: str | None = None,
    id_scheme: str | None = None,
    data_element_id_scheme: str | None = None,
    org_unit_id_scheme: str | None = None,
    skip_audit: bool = False,
    async_job: bool = False,
    chunk_size: int = _DEFAULT_CHUNK_SIZE,
) -> WebMessageResponse:
    """Stream `source` to `POST /api/dataValueSets` and return the typed envelope.

    `content_type` picks which DHIS2 parser handles the body (JSON / XML /
    CSV / ADX). Every param from the standard `/api/dataValueSets` surface
    is forwarded via query string:

    - `dry_run` → `dryRun=true`: validate without committing.
    - `preheat_cache=False` → `preheatCache=false`.
    - `import_strategy`: `CREATE` / `UPDATE` / `CREATE_AND_UPDATE` / `DELETE`.
    - `atomic_mode`: `ALL` rejects the whole import if any row is
      rejected; `OBJECT` commits the rows that pass and skips the rest.
      DHIS2 defaults to `ALL` when unset.
    - `id_scheme` / `data_element_id_scheme` / `org_unit_id_scheme`: pick
      the identifier scheme for the payload (`UID` / `CODE` / `NAME` / ...).
    - `skip_audit=True` → `skipAudit=true`.
    - `async_job=True` → `async=true`: DHIS2 queues the import as a job
      and the returned envelope carries `response.jobType` / `response.id`.
      Poll with `client.tasks.await_completion(envelope.task_ref())`.

    Returns a `WebMessageResponse`. For synchronous imports,
    `envelope.import_count()` gives `ImportCount.imported / updated /
    ignored / deleted`; `envelope.conflicts()` lists per-row rejections.
    Async imports return the task-ref envelope — poll it to completion
    to get the final report from DHIS2.
    """
    params: dict[str, Any] = {}
    if dry_run:
        params["dryRun"] = "true"
    if not preheat_cache:
        params["preheatCache"] = "false"
    if import_strategy is not None:
        params["importStrategy"] = import_strategy
    if atomic_mode is not None:
        params["atomicMode"] = atomic_mode
    if id_scheme is not None:
        params["idScheme"] = id_scheme
    if data_element_id_scheme is not None:
        params["dataElementIdScheme"] = data_element_id_scheme
    if org_unit_id_scheme is not None:
        params["orgUnitIdScheme"] = org_unit_id_scheme
    if skip_audit:
        params["skipAudit"] = "true"
    if async_job:
        params["async"] = "true"

    content = _coerce_stream_source(source, chunk_size=chunk_size)
    response = await self._client._request(  # noqa: SLF001 — accessor is intentionally tight with the client
        "POST",
        "/api/dataValueSets",
        params=params,
        content=content,
        extra_headers={"Content-Type": content_type},
    )
    raw = response.json() if response.content else {}
    return WebMessageResponse.model_validate(raw)

export(*, data_set=None, period=None, org_unit=None, children=False, start_date=None, end_date=None, last_updated=None, last_updated_duration=None, extra_params=None) async

GET /api/dataValueSets and return the parsed DataValueSet envelope.

The read side of the import handled by stream(). data_set, period, and org_unit each accept a single id or a sequence of ids, and repeat as dataSet=/period=/orgUnit= query params the way DHIS2 expects.

• children=True -> children=true: include the org units below each org_unit in the export.
• start_date / end_date (YYYY-MM-DD): export by date range instead of by period.
• last_updated (YYYY-MM-DD) / last_updated_duration (e.g. 2h, 1d): export only values touched since then.
• extra_params covers the rest of the surface (idScheme, dataElementIdScheme, includeDeleted, dataElementGroup, orgUnitGroup, ...). Pass a flat mapping or a list of 2-tuples.

Buffers the whole response into a typed DataValueSet. For a large export, stream it straight to storage with client.stream("GET", "/api/dataValueSets.json", sink, params=...) instead of materialising every row in memory.

Raises Dhis2ApiError on 4xx / 5xx.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/data_values.py

async def export(
    self,
    *,
    data_set: str | Sequence[str] | None = None,
    period: str | Sequence[str] | None = None,
    org_unit: str | Sequence[str] | None = None,
    children: bool = False,
    start_date: str | None = None,
    end_date: str | None = None,
    last_updated: str | None = None,
    last_updated_duration: str | None = None,
    extra_params: Mapping[str, Any] | None = None,
) -> DataValueSet:
    """GET `/api/dataValueSets` and return the parsed `DataValueSet` envelope.

    The read side of the import handled by `stream()`. `data_set`, `period`,
    and `org_unit` each accept a single id or a sequence of ids, and repeat
    as `dataSet=`/`period=`/`orgUnit=` query params the way DHIS2 expects.

    - `children=True` -> `children=true`: include the org units below each
      `org_unit` in the export.
    - `start_date` / `end_date` (`YYYY-MM-DD`): export by date range instead
      of by `period`.
    - `last_updated` (`YYYY-MM-DD`) / `last_updated_duration` (e.g. `2h`,
      `1d`): export only values touched since then.
    - `extra_params` covers the rest of the surface (`idScheme`,
      `dataElementIdScheme`, `includeDeleted`, `dataElementGroup`,
      `orgUnitGroup`, ...). Pass a flat mapping or a list of 2-tuples.

    Buffers the whole response into a typed `DataValueSet`. For a large
    export, stream it straight to storage with
    `client.stream("GET", "/api/dataValueSets.json", sink, params=...)`
    instead of materialising every row in memory.

    Raises `Dhis2ApiError` on 4xx / 5xx.
    """
    params: dict[str, Any] = {}
    for key, value in (("dataSet", data_set), ("period", period), ("orgUnit", org_unit)):
        if value is None:
            continue
        params[key] = [value] if isinstance(value, str) else list(value)
    if children:
        params["children"] = "true"
    if start_date is not None:
        params["startDate"] = start_date
    if end_date is not None:
        params["endDate"] = end_date
    if last_updated is not None:
        params["lastUpdated"] = last_updated
    if last_updated_duration is not None:
        params["lastUpdatedDuration"] = last_updated_duration
    if extra_params:
        params.update(extra_params)
    raw = await self._client.get_raw("/api/dataValueSets", params=params)
    return DataValueSet.model_validate(raw)

import_grouped_by_dataset(values, *, chunk_size=1000, force=False, skip_audit=False) async

Import typed DataValues grouped by dataset — BUGS.md #35 workaround for v43.

v43 added auto-target dataset detection on POST /api/dataValueSets (DefaultDataEntryService.autoTargetDataSet). When a posted DataValue's DataElement is referenced by 2+ DataSets, v43 aborts the entire chunk with 409 E8002 Data set detection failed. v41 + v42 silently picked one matching dataset and imported the row.

Workaround: pre-fetch the DataElement → DataSet membership map, group the input values by their DataSet (lexicographically-first DataSet id when a DE belongs to multiple — deterministic across runs), and POST each group with an explicit envelope {"dataSet": "<id>", "dataValues": [...]}. This shape is accepted by every DHIS2 major.

Splits each per-dataset group into chunk_size rows per POST so the body stays inside httpx2's 300 s read timeout. Returns one WebMessageResponse per chunk; callers that want aggregate counts should walk import_count() across the list and sum.

Skips values whose DataElement isn't in any DataSet (counted in the first response's import-count under ignored).

Source code in packages/dhis2w-client/src/dhis2w_client/v43/data_values.py

async def import_grouped_by_dataset(
    self,
    values: Sequence[DataValue],
    *,
    chunk_size: int = 1000,
    force: bool = False,
    skip_audit: bool = False,
) -> list[WebMessageResponse]:
    """Import typed `DataValue`s grouped by dataset — BUGS.md #35 workaround for v43.

    v43 added auto-target dataset detection on `POST /api/dataValueSets`
    (`DefaultDataEntryService.autoTargetDataSet`). When a posted DataValue's
    DataElement is referenced by 2+ DataSets, v43 aborts the entire chunk
    with `409 E8002 Data set detection failed`. v41 + v42 silently picked
    one matching dataset and imported the row.

    Workaround: pre-fetch the DataElement → DataSet membership map, group
    the input values by their DataSet (lexicographically-first DataSet id
    when a DE belongs to multiple — deterministic across runs), and POST
    each group with an explicit envelope `{"dataSet": "<id>", "dataValues":
    [...]}`. This shape is accepted by every DHIS2 major.

    Splits each per-dataset group into `chunk_size` rows per POST so the
    body stays inside httpx2's 300 s read timeout. Returns one
    `WebMessageResponse` per chunk; callers that want aggregate counts
    should walk `import_count()` across the list and sum.

    Skips values whose DataElement isn't in any DataSet (counted in the
    first response's import-count under `ignored`).
    """
    dataelement_to_dataset = await self._build_dataelement_to_dataset()
    grouped: dict[str, list[dict[str, Any]]] = {}
    for value in values:
        if value.dataElement is None:
            continue
        dataset_id = dataelement_to_dataset.get(value.dataElement)
        if dataset_id is None:
            continue
        grouped.setdefault(dataset_id, []).append(value.model_dump(by_alias=True, exclude_none=True, mode="json"))
    params: dict[str, Any] = {}
    if force:
        params["force"] = "true"
    if skip_audit:
        params["skipAudit"] = "true"
    responses: list[WebMessageResponse] = []
    for dataset_id, dumped in grouped.items():
        for start in range(0, len(dumped), chunk_size):
            chunk = dumped[start : start + chunk_size]
            raw = await self._client._request(  # noqa: SLF001
                "POST",
                "/api/dataValueSets",
                params=params,
                json={"dataSet": dataset_id, "dataValues": chunk},
            )
            body = raw.json() if raw.content else {}
            responses.append(WebMessageResponse.model_validate(body))
    return responses