Skip to content

SQL views

SqlViewsAccessor on Dhis2Client.sql_views covers the workflow surface over /api/sqlViews — list_views, get, execute, refresh, create, delete. SqlViewRunner (also reachable as client.sql_views.runner) is a thin facade for iterative SQL debugging: run(uid, **vars) for saved views, adhoc(name, sql, **vars) for throwaway register-execute-delete flows.

SqlViewResult is the typed parser over DHIS2's listGrid response: columns names the headers, rows keeps them as positional list[list[Any]] (SQL column shape is inherently dynamic), .as_dicts() pivots into name-keyed dicts for ad-hoc inspection, .column_values(name) projects a single column.

DHIS2 quirks worth remembering:

  • Variable and criteria values are sanitised to alphanumeric characters only — wildcards live in the SQL template ('%${q}%'), not the variable value.
  • VIEW and MATERIALIZED_VIEW creation is lazy: the DB object doesn't exist until the first POST /execute. Call refresh() right after create() when a caller plans to read /data immediately afterwards.
  • The DHIS2 SQL allowlist blocks DELETE / UPDATE / INSERT / DROP / etc. For fully free-form queries, connect to Postgres directly.

sql_views

Execution-surface helpers for DHIS2 SqlView workflows.

/api/sqlViews exposes three kinds of saved SQL:

  • VIEW — a standard SQL view, materialised in DHIS2's Postgres schema the first time it is executed.
  • MATERIALIZED_VIEW — persisted result set; refreshable on demand via POST /execute.
  • QUERY — stored SQL executed ad-hoc with optional ${var} substitutions.

Generic CRUD is already covered by the generated accessor (client.resources.sql_views). This module layers the workflow surface that generic CRUD doesn't provide:

  • list_views() — every SqlView with type + sqlQuery eagerly loaded.
  • get(uid) — one view.
  • execute(uid, *, variables=..., criteria=...) — run and return a typed SqlViewResult (columns + rows + title).
  • refresh(uid) — materialised-view refresh + first-time VIEW creation.
  • create(sql_view) / delete(uid) — so one-shot register-run-delete flows work without round-tripping the generic accessor.

Plus SqlViewRunner — a small facade for iterative SQL debugging:

  • run(uid, **variables) — execute a saved view with keyword-style var substitutions.
  • adhoc(name, sql, **variables) — register a throwaway SqlView, execute once, delete on the way out. Useful when iterating on SQL without littering the instance with test metadata.

SqlViewResult model:

{
  "listGrid": {
    "title": ...,
    "subtitle": ...,
    "headers": [{"name": "column1", "type": "TEXT", ...}, ...],
    "rows": [[col1, col2, ...], ...],
    "height": N, "width": M
  }
}

Rows are inherently schema-less (the SQL defines the columns), so rows: list[list[Any]] is the canonical shape — the paired columns list tells callers what each position means. .as_dicts() pivots into column-name-keyed dicts when preferred.

BUGS.md-worthy behaviours to watch for:

  • Variable and criteria values are sanitised server-side to alphanumeric characters only — wildcards and punctuation live in the SQL template, not the variable value ('%${q}%', pass q=abc).
  • VIEW and MATERIALIZED_VIEW creation happens lazily: first GET on /data creates the database object. Call refresh(uid) right after create() when you want the DB view to exist immediately.
  • DHIS2's SQL allowlist blocks DELETE/UPDATE/INSERT/DROP/etc. For fully free-form queries, go around the allowlist with a direct Postgres connection (see the PG-injector example).

Classes

SqlViewColumn

Bases: BaseModel

One column in a SqlView execution result — name + declared SQL type.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
79
80
81
82
83
84
85
86
87
88
class SqlViewColumn(BaseModel):
    """One column in a SqlView execution result — name + declared SQL type."""

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    name: str
    column: str | None = None
    type: str | None = None
    meta: bool | None = None
    hidden: bool | None = None

SqlViewResult

Bases: BaseModel

Typed execution result from /api/sqlViews/{uid}/data.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
 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
class SqlViewResult(BaseModel):
    """Typed execution result from `/api/sqlViews/{uid}/data`."""

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    title: str | None = None
    subtitle: str | None = None
    columns: list[SqlViewColumn] = Field(default_factory=list)
    rows: list[list[Any]] = Field(default_factory=list)
    height: int = 0
    width: int = 0

    @classmethod
    def from_api(cls, body: dict[str, Any]) -> SqlViewResult:
        """Parse DHIS2's `listGrid`-nested response envelope into a typed result."""
        grid_raw = body.get("listGrid")
        grid = grid_raw if isinstance(grid_raw, dict) else body
        header_rows = grid.get("headers") or []
        columns: list[SqlViewColumn] = []
        for header in header_rows:
            if isinstance(header, dict):
                columns.append(SqlViewColumn.model_validate(header))
        rows: list[list[Any]] = []
        for row in grid.get("rows") or []:
            if isinstance(row, list):
                rows.append(list(row))
        height = grid.get("height")
        width = grid.get("width")
        return cls(
            title=grid.get("title") if isinstance(grid.get("title"), str) else None,
            subtitle=grid.get("subtitle") if isinstance(grid.get("subtitle"), str) else None,
            columns=columns,
            rows=rows,
            height=int(height) if isinstance(height, int) else len(rows),
            width=int(width) if isinstance(width, int) else len(columns),
        )

    def as_dicts(self) -> list[dict[str, Any]]:
        """Pivot rows into column-name-keyed dicts — convenience for ad-hoc inspection.

        The raw `rows` attribute keeps DHIS2's positional shape (zero per-row
        allocation for streaming). Use this helper at the call site when you
        want name-keyed access; do not pass the dicts back across module
        boundaries — the typed model is the canonical form.
        """
        names = [column.name for column in self.columns]
        return [dict(zip(names, row, strict=False)) for row in self.rows]

    def column_values(self, column_name: str) -> list[Any]:
        """Pull one column out across every row — raises KeyError if unknown."""
        names = [column.name for column in self.columns]
        try:
            index = names.index(column_name)
        except ValueError as exc:
            raise KeyError(f"column {column_name!r} not in SqlViewResult ({names!r})") from exc
        return [row[index] if index < len(row) else None for row in self.rows]
Functions
from_api(body) classmethod

Parse DHIS2's listGrid-nested response envelope into a typed result.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@classmethod
def from_api(cls, body: dict[str, Any]) -> SqlViewResult:
    """Parse DHIS2's `listGrid`-nested response envelope into a typed result."""
    grid_raw = body.get("listGrid")
    grid = grid_raw if isinstance(grid_raw, dict) else body
    header_rows = grid.get("headers") or []
    columns: list[SqlViewColumn] = []
    for header in header_rows:
        if isinstance(header, dict):
            columns.append(SqlViewColumn.model_validate(header))
    rows: list[list[Any]] = []
    for row in grid.get("rows") or []:
        if isinstance(row, list):
            rows.append(list(row))
    height = grid.get("height")
    width = grid.get("width")
    return cls(
        title=grid.get("title") if isinstance(grid.get("title"), str) else None,
        subtitle=grid.get("subtitle") if isinstance(grid.get("subtitle"), str) else None,
        columns=columns,
        rows=rows,
        height=int(height) if isinstance(height, int) else len(rows),
        width=int(width) if isinstance(width, int) else len(columns),
    )
as_dicts()

Pivot rows into column-name-keyed dicts — convenience for ad-hoc inspection.

The raw rows attribute keeps DHIS2's positional shape (zero per-row allocation for streaming). Use this helper at the call site when you want name-keyed access; do not pass the dicts back across module boundaries — the typed model is the canonical form.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
128
129
130
131
132
133
134
135
136
137
def as_dicts(self) -> list[dict[str, Any]]:
    """Pivot rows into column-name-keyed dicts — convenience for ad-hoc inspection.

    The raw `rows` attribute keeps DHIS2's positional shape (zero per-row
    allocation for streaming). Use this helper at the call site when you
    want name-keyed access; do not pass the dicts back across module
    boundaries — the typed model is the canonical form.
    """
    names = [column.name for column in self.columns]
    return [dict(zip(names, row, strict=False)) for row in self.rows]
column_values(column_name)

Pull one column out across every row — raises KeyError if unknown.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
139
140
141
142
143
144
145
146
def column_values(self, column_name: str) -> list[Any]:
    """Pull one column out across every row — raises KeyError if unknown."""
    names = [column.name for column in self.columns]
    try:
        index = names.index(column_name)
    except ValueError as exc:
        raise KeyError(f"column {column_name!r} not in SqlViewResult ({names!r})") from exc
    return [row[index] if index < len(row) else None for row in self.rows]

SqlViewsAccessor

Dhis2Client.sql_views — execution + lifecycle helpers over DHIS2 SqlViews.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
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
class SqlViewsAccessor:
    """`Dhis2Client.sql_views` — execution + lifecycle helpers over DHIS2 SqlViews."""

    def __init__(self, client: Dhis2Client) -> None:
        """Bind to the sharing client — reuses its auth + HTTP pool."""
        self._client = client
        self.runner: SqlViewRunner = SqlViewRunner(self)

    async def list_views(self, *, view_type: SqlViewType | str | None = None) -> list[SqlView]:
        """List every SqlView, optionally filtered by type. Sorted by name."""
        filters: list[str] | None = None
        if view_type is not None:
            value = view_type.value if isinstance(view_type, SqlViewType) else view_type
            filters = [f"type:eq:{value}"]
        return cast(
            list[SqlView],
            await self._client.resources.sql_views.list(
                fields=_VIEW_FIELDS,
                filters=filters,
                order=["name:asc"],
                paging=False,
            ),
        )

    async def get(self, uid: str) -> SqlView:
        """Fetch one SqlView, including its `sqlQuery` text."""
        raw = await self._client.get_raw(f"/api/sqlViews/{uid}", params={"fields": _VIEW_FIELDS})
        return SqlView.model_validate(raw)

    async def execute(
        self,
        uid: str,
        *,
        variables: Mapping[str, str] | None = None,
        criteria: Mapping[str, str] | None = None,
    ) -> SqlViewResult:
        """Execute a SqlView and return its result grid.

        `variables` populate `${name}` placeholders on `QUERY` views —
        values are alphanumeric-only (DHIS2 strips punctuation server-side;
        keep wildcards in the SQL template).

        `criteria` filter the output of `VIEW` and `MATERIALIZED_VIEW`
        executions by column value. Passed as repeated `?criteria=col:val`
        query params.
        """
        params: list[tuple[str, str]] = []
        if variables:
            for key, value in variables.items():
                params.append(("var", f"{key}:{value}"))
        if criteria:
            for key, value in criteria.items():
                params.append(("criteria", f"{key}:{value}"))
        raw = await self._client.get_raw(f"/api/sqlViews/{uid}/data", params=dict(params) if params else None)
        return SqlViewResult.from_api(raw)

    async def refresh(self, uid: str) -> WebMessageResponse:
        """Refresh a `MATERIALIZED_VIEW` (or create a lazy `VIEW` on first call).

        DHIS2 creates the underlying Postgres view on the first execute
        call. Invoking this right after `create()` guarantees the view
        exists in the DB before any query hits `/data`.
        """
        raw = await self._client.post_raw(f"/api/sqlViews/{uid}/execute")
        return WebMessageResponse.model_validate(raw)

    async def create(self, sql_view: SqlView) -> SqlView:
        """POST a new SqlView; returns the created view with server-assigned metadata."""
        body = sql_view.model_dump(by_alias=True, exclude_none=True, mode="json")
        await self._client.post_raw("/api/sqlViews", body=body)
        if sql_view.id is None:
            raise ValueError("SqlView.id must be set for create() so the caller can round-trip the fetched model")
        return await self.get(sql_view.id)

    async def delete(self, uid: str) -> None:
        """DELETE a SqlView by UID. No-op return on success."""
        await self._client.resources.sql_views.delete(uid)
Functions
__init__(client)

Bind to the sharing client — reuses its auth + HTTP pool.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
152
153
154
155
def __init__(self, client: Dhis2Client) -> None:
    """Bind to the sharing client — reuses its auth + HTTP pool."""
    self._client = client
    self.runner: SqlViewRunner = SqlViewRunner(self)
list_views(*, view_type=None) async

List every SqlView, optionally filtered by type. Sorted by name.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
async def list_views(self, *, view_type: SqlViewType | str | None = None) -> list[SqlView]:
    """List every SqlView, optionally filtered by type. Sorted by name."""
    filters: list[str] | None = None
    if view_type is not None:
        value = view_type.value if isinstance(view_type, SqlViewType) else view_type
        filters = [f"type:eq:{value}"]
    return cast(
        list[SqlView],
        await self._client.resources.sql_views.list(
            fields=_VIEW_FIELDS,
            filters=filters,
            order=["name:asc"],
            paging=False,
        ),
    )
get(uid) async

Fetch one SqlView, including its sqlQuery text.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
173
174
175
176
async def get(self, uid: str) -> SqlView:
    """Fetch one SqlView, including its `sqlQuery` text."""
    raw = await self._client.get_raw(f"/api/sqlViews/{uid}", params={"fields": _VIEW_FIELDS})
    return SqlView.model_validate(raw)
execute(uid, *, variables=None, criteria=None) async

Execute a SqlView and return its result grid.

variables populate ${name} placeholders on QUERY views — values are alphanumeric-only (DHIS2 strips punctuation server-side; keep wildcards in the SQL template).

criteria filter the output of VIEW and MATERIALIZED_VIEW executions by column value. Passed as repeated ?criteria=col:val query params.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
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
async def execute(
    self,
    uid: str,
    *,
    variables: Mapping[str, str] | None = None,
    criteria: Mapping[str, str] | None = None,
) -> SqlViewResult:
    """Execute a SqlView and return its result grid.

    `variables` populate `${name}` placeholders on `QUERY` views —
    values are alphanumeric-only (DHIS2 strips punctuation server-side;
    keep wildcards in the SQL template).

    `criteria` filter the output of `VIEW` and `MATERIALIZED_VIEW`
    executions by column value. Passed as repeated `?criteria=col:val`
    query params.
    """
    params: list[tuple[str, str]] = []
    if variables:
        for key, value in variables.items():
            params.append(("var", f"{key}:{value}"))
    if criteria:
        for key, value in criteria.items():
            params.append(("criteria", f"{key}:{value}"))
    raw = await self._client.get_raw(f"/api/sqlViews/{uid}/data", params=dict(params) if params else None)
    return SqlViewResult.from_api(raw)
refresh(uid) async

Refresh a MATERIALIZED_VIEW (or create a lazy VIEW on first call).

DHIS2 creates the underlying Postgres view on the first execute call. Invoking this right after create() guarantees the view exists in the DB before any query hits /data.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
205
206
207
208
209
210
211
212
213
async def refresh(self, uid: str) -> WebMessageResponse:
    """Refresh a `MATERIALIZED_VIEW` (or create a lazy `VIEW` on first call).

    DHIS2 creates the underlying Postgres view on the first execute
    call. Invoking this right after `create()` guarantees the view
    exists in the DB before any query hits `/data`.
    """
    raw = await self._client.post_raw(f"/api/sqlViews/{uid}/execute")
    return WebMessageResponse.model_validate(raw)
create(sql_view) async

POST a new SqlView; returns the created view with server-assigned metadata.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
215
216
217
218
219
220
221
async def create(self, sql_view: SqlView) -> SqlView:
    """POST a new SqlView; returns the created view with server-assigned metadata."""
    body = sql_view.model_dump(by_alias=True, exclude_none=True, mode="json")
    await self._client.post_raw("/api/sqlViews", body=body)
    if sql_view.id is None:
        raise ValueError("SqlView.id must be set for create() so the caller can round-trip the fetched model")
    return await self.get(sql_view.id)
delete(uid) async

DELETE a SqlView by UID. No-op return on success.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
223
224
225
async def delete(self, uid: str) -> None:
    """DELETE a SqlView by UID. No-op return on success."""
    await self._client.resources.sql_views.delete(uid)

SqlViewRunner

Lightweight facade for interactive SQL view execution.

The accessor (client.sql_views) is the full API. This runner wraps it with two debugging shortcuts:

  • run(uid, **variables) — execute a saved view, passing ${name} substitutions as kwargs.
  • adhoc(name, sql, **variables) — register a throwaway SqlView, execute it once, delete it on the way out. Perfect for iterating on SQL without leaving test metadata behind.

Use standalone: runner = SqlViewRunner(client.sql_views), or reach it off the accessor: runner = client.sql_views.runner.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
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
class SqlViewRunner:
    """Lightweight facade for interactive SQL view execution.

    The accessor (`client.sql_views`) is the full API. This runner wraps
    it with two debugging shortcuts:

    - `run(uid, **variables)` — execute a saved view, passing `${name}`
      substitutions as kwargs.
    - `adhoc(name, sql, **variables)` — register a throwaway SqlView,
      execute it once, delete it on the way out. Perfect for iterating
      on SQL without leaving test metadata behind.

    Use standalone: `runner = SqlViewRunner(client.sql_views)`, or
    reach it off the accessor: `runner = client.sql_views.runner`.
    """

    def __init__(self, accessor: SqlViewsAccessor) -> None:
        """Bind to the owning accessor — the runner delegates every call back."""
        self._accessor = accessor

    async def run(self, uid: str, **variables: str) -> SqlViewResult:
        """Execute a saved view by UID, passing `${name}` substitutions as kwargs."""
        return await self._accessor.execute(uid, variables=variables if variables else None)

    async def run_with_criteria(
        self,
        uid: str,
        *,
        criteria: Mapping[str, str],
        **variables: str,
    ) -> SqlViewResult:
        """Execute a saved view with both variable substitutions and column criteria."""
        return await self._accessor.execute(
            uid,
            variables=variables if variables else None,
            criteria=criteria,
        )

    async def adhoc(
        self,
        name: str,
        sql: str,
        *,
        view_type: SqlViewType | str = SqlViewType.QUERY,
        keep: bool = False,
        **variables: str,
    ) -> SqlViewResult:
        """Register a throwaway SqlView, execute it once, then delete it.

        Set `keep=True` to leave the view in place afterwards (handy when
        the next iteration wants to inspect it via the DHIS2 UI).

        Subject to the DHIS2 SQL allowlist — for fully free-form SQL, use
        the Postgres injector example instead.
        """
        from dhis2w_client.uids import generate_uid

        candidate = SqlView.model_validate(
            {
                "id": generate_uid(),
                "name": name,
                "type": view_type.value if isinstance(view_type, SqlViewType) else view_type,
                "sqlQuery": sql,
            },
        )
        created = await self._accessor.create(candidate)
        if created.id is None:
            raise RuntimeError("adhoc create did not return a UID — server may have rejected the view")
        try:
            if created.type in (SqlViewType.VIEW, SqlViewType.MATERIALIZED_VIEW):
                await self._accessor.refresh(created.id)
            return await self._accessor.execute(created.id, variables=variables if variables else None)
        finally:
            if not keep:
                await self._accessor.delete(created.id)
Functions
__init__(accessor)

Bind to the owning accessor — the runner delegates every call back.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
244
245
246
def __init__(self, accessor: SqlViewsAccessor) -> None:
    """Bind to the owning accessor — the runner delegates every call back."""
    self._accessor = accessor
run(uid, **variables) async

Execute a saved view by UID, passing ${name} substitutions as kwargs.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
248
249
250
async def run(self, uid: str, **variables: str) -> SqlViewResult:
    """Execute a saved view by UID, passing `${name}` substitutions as kwargs."""
    return await self._accessor.execute(uid, variables=variables if variables else None)
run_with_criteria(uid, *, criteria, **variables) async

Execute a saved view with both variable substitutions and column criteria.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
async def run_with_criteria(
    self,
    uid: str,
    *,
    criteria: Mapping[str, str],
    **variables: str,
) -> SqlViewResult:
    """Execute a saved view with both variable substitutions and column criteria."""
    return await self._accessor.execute(
        uid,
        variables=variables if variables else None,
        criteria=criteria,
    )
adhoc(name, sql, *, view_type=SqlViewType.QUERY, keep=False, **variables) async

Register a throwaway SqlView, execute it once, then delete it.

Set keep=True to leave the view in place afterwards (handy when the next iteration wants to inspect it via the DHIS2 UI).

Subject to the DHIS2 SQL allowlist — for fully free-form SQL, use the Postgres injector example instead.

Source code in packages/dhis2w-client/src/dhis2w_client/sql_views.py 
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
async def adhoc(
    self,
    name: str,
    sql: str,
    *,
    view_type: SqlViewType | str = SqlViewType.QUERY,
    keep: bool = False,
    **variables: str,
) -> SqlViewResult:
    """Register a throwaway SqlView, execute it once, then delete it.

    Set `keep=True` to leave the view in place afterwards (handy when
    the next iteration wants to inspect it via the DHIS2 UI).

    Subject to the DHIS2 SQL allowlist — for fully free-form SQL, use
    the Postgres injector example instead.
    """
    from dhis2w_client.uids import generate_uid

    candidate = SqlView.model_validate(
        {
            "id": generate_uid(),
            "name": name,
            "type": view_type.value if isinstance(view_type, SqlViewType) else view_type,
            "sqlQuery": sql,
        },
    )
    created = await self._accessor.create(candidate)
    if created.id is None:
        raise RuntimeError("adhoc create did not return a UID — server may have rejected the view")
    try:
        if created.type in (SqlViewType.VIEW, SqlViewType.MATERIALIZED_VIEW):
            await self._accessor.refresh(created.id)
        return await self._accessor.execute(created.id, variables=variables if variables else None)
    finally:
        if not keep:
            await self._accessor.delete(created.id)