Skip to content

dhis2w-client: step-by-step guide

End-to-end tutorial for the dhis2w-client Python library. Every block is runnable; paste it into a file, set your env, and run.

If you already use the dhis2 CLI or the MCP server, this library is what those layers sit on. Use it directly when you're writing Python scripts or your own tooling.

Prerequisites

  • Python 3.13+
  • A reachable DHIS2 v42+ instance. Local: make dhis2-run; remote: your own install or https://play.im.dhis2.org/stable-2-42.
  • Credentials; PAT, username+password, or OAuth2 client config.

Install

The client is a workspace member. If you're inside this repo:

uv sync --all-packages        # installs every member

Standalone (outside the repo, once published):

uv add dhis2w-client             # core client only; profile-agnostic
uv add dhis2w-client dhis2w-core  # + profiles / plugin runtime

Concepts: auth + profiles

Two concepts to internalise before any code:

Auth provider. An AuthProvider is "how to prove who you are to DHIS2 on every request." Three shipped variants — BasicAuth, PatAuth, OAuth2Auth — each implementing the same Protocol. The rest of the client doesn't care which one you pick.

Profile. A Profile is "a named bundle of how to reach one DHIS2 instance" — a base URL plus the parameters needed to build the right AuthProvider. A profile can be:

  • Resolved from a TOML file (~/.config/dhis2/profiles.toml or ./.dhis2/profiles.toml) — the same files the dhis2 CLI manages.
  • Built from env variables (DHIS2_URL + DHIS2_PAT / DHIS2_USERNAME + DHIS2_PASSWORD / DHIS2_OAUTH_*).
  • Constructed in-memory from any Python code — no disk, no env, just a Pydantic model.

Profiles are the preferred entry point for every Python script. They're what the CLI uses, what the MCP server uses, and what every plugin service.py uses. Use them unless you have a specific reason to skip them (see the direct-client section at the end).

The split:

  • dhis2w-client — pure HTTP client + AuthProvider implementations. Profile-agnostic. Standalone PyPI package.
  • dhis2w-core — profile resolution, TOML discovery, OAuth2 token caching, open_client(profile) helper. Depends on dhis2w-client.

Every code block in this guide uses dhis2w-core.open_client(profile) as the happy path. The "direct-client" form (Dhis2Client(base_url, auth=...)) is covered at the end for the cases that genuinely need it.

Your first call

Simplest working script. open_client(profile) is an async context manager that resolves the profile's auth, opens a Dhis2Client, runs the DHIS2-version handshake, and yields the connected client.

import asyncio

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


async def main() -> None:
    async with open_client(profile_from_env()) as client:
        print(f"Connected to DHIS2 {client.raw_version}")
        me = await client.system.me()
        print(f"  logged in as: {me.username} ({me.displayName})")


asyncio.run(main())

profile_from_env() resolves through the full precedence chain: explicit DHIS2_PROFILE env first, then the default in the nearest .dhis2/profiles.toml or ~/.config/dhis2/profiles.toml, then falls back to raw DHIS2_URL + DHIS2_PAT / DHIS2_USERNAME + DHIS2_PASSWORD / DHIS2_OAUTH_* env vars.

What happens on __aenter__: 1. Resolves the Profile into a concrete AuthProvider. 2. Opens an httpx.AsyncClient connection pool. 3. Calls /api/system/info to discover the DHIS2 version. 4. Binds the matching generated resource accessors (client.resources.*).

client.version_key is "v42" afterwards; client.raw_version is "2.42.4".

Profiles — three ways to build one

A. Named profile from the TOML file

Exactly what the CLI sees — no difference in resolution path:

from dhis2w_core.client_context import open_client
from dhis2w_core.profile import resolve_profile

async with open_client(resolve_profile("staging")) as client:
    me = await client.system.me()

If you don't pass a name, resolve_profile() / profile_from_env() use the precedence chain (env → TOML default → env fallback). See profiles for the file format, scope rules (global vs project), and precedence order.

B. From env vars (no TOML file needed)

The profile_from_env() fallback kicks in when no TOML is found:

export DHIS2_URL=http://localhost:8080
export DHIS2_PAT=d2p_abc123

async with open_client(profile_from_env()) as client:
    ...

Useful in CI, Docker, any scenario where you'd rather not mount a config file.

C. In-memory, never touches disk

Profile is a plain Pydantic model — construct it like any other typed config. Nothing gets persisted anywhere.

from dhis2w_core.client_context import open_client
from dhis2w_core.profile import Profile

# PAT
profile = Profile(base_url="http://localhost:8080", auth="pat", token=os.environ["MY_PAT"])

# Basic (dev only — production should use PAT or OAuth2)
profile = Profile(
    base_url="http://localhost:8080",
    auth="basic",
    username="admin",
    password="district",
)

# OAuth2 client-credentials — for service-account style flows
profile = Profile(
    base_url="http://localhost:8080",
    auth="oauth2",
    client_id=os.environ["DHIS2_OAUTH_CLIENT_ID"],
    client_secret=os.environ["DHIS2_OAUTH_CLIENT_SECRET"],
)

async with open_client(profile) as client:
    me = await client.system.me()

Use this when:

  • A caller passes you connection details at runtime (SaaS multi-tenant, agent workflows)
  • You want to spin up a throwaway client against a per-test fixture
  • You want strict control over where the credentials come from

The same open_client + client.* API works regardless of which of the three paths you used. The rest of this guide assumes any of them.

Managing on-disk profiles from Python

Every dhis2 profile ... CLI command maps 1:1 onto a function in dhis2w_core.plugins.profile.service:

CLI Python
dhis2 profile add NAME ... service.add_profile(name, profile, *, scope, make_default)
dhis2 profile list service.list_profiles(*, include_shadowed) → list[ProfileSummary]
dhis2 profile show NAME service.show_profile(name, *, include_secrets) → ProfileView
dhis2 profile rename OLD NEW service.rename_profile(old, new)
dhis2 profile set-default NAME service.set_default_profile(name, *, scope)
dhis2 profile remove NAME service.remove_profile(name, *, scope)
dhis2 profile verify NAME await service.verify_profile(name)

Pass a start: Path argument to scope the write to a specific project directory — service.add_profile(..., scope="project", start=tmp_path) writes to <tmp_path>/.dhis2/profiles.toml instead of the user's real ~/.config/dhis2/profiles.toml. Handy for tests and isolation.

examples/client/profile_crud.py walks both paths — in-memory Profile(...) and on-disk add / rename / set-default / remove — against an isolated temp directory so it's safe to re-run.

Auth providers in detail

Profile.auth is a Literal["pat", "basic", "oauth2"] tag; open_client builds the right AuthProvider internally. When constructing a profile in-memory, fill the fields for the auth type you pick:

auth value Required fields Optional
"pat" token —
"basic" username, password —
"oauth2" client_id, client_secret scope, redirect_uri (for interactive flows)

PATs are user-scoped, long-lived, revocable. They travel as Authorization: ApiToken <token> (DHIS2-flavoured, not Bearer).

profile = Profile(base_url="http://localhost:8080", auth="pat", token=os.environ["DHIS2_PAT"])
async with open_client(profile) as client:
    ...

Mint a PAT:

  • CLI: dhis2 dev pat create --url <url> --admin-user admin --description "example" (needs DHIS2_ADMIN_PAT or DHIS2_ADMIN_PASSWORD in env)
  • Web UI: every user's profile page at /dhis-web-user-profile
  • Per the seeded e2e fixture: make dhis2-run writes one to infra/home/credentials/.env.auth

Basic (dev only)

BasicAuth sends Authorization: Basic base64(user:pw) on every request. Use against dev/play instances only.

profile = Profile(base_url="http://localhost:8080", auth="basic", username="admin", password="district")
async with open_client(profile) as client:
    ...

OAuth2 / OIDC (short-lived tokens, auto-refresh)

OAuth2 profiles need one extra step for interactive flows: the user runs dhis2 profile login <name> once, which walks the browser flow and persists an access+refresh token in .dhis2/tokens.sqlite. open_client then picks up the persisted token automatically on later runs.

profile = Profile(
    base_url="http://localhost:8080",
    auth="oauth2",
    client_id=os.environ["DHIS2_OAUTH_CLIENT_ID"],
    client_secret=os.environ["DHIS2_OAUTH_CLIENT_SECRET"],
    scope="openid",
    redirect_uri="http://localhost:8765",
)

async with open_client(profile, profile_name="my-oauth-profile") as client:
    # Access token auto-refreshes when within 30s of expiry;
    # refreshed tokens written back to the store keyed on (scope, profile_name).
    ...

For a complete standalone OAuth2 demo including PKCE, FastAPI redirect receiver, and SQLite token store, see examples/client/oidc_login.py. Architecture details in Pluggable auth.

Route API auth (for /api/routes objects, not for client auth)

DHIS2's Route API proxies requests to upstream services; its auth field is a separate discriminated union unrelated to the client-to-DHIS2 auth described above. The typed union lives in dhis2w_client.AuthScheme:

from dhis2w_client import AuthSchemeAdapter, HttpBasicAuthScheme

scheme = HttpBasicAuthScheme(type="http-basic", username="svc", password="secret")
parsed = AuthSchemeAdapter.validate_python({"type": "api-token", "token": "..."})

Typed resource CRUD

After __aenter__, client.resources exposes a typed accessor for every metadata resource DHIS2 publishes at /api/<plural>. Attribute names are snake-cased plurals: data_elements, organisation_units, category_combos, etc.

from dhis2w_client import generate_uid
from dhis2w_client.generated.v42.common import Reference
from dhis2w_client.generated.v42.enums import AggregationType, DataElementDomain, ValueType
from dhis2w_client.generated.v42.schemas.data_element import DataElement
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:
    uid = generate_uid()  # 11-char, client-side, matches dhis2w-core/CodeGenerator.java

    # CREATE
    new_de = DataElement(
        id=uid,
        code=f"EX_DE_{uid}",
        name=f"Example DE {uid}",
        shortName=f"Ex {uid[:4]}",
        domainType=DataElementDomain.AGGREGATE,
        valueType=ValueType.NUMBER,
        aggregationType=AggregationType.SUM,
        categoryCombo=Reference(id="bjDvmb4bfuf"),  # default CC from the seed
    )
    await client.resources.data_elements.create(new_de)

    # READ
    fetched = await client.resources.data_elements.get(uid, fields="id,name,valueType")
    print(f"fetched: {fetched.name} ({fetched.valueType})")

    # UPDATE; PUT the whole model back
    fetched.name = "Renamed"
    await client.resources.data_elements.update(fetched)

    # PATCH; partial update via RFC 6902 JSON Patch
    from dhis2w_client import ReplaceOp
    await client.resources.data_elements.patch(uid, [ReplaceOp(path="/shortName", value="Px")])

    # DELETE
    await client.resources.data_elements.delete(uid)

Enum fields are StrEnums; DataElementDomain.AGGREGATE == "AGGREGATE" is true, so you can pass bare strings too. Reference has both id and code fields; pick whichever your calling import strategy uses.

Filters, order, paging

async with open_client(profile_from_env()) as client:
    # Single filter
    recent = await client.resources.data_elements.list(
        filters=["name:like:Penta"],
        fields="id,name",
    )

    # Multi-filter OR
    matches = await client.resources.data_elements.list(
        filters=["name:like:Penta", "code:eq:DE_PENTA1"],
        root_junction="OR",
    )

    # Server-side paging + ordering
    page = await client.resources.organisation_units.list(
        order=["level:asc", "name:asc"],
        page_size=20,
        page=2,
    )

    # Dump everything in one request (paging=False)
    everything = await client.resources.indicators.list(paging=False, fields=":identifiable")

    # Or walk pages to get the `pager` block
    envelope = await client.resources.data_elements.list_raw(page_size=50, page=1)
    pager = envelope.get("pager", {})  # {"page", "pageSize", "total", "pageCount"}

Filter syntax is DHIS2's native property:operator:value form. Operators: eq, ieq, ne, like, !like, ilike, in, !in, null, !null, gt, ge, lt, le, token. Use Python f-strings for interpolation; f"valueType:eq:{ValueType.NUMBER}".

Bulk operations

Every write endpoint returns a WebMessageResponse envelope. It's the same shape across DHIS2 so we model it once and reuse.

from dhis2w_client import DataValue, DataValueSet, WebMessageResponse
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:
    payload = DataValueSet(
        dataValues=[
            DataValue(
                dataElement="fClA2Erf6IO",
                period="202603",
                orgUnit="PMa2VCrupOd",
                value="42",
            ),
        ],
    )
    raw = await client.post_raw("/api/dataValueSets", payload.model_dump(exclude_none=True))
    response = WebMessageResponse.model_validate(raw)

    print(f"status={response.status}")
    counts = response.import_count()
    if counts is not None:
        print(
            f"  imported={counts.imported} updated={counts.updated} "
            f"ignored={counts.ignored} deleted={counts.deleted}"
        )
    for conflict in response.conflicts():
        print(f"  conflict: {conflict.property} = {conflict.value} [{conflict.errorCode}]")

Helpers on WebMessageResponse:

  • .status, .httpStatus, .httpStatusCode, .message — envelope scalar fields
  • .created_uid — UID from an object-report envelope (handles DHIS2's response.uid vs id inconsistency, see BUGS.md #4f)
  • .import_count() → typed ImportCount (flat OR nested response.importCount forms)
  • .conflicts() → list[Conflict] — per-row rejections with property, value, errorCode
  • .rejected_indexes() → list[int] — payload-array indexes DHIS2 refused
  • .import_report() → typed ImportReport for /api/metadata bulk responses
  • .task_ref() → (job_type, task_uid) tuple when DHIS2 returned a job-kickoff envelope

Error handling

DHIS2 returns 4xx with a JSON body describing what went wrong. The client always raises for ≥400; and always captures the body, even when it's a WebMessageResponse.

from dhis2w_client import AuthenticationError, Dhis2ApiError
from dhis2w_core.client_context import open_client
from dhis2w_core.profile import Profile

bogus = Profile(base_url="http://localhost:8080", auth="pat", token="not-a-real-pat")

async with open_client(bogus) as client:
    try:
        await client.system.me()
    except AuthenticationError as exc:
        print(f"401: {exc}")
    except Dhis2ApiError as exc:
        # Any non-401 4xx/5xx
        print(f"{exc.status_code}: {exc.message}")
        envelope = exc.web_message  # typed WebMessageResponse, or None
        if envelope is not None:
            for conflict in envelope.conflicts():
                print(f"  {conflict.property}: {conflict.value}")

Exception hierarchy:

  • Dhis2ClientError — base
  • Dhis2ApiError — any non-success response; .body carries the parsed JSON / text, .web_message lazily parses it as WebMessageResponse
  • AuthenticationError — 401 specifically
  • OAuth2FlowError — state mismatch, missing code, refresh failure
  • UnsupportedVersionError — no generated client for the DHIS2 version the server reports

Analytics queries

The /api/analytics endpoint has three response shapes. Pass shape="table" (default), "raw", or "dvs" (DataValueSet).

from dhis2w_client import AnalyticsMetaData, DataValueSet, Grid
from dhis2w_core.plugins.analytics import service

response = await service.query_analytics(
    profile_from_env(),
    dimensions=["dx:fClA2Erf6IO", "pe:LAST_12_MONTHS", "ou:ImspTQPwCqd"],
)
match response:
    case Grid(rows=rows, headers=headers, metaData=meta):
        # `rows` / `headers` / `metaData` are all `| None` per the OAS spec —
        # guard with `or []` / `or {}` or narrow explicitly.
        for row in rows or []:
            print(row)
        # `metaData` is `dict[str, Any]` on the wire; lift to typed when needed.
        if meta:
            typed = AnalyticsMetaData.model_validate(meta)
            print(typed.dimensions["dx"])
    case DataValueSet(dataValues=values):
        for dv in values or []:
            print(f"{dv.dataElement} {dv.period} {dv.orgUnit} = {dv.value}")

Grid / GridHeader are the OAS-emitted canonical types. AnalyticsMetaData is a typed parser helper over Grid.metaData — use it when you want the structured {items, dimensions} view; skip it when you're iterating rows.

For streaming large exports to disk, reach for client.analytics.stream_to(path, params=..., endpoint=...) — that feeds httpx's chunked transfer straight to a file without buffering the full body.

For resource-table regeneration after a data push, the typed service wrappers avoid the raw-call boilerplate:

envelope = await service.refresh_analytics(profile, last_years=1)
ref = envelope.task_ref()
assert ref is not None
async with open_client(profile) as client:
    completion = await client.tasks.await_completion(ref, timeout=300.0)
print(f"refresh done — {len(completion.notifications)} notifications")

service.refresh_resource_tables(profile) and service.refresh_monitoring(profile) cover the two sibling endpoints (/api/resourceTables without a suffix, /api/resourceTables/monitoring) for OU-hierarchy or validation-monitoring rebuilds.

Tracker reads

Reads return typed instance models from dhis2w_client.generated.v42.tracker (version-scoped — tracker shapes drift across DHIS2 majors). Writes go through the typed TrackerBundle from the same module.

from dhis2w_client.generated.v42.tracker import TrackerTrackedEntity

async with open_client(profile_from_env()) as client:
    raw = await client.get_raw(
        "/api/tracker/trackedEntities",
        params={"program": "<PROG_UID>", "pageSize": 10},
    )
    for entity in raw.get("instances", []):
        te = TrackerTrackedEntity.model_validate(entity)
        print(f"{te.trackedEntity}  type={te.trackedEntityType}  orgUnit={te.orgUnit}")

EventStatus and EnrollmentStatus are StrEnums; EventStatus.COMPLETED == "COMPLETED".

Task polling

Every async DHIS2 op (analytics refresh, metadata import, data-integrity run, tracker async push) returns a JobConfigurationWebMessageResponse carrying jobType + task UID. Use .task_ref() to pull the polling tuple, then client.tasks.await_completion(...) to block until the job finishes:

from dhis2w_client import WebMessageResponse
from dhis2w_client.tasks import TaskTimeoutError

async with open_client(profile_from_env()) as client:
    raw = await client.post_raw("/api/resourceTables/analytics", params={"lastYears": 1})
    envelope = WebMessageResponse.model_validate(raw)
    ref = envelope.task_ref()
    if ref is None:
        raise RuntimeError("response had no jobType/id; nothing to watch")

    try:
        completion = await client.tasks.await_completion(
            ref,
            timeout=300.0,        # seconds; pass None to wait forever
            poll_interval=1.0,    # seconds between polls
        )
    except TaskTimeoutError as exc:
        print(f"task didn't finish in time: {exc}")
        return

    print(f"{completion.level}  {completion.message}")
    # completion.notifications is the full chronological list
    # completion.final is the terminal row (completed=True)

await_completion handles the polling loop, de-duplicates notifications across polls (so the same progress message isn't yielded twice), and reuses the client's open HTTP connection (no new TCP handshake per poll). Pass a (job_type, uid) tuple or a "JOB_TYPE/uid" string interchangeably.

For custom rendering (Rich progress bars, server-sent-event bridges), iterate the raw stream instead:

async for notification in client.tasks.iter_notifications(ref, poll_interval=1.0):
    level = (notification.level or "INFO").upper()
    marker = "[x]" if notification.completed else "[ ]"
    print(f"  {level:<5} {marker} {notification.message}")

See examples/client/task_await.py for a runnable demo. The CLI --watch flag (dhis2 maintenance refresh analytics --watch, dhis2 maintenance dataintegrity run --watch) uses a Rich-progress wrapper on top of the same primitive.

Streaming data-integrity issues

client.maintenance.iter_integrity_issues(...) gives a flat stream over the {check_name: {issues: [...]}} map DHIS2 returns from /api/dataIntegrity/details. Each yielded IntegrityIssueRow carries the owning check's name, display name, and severity:

async with open_client(profile_from_env()) as client:
    async for row in client.maintenance.iter_integrity_issues():
        if (row.severity or "").upper() == "WARNING":
            print(f"{row.check_name:40}  {row.issue.id}  {row.issue.name}")

Use client.maintenance.get_integrity_report(details=True) (or details=False for the cheaper summary endpoint) when you want the full typed report instead. See examples/client/integrity_issues_stream.py for a runnable demo.

System cache

Every client ships a TTL-bounded in-memory cache for the high-repetition system-level reads. client.system.info() is primed on connect() (no second round-trip); client.system.default_category_combo_uid() + client.system.setting(key) cache per call-site. Default TTL is 300 s:

async with open_client(profile_from_env()) as client:
    info = await client.system.info()                       # primed; free
    default_cc = await client.system.default_category_combo_uid()  # fetched once; cached
    title = await client.system.setting("applicationTitle")        # per-key cache

    client.system.invalidate_cache()                       # drop everything
    # or: client.system.invalidate_cache(key="setting:applicationTitle")

Tune via open_client(profile, system_cache_ttl=600.0) or pass None to disable. See examples/client/system_cache.py for a timed demo.

UID generation

DHIS2 UIDs are 11-char strings matching ^[A-Za-z][A-Za-z0-9]{10}$. Instead of /api/system/id round-trips, generate them client-side; same algorithm as dhis2w-core/CodeGenerator.java:

from dhis2w_client import UID_RE, generate_uid, generate_uids, is_valid_uid

generate_uid()              # "aB3dEf5gH7i"
generate_uids(100)          # list of 100 unique UIDs
is_valid_uid("fClA2Erf6IO") # True
UID_RE.pattern              # '^[A-Za-z][A-Za-z0-9]{10}$'

Uses secrets.choice (CSPRNG), matches the SecureRandom path upstream. No client connection needed.

Versions + fallback

On connect, the client pulls /api/system/info, extracts the minor version, and binds the matching generated module. Versions shipped: v40, v41, v42, v44. If the reported version has no generated module, construction fails with UnsupportedVersionError unless you opt into fallback.

Pass the knob through open_client:

async with open_client(profile_from_env(), allow_version_fallback=True) as client:
    # Falls back to the highest generated version <= the server's reported version.
    ...

To pin a specific version regardless of what the server reports, you need the direct-client path — see below. Regenerate codegen for a new version with dhis2 dev codegen rebuild or point at a live instance with dhis2 dev codegen generate --url <url>.

Retry on transient failures

Batch workflows hitting a live DHIS2 instance sometimes see transient 5xxs (503 during an analytics refresh) or connection resets (TCP keepalive drops on long idle periods). Opt in to retries via RetryPolicy:

from dhis2w_client import RetryPolicy

policy = RetryPolicy(
    max_attempts=5,
    base_delay=0.2,
    backoff_factor=2.0,
    max_delay=5.0,
    jitter=0.15,
    retry_statuses=frozenset({429, 502, 503, 504}),  # default set
)

# Profile-based (preferred):
async with open_client(profile_from_env(), retry_policy=policy) as client:
    ...

# Direct-client path (no dhis2w-core):
async with Dhis2Client(base_url, auth=auth, retry_policy=policy) as client:
    ...

Retry scope:

  • Connection-level errors (httpx.ConnectError, ReadTimeout, PoolTimeout, RemoteProtocolError) always retry.
  • Status codes listed in policy.retry_statuses (default: 429, 502, 503, 504) retry.
  • Non-idempotent methods (POST, PATCH) are exempt by default — double-writes risk DHIS2-side duplicates. Opt in for specific endpoints where you know it's safe (analytics-refresh kick-offs, for instance):
RetryPolicy(retry_non_idempotent=True)
  • Retry-After header from the server (sent on 429 / 503) overrides the computed backoff for that attempt.

Backoff: delay = min(max_delay, base_delay * backoff_factor ** (attempt - 1)) with a fractional jitter applied before sleeping.

See examples/client/retry_policy.py for a runnable demo across default / aggressive / non-idempotent-opt-in policies.

Concurrency

The client's httpx.AsyncClient is shared across concurrent calls on the same Dhis2Client instance; safe to use with asyncio.gather.

import asyncio

async with open_client(profile_from_env()) as client:
    # Parallel GETs
    uids = ["fClA2Erf6IO", "UOlfIjgN8X6", "I78gJm4KBo7"]
    elements = await asyncio.gather(
        *(client.resources.data_elements.get(uid) for uid in uids),
    )

    # Parallel writes — DHIS2 is generally fine with moderate concurrency
    # on independent records. Tune to your instance's capacity.
    await asyncio.gather(*(client.resources.data_elements.update(de) for de in elements))

Connection pool defaults are httpx's defaults (100 max connections, 20 keepalive).

Raw escape hatches

Every endpoint DHIS2 has ever built is reachable through five raw methods. Prefer the typed accessors when they exist; these never lose.

async with open_client(profile_from_env()) as client:
    raw = await client.get_raw("/api/some/unusual/path", params={"key": "val"})
    raw = await client.post_raw("/api/metadata", {"dataElements": [...]})
    raw = await client.put_raw("/api/dataElements/abc", {"name": "..."})
    raw = await client.delete_raw("/api/dataElements/abc")
    raw = await client.patch_raw("/api/dataElements/abc", [{"op": "replace", "path": "/name", "value": "..."}])

    # Typed GET against your own model:
    from pydantic import BaseModel

    class MyModel(BaseModel):
        id: str
        name: str

    item = await client.get("/api/something", model=MyModel)

get_raw always returns dict[str, Any]. When DHIS2 returns a bare JSON array (e.g. /api/system/id), it's wrapped under {"data": [...]} so the return type stays consistent.

When to skip profiles (direct-client path)

dhis2w-core depends on dhis2w-client, not the other way around. If you're writing library code that should live without the dhis2w-core dependency — a downstream SDK, a minimal Lambda handler, a test fixture — you can drive Dhis2Client directly:

from dhis2w_client import BasicAuth, Dhis2Client, PatAuth

# PAT
async with Dhis2Client(
    base_url="http://localhost:8080",
    auth=PatAuth(token=os.environ["DHIS2_PAT"]),
) as client:
    me = await client.system.me()

# Basic (dev only)
async with Dhis2Client(
    base_url="http://localhost:8080",
    auth=BasicAuth(username="admin", password="district"),
) as client:
    ...

OAuth2 at the direct-client layer skips the profile's token-store key plumbing — see the OAuth2Auth direct-construction block further down.

When to use which path:

You're writing Use
A script / CLI tool / internal app (picks up CLI profiles) open_client(profile_from_env())
A service that takes connection details at runtime open_client(Profile(...)) (in-memory)
A plugin / service-layer function inside this workspace open_client(profile) (plugin layer already resolved it)
A library that imports dhis2w-client from PyPI without dhis2w-core Dhis2Client(base_url, auth=...) directly
Pinning a specific DHIS2 version regardless of server-reported version Dhis2Client(..., version=Dhis2.V42) directly

Direct-client OAuth2 (reference only)

For the rare case where you need OAuth2Auth without going through a profile — e.g. you have your own TokenStore implementation and want to wire token persistence yourself:

from dhis2w_client import Dhis2Client
from dhis2w_client.auth.oauth2 import OAuth2Auth
from dhis2w_core.token_store import token_store_for_scope

store = token_store_for_scope("global")
token = await store.get("my-oauth-profile")

auth = OAuth2Auth(
    token=token,
    token_store=store,
    token_key="my-oauth-profile",
    token_url="http://localhost:8080/oauth2/token",
    client_id=os.environ["DHIS2_OAUTH_CLIENT_ID"],
    client_secret=os.environ["DHIS2_OAUTH_CLIENT_SECRET"],
)

async with Dhis2Client(base_url="http://localhost:8080", auth=auth) as client:
    ...

Under open_client(profile), all the above wiring happens automatically from the Profile fields — there's a reason it's the default.

Where next?