Skip to content

Analytics streaming

AnalyticsAccessor on Dhis2Client.analytics — the /api/analytics* endpoint family: aggregate for a parsed pivot, event_query / enrollment_query for the tracker line lists, and stream / stream_to for chunked downloads that pipe the response body straight to disk without buffering it. Counterpart to Data values (import + export) for the read side of aggregate data.

When to reach for it

  • Exporting an analytics pivot too large to materialise as a single string in memory (city-level monthly population indicators, multi-year datasets, etc.).
  • Writing a CSV or JSON file an analytics tool downstream will read.
  • Snapshotting rawData for offline analysis (the rawData endpoint can produce hundreds of MB).

For small / interactive queries, client.analytics.aggregate(dx=..., pe=..., ou=...) returns a parsed Grid synchronously — see the parent Analytics page. Streaming is the right tool only when the response size makes the synchronous shape impractical.

Worked example — stream a CSV to disk

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:
    # Bytes-iterator interface: useful when you want to pipe somewhere else.
    async for chunk in client.analytics.stream(
        dx="fbfJHSPpUQD",
        pe="LAST_12_MONTHS",
        ou="ImspTQPwCqd",
        output_format="csv",
    ):
        ...

    # `stream_to(Path, ...)` writes directly; httpx2 never buffers the body.
    bytes_written = await client.analytics.stream_to(
        Path("./monthly-anc.csv"),
        dx="fbfJHSPpUQD",
        pe="LAST_12_MONTHS",
        ou="ImspTQPwCqd",
        output_format="csv",
    )
    print(f"wrote {bytes_written:,} bytes to ./monthly-anc.csv")

Output formats

  • csv — the most common downstream-tool target.
  • json — analytics pivot in JSON; equivalent to the parsed Grid but un-parsed.
  • rawData — DHIS2 returns rows in the source schema (data values, not pivoted cells). Use for offline re-analysis.

stream_to(..., output_format="rawData") works the same way; the suffix on the URL changes.

Event and enrollment line lists

event_query(program, ...) GETs /api/analytics/events/query/<program>, one row per event; enrollment_query(program, ...) GETs /api/analytics/enrollments/query/<program>, one row per enrollment. Both take the same dimension / filter tokens as the aggregate endpoint (pe:LAST_12_MONTHS, ou:<uid>, <deUid>:GT:5), a start_date / end_date bound, page / page_size, and return the parsed Grid. event_query also scopes by stage, output_type (EVENT / ENROLLMENT / TRACKED_ENTITY_INSTANCE) and event_status; both filter by program_status. extra_params carries the rest (aggregationType, ouMode, skipMeta, ...).

async with open_client(profile_from_env()) as client:
    events = await client.analytics.event_query(
        "IpHINAT79UW",
        dimension=["ou:ImspTQPwCqd", "pe:LAST_12_MONTHS"],
        output_type="EVENT",
        page_size=50,
    )
    headers = [header.name for header in events.headers or []]
    for row in events.rows or []:
        print(dict(zip(headers, row, strict=False)))

analytics_stream

/api/analytics* access — client.analytics.aggregate + client.analytics.stream_to.

Two flavours of analytics call live here:

  • aggregate(dx=..., pe=..., ou=...) returns a parsed Grid. Right for small / medium responses where you want typed cells immediately.
  • stream_to(destination=...) writes the response straight to disk without buffering. Right for very large responses (yearly district-level pivots, etc.).

DHIS2's analytics endpoint family can return very large responses on reasonable queries. The two methods cover the speed/memory trade-off explicitly so callers don't have to think about it.

Endpoints covered (pass the full path including extension / sub-resource):

  • /api/analytics.json (default)
  • /api/analytics.csv
  • /api/analytics.xlsx
  • /api/analytics/rawData.json (requires .json suffix, see BUGS.md #1)
  • /api/analytics/dataValueSet.json (same)
  • /api/analytics/events/query/<program>.json

params (on stream_to) is forwarded verbatim — DHIS2's repeated-param pattern (dimension=dx:...&dimension=pe:...&dimension=ou:...) expects either a mapping with list values ({"dimension": ["dx:...", ...]}) or a list of 2-tuples ([("dimension", "dx:..."), ...]).

Classes

AnalyticsAccessor

Dhis2Client.analytics — typed query (aggregate) + streaming download (stream_to).

Use aggregate(...) for a parsed Grid (small / medium responses); stream_to(...) writes straight to disk for very large pivots without buffering. Both accept the same dimension dx/pe/ou convenience kwargs plus arbitrary extra params for the rest of DHIS2's analytics flag surface.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
 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
class AnalyticsAccessor:
    """`Dhis2Client.analytics` — typed query (`aggregate`) + streaming download (`stream_to`).

    Use `aggregate(...)` for a parsed `Grid` (small / medium responses);
    `stream_to(...)` writes straight to disk for very large pivots
    without buffering. Both accept the same dimension dx/pe/ou
    convenience kwargs plus arbitrary extra params for the rest of
    DHIS2's analytics flag surface.
    """

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

    async def aggregate(
        self,
        *,
        dx: str | Sequence[str] | None = None,
        pe: str | Sequence[str] | None = None,
        ou: str | Sequence[str] | None = None,
        endpoint: str = "/api/analytics.json",
        extra_params: Mapping[str, Any] | None = None,
    ) -> Grid:
        """Run an analytics query and return the parsed `Grid` envelope.

        `dx` / `pe` / `ou` are convenience for the three core dimensions.
        Each accepts a single id (`"fbfJHSPpUQD"`), a colon-joined token
        (`"LAST_12_MONTHS"`), a list of ids (`["fbfJHSPpUQD", "cYeuwXTCPkU"]`),
        or `None`. Multiple values within a dimension are colon-joined into
        the single `dimension=<axis>:v1;v2;v3` form DHIS2 expects.

        `extra_params` covers the rest (`aggregationType`, `outputIdScheme`,
        `displayProperty`, `skipMeta`, etc.). Pass either a flat
        `{"aggregationType": "SUM"}` or a list of 2-tuples for repeated keys.

        Raises `Dhis2ApiError` on 4xx / 5xx.
        """
        dimension: list[str] = []
        for axis, value in (("dx", dx), ("pe", pe), ("ou", ou)):
            if value is None:
                continue
            ids = [value] if isinstance(value, str) else list(value)
            if ids:
                dimension.append(f"{axis}:{';'.join(ids)}")
        params: dict[str, Any] = {"dimension": dimension} if dimension else {}
        if extra_params:
            params.update(extra_params)
        raw = await self._client.get_raw(endpoint, params=params)
        return Grid.model_validate(raw)

    async def stream_to(
        self,
        destination: Path,
        *,
        params: AnalyticsQuery,
        endpoint: str = "/api/analytics.json",
        chunk_size: int = _DEFAULT_CHUNK_SIZE,
    ) -> int:
        """Stream a GET on `endpoint` straight to `destination`; return bytes written.

        `params` forwards exactly what DHIS2 accepts — use a list of
        2-tuples when you need repeated `dimension=` params, or a mapping
        whose values are lists when a key can appear more than once.

        `endpoint` is the full path including extension + sub-resource
        (`/api/analytics.csv`, `/api/analytics/rawData.json`, ...).
        `client.system.info()` uses the same httpx2 pool, so auth + retry +
        pool-tuning all still apply.

        Raises `Dhis2ApiError` on 4xx / 5xx (the error body is buffered —
        errors are small and readable).
        """
        return await stream_to_sink(
            self._client,
            "GET",
            endpoint,
            destination,
            params=params,
            chunk_size=chunk_size,
        )

    async def event_query(
        self,
        program: str,
        *,
        stage: str | None = None,
        dimension: str | Sequence[str] | None = None,
        filter: str | Sequence[str] | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
        output_type: str | None = None,
        event_status: str | None = None,
        program_status: str | None = None,
        page: int | None = None,
        page_size: int | None = None,
        extra_params: Mapping[str, Any] | None = None,
    ) -> Grid:
        """Run an event analytics query and return the parsed `Grid` envelope.

        GET `/api/analytics/events/query/<program>` — the event-level analytics
        read (one row per event). `dimension` and `filter` each accept a single
        pre-formed token (`"pe:LAST_12_MONTHS"`, `"ou:ImspTQPwCqd"`,
        `"<deUid>:GT:5"`) or a sequence, repeating as `dimension=` / `filter=`
        params. `stage` scopes to one program stage. `start_date` / `end_date`
        bound the event date (`YYYY-MM-DD`). `output_type` picks the row grain
        (`EVENT` / `ENROLLMENT` / `TRACKED_ENTITY_INSTANCE`). `event_status`
        and `program_status` filter by state. `extra_params` covers the rest
        (`aggregationType`, `outputIdScheme`, `ouMode`, `coordinatesOnly`, ...).

        Raises `Dhis2ApiError` on 4xx / 5xx.
        """
        return await self._events_query(
            "events",
            program,
            stage=stage,
            dimension=dimension,
            filter=filter,
            start_date=start_date,
            end_date=end_date,
            output_type=output_type,
            event_status=event_status,
            program_status=program_status,
            page=page,
            page_size=page_size,
            extra_params=extra_params,
        )

    async def enrollment_query(
        self,
        program: str,
        *,
        dimension: str | Sequence[str] | None = None,
        filter: str | Sequence[str] | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
        output_type: str | None = None,
        program_status: str | None = None,
        page: int | None = None,
        page_size: int | None = None,
        extra_params: Mapping[str, Any] | None = None,
    ) -> Grid:
        """Run an enrollment analytics query and return the parsed `Grid` envelope.

        GET `/api/analytics/enrollments/query/<program>` — the enrollment-level
        analytics read (one row per enrollment, program-indicator and
        repeated-stage values flattened across the enrollment). `dimension` and
        `filter` follow the same single-or-sequence, pre-formed-token shape as
        `event_query`. `start_date` / `end_date` bound the enrollment date.
        `program_status` filters by enrollment state. `extra_params` covers the
        rest of the surface.

        Raises `Dhis2ApiError` on 4xx / 5xx.
        """
        return await self._events_query(
            "enrollments",
            program,
            stage=None,
            dimension=dimension,
            filter=filter,
            start_date=start_date,
            end_date=end_date,
            output_type=output_type,
            event_status=None,
            program_status=program_status,
            page=page,
            page_size=page_size,
            extra_params=extra_params,
        )

    async def _events_query(
        self,
        mode: str,
        program: str,
        *,
        stage: str | None,
        dimension: str | Sequence[str] | None,
        filter: str | Sequence[str] | None,
        start_date: str | None,
        end_date: str | None,
        output_type: str | None,
        event_status: str | None,
        program_status: str | None,
        page: int | None,
        page_size: int | None,
        extra_params: Mapping[str, Any] | None,
    ) -> Grid:
        """Build + run an `/api/analytics/{events,enrollments}/query/<program>` GET."""
        params: dict[str, Any] = {}
        _repeat(params, "dimension", dimension)
        _repeat(params, "filter", filter)
        if stage is not None:
            params["stage"] = stage
        if start_date is not None:
            params["startDate"] = start_date
        if end_date is not None:
            params["endDate"] = end_date
        if output_type is not None:
            params["outputType"] = output_type
        if event_status is not None:
            params["eventStatus"] = event_status
        if program_status is not None:
            params["programStatus"] = program_status
        if page is not None:
            params["page"] = page
        if page_size is not None:
            params["pageSize"] = page_size
        if extra_params:
            params.update(extra_params)
        # `.json` suffix keeps the events/enrollments query on the JSON grid
        # response across majors (BUGS.md #1).
        raw = await self._client.get_raw(f"/api/analytics/{mode}/query/{program}.json", params=params)
        return Grid.model_validate(raw)
Methods:
__init__(client)

Bind to the sharing client.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
58
59
60
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client."""
    self._client = client
aggregate(*, dx=None, pe=None, ou=None, endpoint='/api/analytics.json', extra_params=None) async

Run an analytics query and return the parsed Grid envelope.

dx / pe / ou are convenience for the three core dimensions. Each accepts a single id ("fbfJHSPpUQD"), a colon-joined token ("LAST_12_MONTHS"), a list of ids (["fbfJHSPpUQD", "cYeuwXTCPkU"]), or None. Multiple values within a dimension are colon-joined into the single dimension=<axis>:v1;v2;v3 form DHIS2 expects.

extra_params covers the rest (aggregationType, outputIdScheme, displayProperty, skipMeta, etc.). Pass either a flat {"aggregationType": "SUM"} or a list of 2-tuples for repeated keys.

Raises Dhis2ApiError on 4xx / 5xx.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
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
async def aggregate(
    self,
    *,
    dx: str | Sequence[str] | None = None,
    pe: str | Sequence[str] | None = None,
    ou: str | Sequence[str] | None = None,
    endpoint: str = "/api/analytics.json",
    extra_params: Mapping[str, Any] | None = None,
) -> Grid:
    """Run an analytics query and return the parsed `Grid` envelope.

    `dx` / `pe` / `ou` are convenience for the three core dimensions.
    Each accepts a single id (`"fbfJHSPpUQD"`), a colon-joined token
    (`"LAST_12_MONTHS"`), a list of ids (`["fbfJHSPpUQD", "cYeuwXTCPkU"]`),
    or `None`. Multiple values within a dimension are colon-joined into
    the single `dimension=<axis>:v1;v2;v3` form DHIS2 expects.

    `extra_params` covers the rest (`aggregationType`, `outputIdScheme`,
    `displayProperty`, `skipMeta`, etc.). Pass either a flat
    `{"aggregationType": "SUM"}` or a list of 2-tuples for repeated keys.

    Raises `Dhis2ApiError` on 4xx / 5xx.
    """
    dimension: list[str] = []
    for axis, value in (("dx", dx), ("pe", pe), ("ou", ou)):
        if value is None:
            continue
        ids = [value] if isinstance(value, str) else list(value)
        if ids:
            dimension.append(f"{axis}:{';'.join(ids)}")
    params: dict[str, Any] = {"dimension": dimension} if dimension else {}
    if extra_params:
        params.update(extra_params)
    raw = await self._client.get_raw(endpoint, params=params)
    return Grid.model_validate(raw)
stream_to(destination, *, params, endpoint='/api/analytics.json', chunk_size=_DEFAULT_CHUNK_SIZE) async

Stream a GET on endpoint straight to destination; return bytes written.

params forwards exactly what DHIS2 accepts — use a list of 2-tuples when you need repeated dimension= params, or a mapping whose values are lists when a key can appear more than once.

endpoint is the full path including extension + sub-resource (/api/analytics.csv, /api/analytics/rawData.json, ...). client.system.info() uses the same httpx2 pool, so auth + retry + pool-tuning all still apply.

Raises Dhis2ApiError on 4xx / 5xx (the error body is buffered — errors are small and readable).

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
 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
async def stream_to(
    self,
    destination: Path,
    *,
    params: AnalyticsQuery,
    endpoint: str = "/api/analytics.json",
    chunk_size: int = _DEFAULT_CHUNK_SIZE,
) -> int:
    """Stream a GET on `endpoint` straight to `destination`; return bytes written.

    `params` forwards exactly what DHIS2 accepts — use a list of
    2-tuples when you need repeated `dimension=` params, or a mapping
    whose values are lists when a key can appear more than once.

    `endpoint` is the full path including extension + sub-resource
    (`/api/analytics.csv`, `/api/analytics/rawData.json`, ...).
    `client.system.info()` uses the same httpx2 pool, so auth + retry +
    pool-tuning all still apply.

    Raises `Dhis2ApiError` on 4xx / 5xx (the error body is buffered —
    errors are small and readable).
    """
    return await stream_to_sink(
        self._client,
        "GET",
        endpoint,
        destination,
        params=params,
        chunk_size=chunk_size,
    )
event_query(program, *, stage=None, dimension=None, filter=None, start_date=None, end_date=None, output_type=None, event_status=None, program_status=None, page=None, page_size=None, extra_params=None) async

Run an event analytics query and return the parsed Grid envelope.

GET /api/analytics/events/query/<program> — the event-level analytics read (one row per event). dimension and filter each accept a single pre-formed token ("pe:LAST_12_MONTHS", "ou:ImspTQPwCqd", "<deUid>:GT:5") or a sequence, repeating as dimension= / filter= params. stage scopes to one program stage. start_date / end_date bound the event date (YYYY-MM-DD). output_type picks the row grain (EVENT / ENROLLMENT / TRACKED_ENTITY_INSTANCE). event_status and program_status filter by state. extra_params covers the rest (aggregationType, outputIdScheme, ouMode, coordinatesOnly, ...).

Raises Dhis2ApiError on 4xx / 5xx.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
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
async def event_query(
    self,
    program: str,
    *,
    stage: str | None = None,
    dimension: str | Sequence[str] | None = None,
    filter: str | Sequence[str] | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    output_type: str | None = None,
    event_status: str | None = None,
    program_status: str | None = None,
    page: int | None = None,
    page_size: int | None = None,
    extra_params: Mapping[str, Any] | None = None,
) -> Grid:
    """Run an event analytics query and return the parsed `Grid` envelope.

    GET `/api/analytics/events/query/<program>` — the event-level analytics
    read (one row per event). `dimension` and `filter` each accept a single
    pre-formed token (`"pe:LAST_12_MONTHS"`, `"ou:ImspTQPwCqd"`,
    `"<deUid>:GT:5"`) or a sequence, repeating as `dimension=` / `filter=`
    params. `stage` scopes to one program stage. `start_date` / `end_date`
    bound the event date (`YYYY-MM-DD`). `output_type` picks the row grain
    (`EVENT` / `ENROLLMENT` / `TRACKED_ENTITY_INSTANCE`). `event_status`
    and `program_status` filter by state. `extra_params` covers the rest
    (`aggregationType`, `outputIdScheme`, `ouMode`, `coordinatesOnly`, ...).

    Raises `Dhis2ApiError` on 4xx / 5xx.
    """
    return await self._events_query(
        "events",
        program,
        stage=stage,
        dimension=dimension,
        filter=filter,
        start_date=start_date,
        end_date=end_date,
        output_type=output_type,
        event_status=event_status,
        program_status=program_status,
        page=page,
        page_size=page_size,
        extra_params=extra_params,
    )
enrollment_query(program, *, dimension=None, filter=None, start_date=None, end_date=None, output_type=None, program_status=None, page=None, page_size=None, extra_params=None) async

Run an enrollment analytics query and return the parsed Grid envelope.

GET /api/analytics/enrollments/query/<program> — the enrollment-level analytics read (one row per enrollment, program-indicator and repeated-stage values flattened across the enrollment). dimension and filter follow the same single-or-sequence, pre-formed-token shape as event_query. start_date / end_date bound the enrollment date. program_status filters by enrollment state. extra_params covers the rest of the surface.

Raises Dhis2ApiError on 4xx / 5xx.

Source code in packages/dhis2w-client/src/dhis2w_client/v43/analytics_stream.py 
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
async def enrollment_query(
    self,
    program: str,
    *,
    dimension: str | Sequence[str] | None = None,
    filter: str | Sequence[str] | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
    output_type: str | None = None,
    program_status: str | None = None,
    page: int | None = None,
    page_size: int | None = None,
    extra_params: Mapping[str, Any] | None = None,
) -> Grid:
    """Run an enrollment analytics query and return the parsed `Grid` envelope.

    GET `/api/analytics/enrollments/query/<program>` — the enrollment-level
    analytics read (one row per enrollment, program-indicator and
    repeated-stage values flattened across the enrollment). `dimension` and
    `filter` follow the same single-or-sequence, pre-formed-token shape as
    `event_query`. `start_date` / `end_date` bound the enrollment date.
    `program_status` filters by enrollment state. `extra_params` covers the
    rest of the surface.

    Raises `Dhis2ApiError` on 4xx / 5xx.
    """
    return await self._events_query(
        "enrollments",
        program,
        stage=None,
        dimension=dimension,
        filter=filter,
        start_date=start_date,
        end_date=end_date,
        output_type=output_type,
        event_status=None,
        program_status=program_status,
        page=page,
        page_size=page_size,
        extra_params=extra_params,
    )

Functions: