Client + lifecycle

The async Dhis2Client is the entry point for every DHIS2 call. It owns the httpx connection pool, performs the version handshake on connect, binds the matching generated accessors, and exposes raw HTTP escape hatches for endpoints that don't have a typed wrapper yet.

When to reach for it

• Standalone PyPI use (no profile system) — instantiate Dhis2Client(base_url, auth=...) directly.
• Inside dhis2w-core integrations — the canonical path is dhis2w_core.client_context.open_client(profile), which returns the same Dhis2Client but wraps profile resolution + cleanup.
• Subclassing or wrapping for a custom workflow (the class is a BaseModel-free regular async class, not frozen).

Worked example — standalone library use

import asyncio
from dhis2w_client import BasicAuth, Dhis2Client, RetryPolicy

import httpx


async def main() -> None:
    """Connect to DHIS2 via plain Basic auth + a retry policy + bigger pool."""
    async with Dhis2Client(
        base_url="https://play.im.dhis2.org/dev-2-43",
        auth=BasicAuth("admin", "district"),
        retry_policy=RetryPolicy(),                              # 429/5xx + connection errors
        http_limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
    ) as client:
        print(f"version_key  = {client.version_key}")            # 'v42' / 'v43' / 'v41'
        print(f"raw_version  = {client.raw_version}")            # '2.43.0' etc.
        me = await client.system.me()
        print(f"as {me.username} ({me.displayName})")


asyncio.run(main())

Worked example — workspace-integrated use (dhis2w-core)

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:
    me = await client.system.me()

The two paths return the same Dhis2Client; open_client just resolves the profile + chooses the matching AuthProvider for you. Every example under examples/v42/client/ uses this form because the workspace has profiles configured.

Raw HTTP escape hatches

When an endpoint doesn't have a typed wrapper yet (or you want the literal wire shape for debugging), drop down to:

• await client.get_raw(path, params=...) -> dict[str, Any]
• await client.post_raw(path, body=..., params=...) -> dict[str, Any]
• await client.put_raw(path, body=...) -> dict[str, Any]
• await client.patch_raw(path, body=...) -> dict[str, Any]
• await client.delete_raw(path) -> dict[str, Any]
• await client.get(path, model=MyBaseModel) -> typed via your own pydantic model
• await client.get_response(path, params=..., extra_headers=...) -> httpx.Response (no raise on 4xx/5xx)

Keep the use of raw helpers narrow — every typed accessor client.X.Y() is preferable for production code (the typed return is what makes the rest of the codebase work). The architecture page Client library covers the lifecycle states (unconnected -> connecting -> connected -> closed) and how connect() binds the version-specific accessors.

get_response is the only helper that bypasses the 4xx/5xx raise in _request. Use it when a non-2xx is a fact you want to report rather than an exception — health-checkers, SSO / proxy-page detection via Content-Type, or reverse-proxied routes under /api/routes/<code>/run/... where a 502 means "DHIS2 reached, downstream didn't".

Constructor knobs for non-default lifecycles

The defaults (verify=True, full version probe on connect(), raise on 4xx/5xx) suit typed-accessor flows. Three opt-in kwargs unlock other shapes:

• verify: bool | str = True — TLS certificate verification. Threaded through the main httpx pool plus the canonical-URL and DHIS2-shape probes. Pass False for self-signed staging boxes or a path to a custom CA bundle.
• skip_version_probe: bool = False — when True, connect() opens the HTTP pool without the canonical-URL probe or the /api/system/info round-trip and returns. version_key, raw_version, and resources raise on access (the generated tree never binds), so only the raw-path methods (get_raw, post_raw, get_response, etc.) are usable. Suits health-checkers that want to report on those endpoints, very-low-privilege PATs that can't read /api/system/info, and tests injecting a mock transport.

See examples/v42/client/health_check.py for the full health-checker pattern.

client

Async DHIS2 client with pluggable auth and version-aware generated dispatch.

Classes

Dhis2Client

Async DHIS2 client for v42 (the canonical baseline); version is discovered via /api/system/info on connect.

This class's accessor attributes (self.metadata, self.apps, etc.) are typed against the v42 hand-written tree (dhis2w_client.v42.*). At runtime, connect() calls dhis2w_client._dispatch.rebind_accessors_for_version which swaps the instances for v41 / v43 versions when the server differs — so behaviour is correct, but the static types stay v42.

For static-type purity against a non-v42 server, import the per-version class explicitly:

from dhis2w_client.v43 import Dhis2Client  # v43-typed accessors
from dhis2w_client.v41 import Dhis2Client  # v41-typed accessors

The top-level from dhis2w_client import Dhis2Client re-exports this v42 class for backwards compatibility — the runtime dispatch keeps it correct against v41 / v43 stacks; only the static type chain is v42-flavoured.

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

class Dhis2Client:
    """Async DHIS2 client for v42 (the canonical baseline); version is discovered via /api/system/info on connect.

    This class's accessor attributes (`self.metadata`, `self.apps`, etc.)
    are typed against the v42 hand-written tree (`dhis2w_client.v42.*`).
    At runtime, `connect()` calls `dhis2w_client._dispatch.rebind_accessors_for_version`
    which swaps the *instances* for v41 / v43 versions when the server
    differs — so behaviour is correct, but the static *types* stay v42.

    For static-type purity against a non-v42 server, import the
    per-version class explicitly:

    ```python
    from dhis2w_client.v43 import Dhis2Client  # v43-typed accessors
    from dhis2w_client.v41 import Dhis2Client  # v41-typed accessors
    ```

    The top-level `from dhis2w_client import Dhis2Client` re-exports this
    v42 class for backwards compatibility — the runtime dispatch keeps it
    correct against v41 / v43 stacks; only the static type chain is
    v42-flavoured.
    """

    def __init__(
        self,
        base_url: str,
        auth: AuthProvider,
        *,
        timeout: float = 30.0,
        connect_timeout: float = 60.0,
        allow_version_fallback: bool = False,
        version: Dhis2 | None = None,
        allow_version_mismatch: bool = False,
        retry_policy: RetryPolicy | None = None,
        http_limits: httpx.Limits | None = None,
        system_cache_ttl: float | None = 300.0,
        verify: bool | str = True,
        skip_version_probe: bool = False,
    ) -> None:
        """Build a client. Call connect() or use as an async context manager before API calls.

        `version` defaults to `Dhis2.V42` — the workspace targets v42 and v43.
        Set `Dhis2.V43` to pin to v43 explicitly, or pass `None` to let the
        client auto-detect via `/api/system/info` on `connect()`. Pinning
        skips that round-trip and fails fast on a server version with no
        matching generated module.

        `retry_policy` (default: no retries) enables exponential-backoff
        retries on transient failures — connection errors plus the status
        codes listed on `RetryPolicy.retry_statuses` (default 429 / 502 /
        503 / 504). Non-idempotent methods (POST, PATCH) are exempt unless
        the policy sets `retry_non_idempotent=True`. See
        `dhis2w_client.retry.RetryPolicy` for tuning knobs.

        `http_limits` overrides the httpx connection-pool defaults (100
        max connections, 20 keepalive). Raise them for high-concurrency
        batch workflows; lower them to protect a small DHIS2 instance
        from a large `asyncio.gather`. See
        `docs/architecture/client.md` for guidance.

        `system_cache_ttl` (default 300 s) caps how long cached system-level
        reads (`client.system.info()`, the default categoryCombo UID, and
        per-key system settings) stay fresh before the next call refetches.
        Pass `None` to disable the cache entirely. `connect()` primes the
        cache from the info fetch it already performs, so the first
        `client.system.info()` after connect costs zero round-trips.

        `verify` controls TLS certificate verification on every internal
        `httpx.AsyncClient` (the main pool plus the canonical-URL and
        DHIS2-shape probes). Pass `False` to disable verification (only
        safe against self-signed staging boxes) or a path to a custom CA
        bundle. Default `True`.

        `skip_version_probe` (default `False`) opens the HTTP pool without
        the two probes `connect()` normally runs (canonical-URL resolution
        plus `/api/system/info` for version detection). Useful for
        health-checkers that want to *report* on those endpoints rather
        than have them raise, very-low-privilege PATs that can't read
        `/api/system/info`, and tests that inject a mock transport. With
        this flag set, `version_key`, `raw_version`, and `resources` raise
        on access — only `get_raw`, `post_raw`, `get_response`, and
        friends are usable.
        """
        self._base_url = base_url.rstrip("/")
        self._auth = auth
        self._timeout = httpx.Timeout(timeout, connect=connect_timeout)
        self._retry_policy = retry_policy
        self._http_limits = http_limits
        self._http: httpx.AsyncClient | None = None
        self._verify = verify
        self._skip_version_probe = skip_version_probe
        self._version_key: str | None = None
        self._raw_version: str | None = None
        self._generated: ModuleType | None = None
        self._allow_fallback = allow_version_fallback
        self._version = version
        self._allow_version_mismatch = allow_version_mismatch
        self._resources: Any = None
        self._system_cache: SystemCache | None = (
            SystemCache(ttl=system_cache_ttl) if system_cache_ttl is not None else None
        )
        self.system: SystemModule = SystemModule(self)
        self.customize: CustomizeAccessor = CustomizeAccessor(self)
        self.tasks: TaskModule = TaskModule(self)
        self.maintenance: MaintenanceAccessor = MaintenanceAccessor(self)
        self.messaging: MessagingAccessor = MessagingAccessor(self)
        self.metadata: MetadataAccessor = MetadataAccessor(self)
        self.routes: RoutesAccessor = RoutesAccessor(self)
        self.datastore: DatastoreAccessor = DatastoreAccessor(self)
        self.maps: MapsAccessor = MapsAccessor(self)
        self.files: FilesAccessor = FilesAccessor(self)
        self.legend_sets: LegendSetsAccessor = LegendSetsAccessor(self)
        self.validation: ValidationAccessor = ValidationAccessor(self)
        self.attribute_values: AttributeValuesAccessor = AttributeValuesAccessor(self)
        self.option_sets: OptionSetsAccessor = OptionSetsAccessor(self)
        self.organisation_units: OrganisationUnitsAccessor = OrganisationUnitsAccessor(self)
        self.organisation_unit_groups: OrganisationUnitGroupsAccessor = OrganisationUnitGroupsAccessor(self)
        self.organisation_unit_group_sets: OrganisationUnitGroupSetsAccessor = OrganisationUnitGroupSetsAccessor(self)
        self.organisation_unit_levels: OrganisationUnitLevelsAccessor = OrganisationUnitLevelsAccessor(self)
        self.predictors: PredictorsAccessor = PredictorsAccessor(self)
        self.program_rules: ProgramRulesAccessor = ProgramRulesAccessor(self)
        self.sql_views: SqlViewsAccessor = SqlViewsAccessor(self)
        self.visualizations: VisualizationsAccessor = VisualizationsAccessor(self)
        self.dashboards: DashboardsAccessor = DashboardsAccessor(self)
        self.data_values: DataValuesAccessor = DataValuesAccessor(self)
        self.analytics: AnalyticsAccessor = AnalyticsAccessor(self)
        self.tracker: TrackerAccessor = TrackerAccessor(self)
        self.apps: AppsAccessor = AppsAccessor(self)
        self.data_elements: DataElementsAccessor = DataElementsAccessor(self)
        self.data_element_groups: DataElementGroupsAccessor = DataElementGroupsAccessor(self)
        self.data_element_group_sets: DataElementGroupSetsAccessor = DataElementGroupSetsAccessor(self)
        self.indicators: IndicatorsAccessor = IndicatorsAccessor(self)
        self.indicator_groups: IndicatorGroupsAccessor = IndicatorGroupsAccessor(self)
        self.indicator_group_sets: IndicatorGroupSetsAccessor = IndicatorGroupSetsAccessor(self)
        self.program_indicators: ProgramIndicatorsAccessor = ProgramIndicatorsAccessor(self)
        self.program_indicator_groups: ProgramIndicatorGroupsAccessor = ProgramIndicatorGroupsAccessor(self)
        self.category_options: CategoryOptionsAccessor = CategoryOptionsAccessor(self)
        self.category_option_groups: CategoryOptionGroupsAccessor = CategoryOptionGroupsAccessor(self)
        self.category_option_group_sets: CategoryOptionGroupSetsAccessor = CategoryOptionGroupSetsAccessor(self)
        self.categories: CategoriesAccessor = CategoriesAccessor(self)
        self.category_combos: CategoryCombosAccessor = CategoryCombosAccessor(self)
        self.category_option_combos: CategoryOptionCombosAccessor = CategoryOptionCombosAccessor(self)
        self.data_sets: DataSetsAccessor = DataSetsAccessor(self)
        self.sections: SectionsAccessor = SectionsAccessor(self)
        self.validation_rules: ValidationRulesAccessor = ValidationRulesAccessor(self)
        self.validation_rule_groups: ValidationRuleGroupsAccessor = ValidationRuleGroupsAccessor(self)
        self.predictor_groups: PredictorGroupsAccessor = PredictorGroupsAccessor(self)
        self.tracked_entity_attributes: TrackedEntityAttributesAccessor = TrackedEntityAttributesAccessor(self)
        self.tracked_entity_types: TrackedEntityTypesAccessor = TrackedEntityTypesAccessor(self)
        self.programs: ProgramsAccessor = ProgramsAccessor(self)
        self.program_stages: ProgramStagesAccessor = ProgramStagesAccessor(self)

    @property
    def base_url(self) -> str:
        """Root URL of the connected DHIS2 instance."""
        return self._base_url

    @property
    def version_key(self) -> str:
        """Return the generated-client version key (e.g. "v42"); requires connect()."""
        if self._version_key is None:
            raise RuntimeError("Dhis2Client is not connected; call connect() first")
        return self._version_key

    @property
    def raw_version(self) -> str:
        """Return the raw version string reported by the server (e.g. "2.42.0")."""
        if self._raw_version is None:
            raise RuntimeError("Dhis2Client is not connected; call connect() first")
        return self._raw_version

    @property
    def resources(self) -> Any:
        """Return the version-bound generated `Resources` accessor; requires connect()."""
        if self._resources is None:
            raise RuntimeError("Dhis2Client is not connected; call connect() first")
        return self._resources

    @property
    def system_cache(self) -> SystemCache | None:
        """Per-client TTL cache for system-level reads; `None` when caching is disabled."""
        return self._system_cache

    async def __aenter__(self) -> Self:
        """Open the HTTP pool and run version discovery."""
        await self.connect()
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        """Close the HTTP pool."""
        await self.close()

    async def connect(self) -> None:
        """Open the HTTP pool and bind the generated module matching the remote version.

        With `skip_version_probe=True`, skips canonical-URL resolution and
        the `/api/system/info` round-trip — leaves `version_key`,
        `raw_version`, and `resources` unset (they raise on access) and
        returns once the pool is open.
        """
        if not self._skip_version_probe:
            resolved = await self._resolve_canonical_base_url(self._base_url, verify=self._verify)
            if resolved != self._base_url:
                self._base_url = resolved
        if self._http is None:
            kwargs: dict[str, Any] = {
                "base_url": self._base_url,
                "timeout": self._timeout,
                "verify": self._verify,
            }
            if self._retry_policy is not None:
                kwargs["transport"] = build_retry_transport(self._retry_policy)
            if self._http_limits is not None:
                kwargs["limits"] = self._http_limits
            self._http = httpx.AsyncClient(**kwargs)
        if self._skip_version_probe:
            return
        info = await self.get_raw("/api/system/info")
        self._raw_version = str(info.get("version", ""))
        if self._version is not None:
            # Caller asserted a version — verify the server's reported version
            # matches the pinned major. A v43 pin against a v42 server would
            # silently round-trip renamed/added fields wrong, so fail loud
            # unless the caller explicitly opts into the mismatch.
            if not self._allow_version_mismatch:
                self._assert_reported_version_matches_pin(self._raw_version, self._version)
            self._version_key = self._version.value
        else:
            self._version_key = self._pick_version_key(self._raw_version)
        self._generated = load(self._version_key)
        rebind_accessors_for_version(self, self._version_key)
        # Prime the system cache with the info we already fetched so
        # `client.system.info()` right after connect is a free in-process read.
        if self._system_cache is not None:
            self._system_cache.set("info", _SystemInfo.model_validate(info))
        resources_cls = getattr(self._generated, "Resources", None)
        if resources_cls is not None:
            self._resources = resources_cls(self)

    @staticmethod
    async def _resolve_canonical_base_url(base_url: str, *, verify: bool | str = True) -> str:
        """Follow redirects (without auth) to find the canonical DHIS2 base URL.

        httpx strips Authorization headers on cross-host redirects as a security
        measure (it won't leak credentials to a host the user didn't target).
        DHIS2 `play.*` instances redirect `play.dhis2.org/dev` ->
        `play.im.dhis2.org/dev`, so every authenticated call would silently
        drop the Authorization header and get a 401.

        Resolve the chain once, unauthenticated, so subsequent requests go
        directly to the resolved host with credentials preserved.

        Cross-origin redirects are validated against `/api/system/info` on the
        candidate host before being adopted — DHIS2 always returns JSON for
        that endpoint regardless of auth state (200 with `version` field when
        anon-accessible, 401 with a JSON error envelope otherwise), while
        an SSO IdP or unrelated host returns HTML or a 404. Without this
        probe, an SSO-protected deployment whose root redirects to an IdP
        login page would leave subsequent `/api/*` calls pointed at the IdP.
        """
        candidate = base_url.rstrip("/")
        try:
            async with httpx.AsyncClient(
                timeout=httpx.Timeout(10.0, connect=10.0),
                follow_redirects=True,
                verify=verify,
            ) as probe:
                response = await probe.get(f"{candidate}/")
        except Exception:  # noqa: BLE001 — probe is best-effort; fall back to original URL
            return candidate
        final = str(response.url).rstrip("/")
        # Strip common DHIS2 login-page trailing paths so we land on the root.
        for suffix in ("/dhis-web-login", "/login", "/dhis-web-commons/security/login.action"):
            if final.endswith(suffix):
                final = final[: -len(suffix)].rstrip("/")
                break
        if Dhis2Client._same_origin(candidate, final):
            return final
        if await Dhis2Client._probe_looks_like_dhis2(final, verify=verify):
            return final
        return candidate

    @staticmethod
    def _same_origin(left: str, right: str) -> bool:
        """Return True if two URLs share scheme + host + port."""
        a = httpx.URL(left)
        b = httpx.URL(right)
        return (a.scheme, a.host, a.port) == (b.scheme, b.host, b.port)

    @staticmethod
    async def _probe_looks_like_dhis2(base_url: str, *, verify: bool | str = True) -> bool:
        """Return True if `<base_url>/api/system/info` responds DHIS2-shaped.

        DHIS2 returns JSON on `/api/system/info` regardless of auth state.
        An SSO IdP or other host returns HTML / a 404, so the content-type
        check rejects them. Used to validate cross-origin redirect targets
        before adopting them as the canonical base URL.
        """
        try:
            async with httpx.AsyncClient(
                timeout=httpx.Timeout(5.0, connect=5.0),
                follow_redirects=False,
                verify=verify,
            ) as probe:
                response = await probe.get(
                    f"{base_url}/api/system/info",
                    headers={"Accept": "application/json"},
                )
        except Exception:  # noqa: BLE001 — probe is best-effort
            return False
        content_type = response.headers.get("content-type", "").lower()
        return "json" in content_type

    async def close(self) -> None:
        """Close the underlying HTTP pool."""
        if self._http is not None:
            await self._http.aclose()
            self._http = None

    @staticmethod
    def _assert_reported_version_matches_pin(reported: str, pinned: Dhis2) -> None:
        """Raise `VersionPinMismatchError` if the reported server version's major doesn't match the pin.

        Skipped when the server didn't report a parseable version string — in
        that case the caller's explicit pin is the only signal available, so
        we trust it rather than blocking on an unhelpful 'server didn't report
        version' error.
        """
        match = _VERSION_RE.match(reported)
        if not match:
            return
        reported_key = f"v{int(match.group(2))}"
        if reported_key != pinned.value:
            raise VersionPinMismatchError(pinned=pinned.value, reported=reported)

    def _pick_version_key(self, version_str: str) -> str:
        """Select a generated module version key for the reported DHIS2 version."""
        match = _VERSION_RE.match(version_str)
        available = list(available_versions())
        if not match:
            raise UnsupportedVersionError(version=version_str or "unknown", available=available)
        minor = int(match.group(2))
        requested = f"v{minor}"
        if requested in available:
            return requested
        if not self._allow_fallback:
            raise UnsupportedVersionError(version=version_str, available=available)
        lower = [k for k in available if int(k[1:]) <= minor]
        if not lower:
            raise UnsupportedVersionError(version=version_str, available=available)
        return max(lower, key=lambda k: int(k[1:]))

    async def get_raw(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Raw GET returning parsed JSON; used internally and as an escape hatch."""
        response = await self._request("GET", path, params=params)
        return self._parse_json(response)

    async def get_response(
        self,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        extra_headers: dict[str, str] | None = None,
    ) -> httpx.Response:
        """No-raise GET returning the raw `httpx.Response`; escape hatch for caller-side status logic.

        Skips the 4xx/5xx raise block in `_request` so callers can inspect
        status / headers / non-JSON bodies themselves. The auth header is
        still applied. Use when:

        - A 4xx / 5xx is a *fact you want to report* (health-checkers).
        - You need the raw `Content-Type` (SSO / proxy-page detection).
        - You're calling a reverse-proxied route under
          `/api/routes/<code>/run/...` where a 502 means "DHIS2 reached,
          downstream didn't" rather than "API error".
        """
        return await self._request(
            "GET",
            path,
            params=params,
            extra_headers=extra_headers,
            raise_for_status=False,
        )

    async def get[T: BaseModel](
        self,
        path: str,
        model: type[T],
        params: dict[str, Any] | None = None,
    ) -> T:
        """Typed GET returning an instance of `model` parsed from JSON."""
        raw = await self.get_raw(path, params=params)
        return model.model_validate(raw)

    async def post[T: BaseModel](
        self,
        path: str,
        body: Any,
        *,
        model: type[T],
        params: dict[str, Any] | None = None,
    ) -> T:
        """Typed POST returning an instance of `model` parsed from JSON.

        Used most often with `model=WebMessageResponse` to parse
        `/api/metadata` envelopes into the typed summary shape without
        a trailing `WebMessageResponse.model_validate(raw)` at the
        call site.
        """
        raw = await self.post_raw(path, body, params=params)
        return model.model_validate(raw)

    async def put[T: BaseModel](
        self,
        path: str,
        body: Any,
        *,
        model: type[T],
        params: dict[str, Any] | None = None,
    ) -> T:
        """Typed PUT returning an instance of `model` parsed from JSON."""
        raw = await self.put_raw(path, body, params=params)
        return model.model_validate(raw)

    async def patch[T: BaseModel](
        self,
        path: str,
        body: Any,
        *,
        model: type[T],
        params: dict[str, Any] | None = None,
        content_type: str = "application/json-patch+json",
    ) -> T:
        """Typed PATCH returning an instance of `model` parsed from JSON.

        Used with `model=WebMessageResponse` for metadata patches so the
        call site doesn't need a trailing `WebMessageResponse.model_validate(raw)`.
        """
        raw = await self.patch_raw(path, body, params=params, content_type=content_type)
        return model.model_validate(raw)

    async def delete[T: BaseModel](
        self,
        path: str,
        *,
        model: type[T],
        params: dict[str, Any] | None = None,
    ) -> T:
        """Typed DELETE returning an instance of `model` parsed from JSON.

        Most callers use `model=WebMessageResponse` since DHIS2 deletes
        return the standard envelope.
        """
        raw = await self.delete_raw(path, params=params)
        return model.model_validate(raw)

    async def post_raw(self, path: str, body: Any = None, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Raw POST returning parsed JSON."""
        response = await self._request("POST", path, params=params, json=body)
        return self._parse_json(response)

    async def put_raw(self, path: str, body: Any = None, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Raw PUT returning parsed JSON."""
        response = await self._request("PUT", path, params=params, json=body)
        return self._parse_json(response)

    async def delete_raw(self, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
        """Raw DELETE returning parsed JSON (or empty dict)."""
        response = await self._request("DELETE", path, params=params)
        return self._parse_json(response)

    async def patch_raw(
        self,
        path: str,
        body: Any,
        *,
        params: dict[str, Any] | None = None,
        content_type: str = "application/json-patch+json",
    ) -> dict[str, Any]:
        """Raw PATCH returning parsed JSON. Defaults to JSON Patch (RFC 6902) content type."""
        response = await self._request(
            "PATCH",
            path,
            params=params,
            json=body,
            extra_headers={"Content-Type": content_type},
        )
        return self._parse_json(response)

    async def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        json: Any = None,
        content: httpx._types.RequestContent | None = None,
        files: dict[str, tuple[str, bytes, str]] | None = None,
        extra_headers: dict[str, str] | None = None,
        raise_for_status: bool = True,
    ) -> httpx.Response:
        """Dispatch a request through the shared pool with fresh auth headers.

        With `raise_for_status=False`, returns the raw response on any
        status code. Used by `get_response()` so callers can do their own
        content-negotiation / status-based handling — health-checkers in
        particular want a 502 from a reverse-proxied route to surface as
        a fact, not an exception.
        """
        if self._http is None:
            raise RuntimeError("Dhis2Client is not connected; call connect() first")
        headers = await self._auth.headers()
        if extra_headers:
            headers = {**headers, **extra_headers}
        t0 = time.monotonic()
        response = await self._http.request(
            method,
            path,
            params=params,
            json=json,
            content=content,
            files=files,
            headers=headers,
        )
        if _HTTP_LOG.isEnabledFor(logging.DEBUG):
            elapsed_ms = (time.monotonic() - t0) * 1000
            _HTTP_LOG.debug(
                "%s %s -> %d (%d bytes, %.0fms)",
                method,
                str(response.request.url),
                response.status_code,
                len(response.content),
                elapsed_ms,
            )
        if raise_for_status:
            if response.status_code == 401:
                raise AuthenticationError(
                    format_unauthorized_message(method, path, response.headers.get("WWW-Authenticate"))
                )
            if response.status_code >= 400:
                body: Any
                try:
                    body = response.json()
                except ValueError:
                    body = response.text
                raise Dhis2ApiError(status_code=response.status_code, message=response.reason_phrase, body=body)
        return response

    @staticmethod
    def _parse_json(response: httpx.Response) -> dict[str, Any]:
        """Parse a successful response body into a dict (wrapping non-dict JSON under "data")."""
        try:
            parsed = response.json()
        except ValueError:
            return {}
        if isinstance(parsed, dict):
            return parsed
        return {"data": parsed}

Attributes

base_url property

Root URL of the connected DHIS2 instance.

version_key property

Return the generated-client version key (e.g. "v42"); requires connect().

raw_version property

Return the raw version string reported by the server (e.g. "2.42.0").

resources property

Return the version-bound generated Resources accessor; requires connect().

system_cache property

Per-client TTL cache for system-level reads; None when caching is disabled.

Functions

__init__(base_url, auth, *, timeout=30.0, connect_timeout=60.0, allow_version_fallback=False, version=None, allow_version_mismatch=False, retry_policy=None, http_limits=None, system_cache_ttl=300.0, verify=True, skip_version_probe=False)

Build a client. Call connect() or use as an async context manager before API calls.

version defaults to Dhis2.V42 — the workspace targets v42 and v43. Set Dhis2.V43 to pin to v43 explicitly, or pass None to let the client auto-detect via /api/system/info on connect(). Pinning skips that round-trip and fails fast on a server version with no matching generated module.

retry_policy (default: no retries) enables exponential-backoff retries on transient failures — connection errors plus the status codes listed on RetryPolicy.retry_statuses (default 429 / 502 / 503 / 504). Non-idempotent methods (POST, PATCH) are exempt unless the policy sets retry_non_idempotent=True. See dhis2w_client.retry.RetryPolicy for tuning knobs.

http_limits overrides the httpx connection-pool defaults (100 max connections, 20 keepalive). Raise them for high-concurrency batch workflows; lower them to protect a small DHIS2 instance from a large asyncio.gather. See docs/architecture/client.md for guidance.

system_cache_ttl (default 300 s) caps how long cached system-level reads (client.system.info(), the default categoryCombo UID, and per-key system settings) stay fresh before the next call refetches. Pass None to disable the cache entirely. connect() primes the cache from the info fetch it already performs, so the first client.system.info() after connect costs zero round-trips.

verify controls TLS certificate verification on every internal httpx.AsyncClient (the main pool plus the canonical-URL and DHIS2-shape probes). Pass False to disable verification (only safe against self-signed staging boxes) or a path to a custom CA bundle. Default True.

skip_version_probe (default False) opens the HTTP pool without the two probes connect() normally runs (canonical-URL resolution plus /api/system/info for version detection). Useful for health-checkers that want to report on those endpoints rather than have them raise, very-low-privilege PATs that can't read /api/system/info, and tests that inject a mock transport. With this flag set, version_key, raw_version, and resources raise on access — only get_raw, post_raw, get_response, and friends are usable.

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

def __init__(
    self,
    base_url: str,
    auth: AuthProvider,
    *,
    timeout: float = 30.0,
    connect_timeout: float = 60.0,
    allow_version_fallback: bool = False,
    version: Dhis2 | None = None,
    allow_version_mismatch: bool = False,
    retry_policy: RetryPolicy | None = None,
    http_limits: httpx.Limits | None = None,
    system_cache_ttl: float | None = 300.0,
    verify: bool | str = True,
    skip_version_probe: bool = False,
) -> None:
    """Build a client. Call connect() or use as an async context manager before API calls.

    `version` defaults to `Dhis2.V42` — the workspace targets v42 and v43.
    Set `Dhis2.V43` to pin to v43 explicitly, or pass `None` to let the
    client auto-detect via `/api/system/info` on `connect()`. Pinning
    skips that round-trip and fails fast on a server version with no
    matching generated module.

    `retry_policy` (default: no retries) enables exponential-backoff
    retries on transient failures — connection errors plus the status
    codes listed on `RetryPolicy.retry_statuses` (default 429 / 502 /
    503 / 504). Non-idempotent methods (POST, PATCH) are exempt unless
    the policy sets `retry_non_idempotent=True`. See
    `dhis2w_client.retry.RetryPolicy` for tuning knobs.

    `http_limits` overrides the httpx connection-pool defaults (100
    max connections, 20 keepalive). Raise them for high-concurrency
    batch workflows; lower them to protect a small DHIS2 instance
    from a large `asyncio.gather`. See
    `docs/architecture/client.md` for guidance.

    `system_cache_ttl` (default 300 s) caps how long cached system-level
    reads (`client.system.info()`, the default categoryCombo UID, and
    per-key system settings) stay fresh before the next call refetches.
    Pass `None` to disable the cache entirely. `connect()` primes the
    cache from the info fetch it already performs, so the first
    `client.system.info()` after connect costs zero round-trips.

    `verify` controls TLS certificate verification on every internal
    `httpx.AsyncClient` (the main pool plus the canonical-URL and
    DHIS2-shape probes). Pass `False` to disable verification (only
    safe against self-signed staging boxes) or a path to a custom CA
    bundle. Default `True`.

    `skip_version_probe` (default `False`) opens the HTTP pool without
    the two probes `connect()` normally runs (canonical-URL resolution
    plus `/api/system/info` for version detection). Useful for
    health-checkers that want to *report* on those endpoints rather
    than have them raise, very-low-privilege PATs that can't read
    `/api/system/info`, and tests that inject a mock transport. With
    this flag set, `version_key`, `raw_version`, and `resources` raise
    on access — only `get_raw`, `post_raw`, `get_response`, and
    friends are usable.
    """
    self._base_url = base_url.rstrip("/")
    self._auth = auth
    self._timeout = httpx.Timeout(timeout, connect=connect_timeout)
    self._retry_policy = retry_policy
    self._http_limits = http_limits
    self._http: httpx.AsyncClient | None = None
    self._verify = verify
    self._skip_version_probe = skip_version_probe
    self._version_key: str | None = None
    self._raw_version: str | None = None
    self._generated: ModuleType | None = None
    self._allow_fallback = allow_version_fallback
    self._version = version
    self._allow_version_mismatch = allow_version_mismatch
    self._resources: Any = None
    self._system_cache: SystemCache | None = (
        SystemCache(ttl=system_cache_ttl) if system_cache_ttl is not None else None
    )
    self.system: SystemModule = SystemModule(self)
    self.customize: CustomizeAccessor = CustomizeAccessor(self)
    self.tasks: TaskModule = TaskModule(self)
    self.maintenance: MaintenanceAccessor = MaintenanceAccessor(self)
    self.messaging: MessagingAccessor = MessagingAccessor(self)
    self.metadata: MetadataAccessor = MetadataAccessor(self)
    self.routes: RoutesAccessor = RoutesAccessor(self)
    self.datastore: DatastoreAccessor = DatastoreAccessor(self)
    self.maps: MapsAccessor = MapsAccessor(self)
    self.files: FilesAccessor = FilesAccessor(self)
    self.legend_sets: LegendSetsAccessor = LegendSetsAccessor(self)
    self.validation: ValidationAccessor = ValidationAccessor(self)
    self.attribute_values: AttributeValuesAccessor = AttributeValuesAccessor(self)
    self.option_sets: OptionSetsAccessor = OptionSetsAccessor(self)
    self.organisation_units: OrganisationUnitsAccessor = OrganisationUnitsAccessor(self)
    self.organisation_unit_groups: OrganisationUnitGroupsAccessor = OrganisationUnitGroupsAccessor(self)
    self.organisation_unit_group_sets: OrganisationUnitGroupSetsAccessor = OrganisationUnitGroupSetsAccessor(self)
    self.organisation_unit_levels: OrganisationUnitLevelsAccessor = OrganisationUnitLevelsAccessor(self)
    self.predictors: PredictorsAccessor = PredictorsAccessor(self)
    self.program_rules: ProgramRulesAccessor = ProgramRulesAccessor(self)
    self.sql_views: SqlViewsAccessor = SqlViewsAccessor(self)
    self.visualizations: VisualizationsAccessor = VisualizationsAccessor(self)
    self.dashboards: DashboardsAccessor = DashboardsAccessor(self)
    self.data_values: DataValuesAccessor = DataValuesAccessor(self)
    self.analytics: AnalyticsAccessor = AnalyticsAccessor(self)
    self.tracker: TrackerAccessor = TrackerAccessor(self)
    self.apps: AppsAccessor = AppsAccessor(self)
    self.data_elements: DataElementsAccessor = DataElementsAccessor(self)
    self.data_element_groups: DataElementGroupsAccessor = DataElementGroupsAccessor(self)
    self.data_element_group_sets: DataElementGroupSetsAccessor = DataElementGroupSetsAccessor(self)
    self.indicators: IndicatorsAccessor = IndicatorsAccessor(self)
    self.indicator_groups: IndicatorGroupsAccessor = IndicatorGroupsAccessor(self)
    self.indicator_group_sets: IndicatorGroupSetsAccessor = IndicatorGroupSetsAccessor(self)
    self.program_indicators: ProgramIndicatorsAccessor = ProgramIndicatorsAccessor(self)
    self.program_indicator_groups: ProgramIndicatorGroupsAccessor = ProgramIndicatorGroupsAccessor(self)
    self.category_options: CategoryOptionsAccessor = CategoryOptionsAccessor(self)
    self.category_option_groups: CategoryOptionGroupsAccessor = CategoryOptionGroupsAccessor(self)
    self.category_option_group_sets: CategoryOptionGroupSetsAccessor = CategoryOptionGroupSetsAccessor(self)
    self.categories: CategoriesAccessor = CategoriesAccessor(self)
    self.category_combos: CategoryCombosAccessor = CategoryCombosAccessor(self)
    self.category_option_combos: CategoryOptionCombosAccessor = CategoryOptionCombosAccessor(self)
    self.data_sets: DataSetsAccessor = DataSetsAccessor(self)
    self.sections: SectionsAccessor = SectionsAccessor(self)
    self.validation_rules: ValidationRulesAccessor = ValidationRulesAccessor(self)
    self.validation_rule_groups: ValidationRuleGroupsAccessor = ValidationRuleGroupsAccessor(self)
    self.predictor_groups: PredictorGroupsAccessor = PredictorGroupsAccessor(self)
    self.tracked_entity_attributes: TrackedEntityAttributesAccessor = TrackedEntityAttributesAccessor(self)
    self.tracked_entity_types: TrackedEntityTypesAccessor = TrackedEntityTypesAccessor(self)
    self.programs: ProgramsAccessor = ProgramsAccessor(self)
    self.program_stages: ProgramStagesAccessor = ProgramStagesAccessor(self)

__aenter__() async

Open the HTTP pool and run version discovery.

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

async def __aenter__(self) -> Self:
    """Open the HTTP pool and run version discovery."""
    await self.connect()
    return self

__aexit__(exc_type, exc, tb) async

Close the HTTP pool.

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

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc: BaseException | None,
    tb: TracebackType | None,
) -> None:
    """Close the HTTP pool."""
    await self.close()

connect() async

Open the HTTP pool and bind the generated module matching the remote version.

With skip_version_probe=True, skips canonical-URL resolution and the /api/system/info round-trip — leaves version_key, raw_version, and resources unset (they raise on access) and returns once the pool is open.

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

async def connect(self) -> None:
    """Open the HTTP pool and bind the generated module matching the remote version.

    With `skip_version_probe=True`, skips canonical-URL resolution and
    the `/api/system/info` round-trip — leaves `version_key`,
    `raw_version`, and `resources` unset (they raise on access) and
    returns once the pool is open.
    """
    if not self._skip_version_probe:
        resolved = await self._resolve_canonical_base_url(self._base_url, verify=self._verify)
        if resolved != self._base_url:
            self._base_url = resolved
    if self._http is None:
        kwargs: dict[str, Any] = {
            "base_url": self._base_url,
            "timeout": self._timeout,
            "verify": self._verify,
        }
        if self._retry_policy is not None:
            kwargs["transport"] = build_retry_transport(self._retry_policy)
        if self._http_limits is not None:
            kwargs["limits"] = self._http_limits
        self._http = httpx.AsyncClient(**kwargs)
    if self._skip_version_probe:
        return
    info = await self.get_raw("/api/system/info")
    self._raw_version = str(info.get("version", ""))
    if self._version is not None:
        # Caller asserted a version — verify the server's reported version
        # matches the pinned major. A v43 pin against a v42 server would
        # silently round-trip renamed/added fields wrong, so fail loud
        # unless the caller explicitly opts into the mismatch.
        if not self._allow_version_mismatch:
            self._assert_reported_version_matches_pin(self._raw_version, self._version)
        self._version_key = self._version.value
    else:
        self._version_key = self._pick_version_key(self._raw_version)
    self._generated = load(self._version_key)
    rebind_accessors_for_version(self, self._version_key)
    # Prime the system cache with the info we already fetched so
    # `client.system.info()` right after connect is a free in-process read.
    if self._system_cache is not None:
        self._system_cache.set("info", _SystemInfo.model_validate(info))
    resources_cls = getattr(self._generated, "Resources", None)
    if resources_cls is not None:
        self._resources = resources_cls(self)

close() async

Close the underlying HTTP pool.

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

async def close(self) -> None:
    """Close the underlying HTTP pool."""
    if self._http is not None:
        await self._http.aclose()
        self._http = None

get_raw(path, params=None) async

Raw GET returning parsed JSON; used internally and as an escape hatch.

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

async def get_raw(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Raw GET returning parsed JSON; used internally and as an escape hatch."""
    response = await self._request("GET", path, params=params)
    return self._parse_json(response)

get_response(path, *, params=None, extra_headers=None) async

No-raise GET returning the raw httpx.Response; escape hatch for caller-side status logic.

Skips the 4xx/5xx raise block in _request so callers can inspect status / headers / non-JSON bodies themselves. The auth header is still applied. Use when:

• A 4xx / 5xx is a fact you want to report (health-checkers).
• You need the raw Content-Type (SSO / proxy-page detection).
• You're calling a reverse-proxied route under /api/routes/<code>/run/... where a 502 means "DHIS2 reached, downstream didn't" rather than "API error".

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

async def get_response(
    self,
    path: str,
    *,
    params: dict[str, Any] | None = None,
    extra_headers: dict[str, str] | None = None,
) -> httpx.Response:
    """No-raise GET returning the raw `httpx.Response`; escape hatch for caller-side status logic.

    Skips the 4xx/5xx raise block in `_request` so callers can inspect
    status / headers / non-JSON bodies themselves. The auth header is
    still applied. Use when:

    - A 4xx / 5xx is a *fact you want to report* (health-checkers).
    - You need the raw `Content-Type` (SSO / proxy-page detection).
    - You're calling a reverse-proxied route under
      `/api/routes/<code>/run/...` where a 502 means "DHIS2 reached,
      downstream didn't" rather than "API error".
    """
    return await self._request(
        "GET",
        path,
        params=params,
        extra_headers=extra_headers,
        raise_for_status=False,
    )

get(path, model, params=None) async

Typed GET returning an instance of model parsed from JSON.

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

async def get[T: BaseModel](
    self,
    path: str,
    model: type[T],
    params: dict[str, Any] | None = None,
) -> T:
    """Typed GET returning an instance of `model` parsed from JSON."""
    raw = await self.get_raw(path, params=params)
    return model.model_validate(raw)

post(path, body, *, model, params=None) async

Typed POST returning an instance of model parsed from JSON.

Used most often with model=WebMessageResponse to parse /api/metadata envelopes into the typed summary shape without a trailing WebMessageResponse.model_validate(raw) at the call site.

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

async def post[T: BaseModel](
    self,
    path: str,
    body: Any,
    *,
    model: type[T],
    params: dict[str, Any] | None = None,
) -> T:
    """Typed POST returning an instance of `model` parsed from JSON.

    Used most often with `model=WebMessageResponse` to parse
    `/api/metadata` envelopes into the typed summary shape without
    a trailing `WebMessageResponse.model_validate(raw)` at the
    call site.
    """
    raw = await self.post_raw(path, body, params=params)
    return model.model_validate(raw)

put(path, body, *, model, params=None) async

Typed PUT returning an instance of model parsed from JSON.

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

async def put[T: BaseModel](
    self,
    path: str,
    body: Any,
    *,
    model: type[T],
    params: dict[str, Any] | None = None,
) -> T:
    """Typed PUT returning an instance of `model` parsed from JSON."""
    raw = await self.put_raw(path, body, params=params)
    return model.model_validate(raw)

patch(path, body, *, model, params=None, content_type='application/json-patch+json') async

Typed PATCH returning an instance of model parsed from JSON.

Used with model=WebMessageResponse for metadata patches so the call site doesn't need a trailing WebMessageResponse.model_validate(raw).

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

async def patch[T: BaseModel](
    self,
    path: str,
    body: Any,
    *,
    model: type[T],
    params: dict[str, Any] | None = None,
    content_type: str = "application/json-patch+json",
) -> T:
    """Typed PATCH returning an instance of `model` parsed from JSON.

    Used with `model=WebMessageResponse` for metadata patches so the
    call site doesn't need a trailing `WebMessageResponse.model_validate(raw)`.
    """
    raw = await self.patch_raw(path, body, params=params, content_type=content_type)
    return model.model_validate(raw)

delete(path, *, model, params=None) async

Typed DELETE returning an instance of model parsed from JSON.

Most callers use model=WebMessageResponse since DHIS2 deletes return the standard envelope.

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

async def delete[T: BaseModel](
    self,
    path: str,
    *,
    model: type[T],
    params: dict[str, Any] | None = None,
) -> T:
    """Typed DELETE returning an instance of `model` parsed from JSON.

    Most callers use `model=WebMessageResponse` since DHIS2 deletes
    return the standard envelope.
    """
    raw = await self.delete_raw(path, params=params)
    return model.model_validate(raw)

post_raw(path, body=None, *, params=None) async

Raw POST returning parsed JSON.

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

async def post_raw(self, path: str, body: Any = None, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Raw POST returning parsed JSON."""
    response = await self._request("POST", path, params=params, json=body)
    return self._parse_json(response)

put_raw(path, body=None, *, params=None) async

Raw PUT returning parsed JSON.

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

async def put_raw(self, path: str, body: Any = None, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Raw PUT returning parsed JSON."""
    response = await self._request("PUT", path, params=params, json=body)
    return self._parse_json(response)

delete_raw(path, *, params=None) async

Raw DELETE returning parsed JSON (or empty dict).

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

async def delete_raw(self, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Raw DELETE returning parsed JSON (or empty dict)."""
    response = await self._request("DELETE", path, params=params)
    return self._parse_json(response)

patch_raw(path, body, *, params=None, content_type='application/json-patch+json') async

Raw PATCH returning parsed JSON. Defaults to JSON Patch (RFC 6902) content type.

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

async def patch_raw(
    self,
    path: str,
    body: Any,
    *,
    params: dict[str, Any] | None = None,
    content_type: str = "application/json-patch+json",
) -> dict[str, Any]:
    """Raw PATCH returning parsed JSON. Defaults to JSON Patch (RFC 6902) content type."""
    response = await self._request(
        "PATCH",
        path,
        params=params,
        json=body,
        extra_headers={"Content-Type": content_type},
    )
    return self._parse_json(response)