System module
Me, SystemInfo, DhisCalendar, and the small SystemModule accessor bound to Dhis2Client.system.
system
Typed accessors for /api/system/info and /api/me (non-metadata system endpoints).
SystemInfo re-exports from generated/v42/oas — OpenAPI ships the full
shape (46 fields including buildTime, databaseInfo, analytics-table
timings, memory info). Me stays hand-written because /api/me isn't in
the OpenAPI spec under that name.
Classes
SystemInfo
Bases: BaseModel
OpenAPI schema SystemInfo.
Source code in packages/dhis2w-client/src/dhis2w_client/generated/v42/oas/system_info.py
| class SystemInfo(_BaseModel):
"""OpenAPI schema `SystemInfo`."""
model_config = _ConfigDict(extra="allow", populate_by_name=True, defer_build=True)
buildTime: datetime | None = None
calendar: str | None = None
clusterHostname: str | None = None
contextPath: str | None = None
cpuCores: int | None = None
databaseInfo: DatabaseInfo | None = None
dateFormat: str | None = None
emailConfigured: bool | None = None
encryption: bool | None = None
environmentVariable: str | None = None
externalDirectory: str | None = None
fileStoreProvider: str | None = None
instanceBaseUrl: str | None = None
intervalSinceLastAnalyticsTablePartitionSuccess: str | None = None
intervalSinceLastAnalyticsTableSuccess: str | None = None
isMetadataSyncEnabled: bool | None = None
isMetadataVersionEnabled: bool | None = None
jasperReportsVersion: str | None = None
javaOpts: str | None = None
javaVendor: str | None = None
javaVersion: str | None = None
lastAnalyticsTablePartitionRuntime: str | None = None
lastAnalyticsTablePartitionSuccess: datetime | None = None
lastAnalyticsTableRuntime: str | None = None
lastAnalyticsTableSuccess: datetime | None = None
lastMetadataVersionSyncAttempt: datetime | None = None
lastSystemMonitoringSuccess: datetime | None = None
memoryInfo: str | None = None
nodeId: str | None = None
osArchitecture: str | None = None
osName: str | None = None
osVersion: str | None = None
readOnlyMode: str | None = None
readReplicaCount: int | None = None
redisEnabled: bool | None = None
redisHostname: str | None = None
revision: str | None = None
serverDate: datetime | None = None
serverTimeZoneDisplayName: str | None = None
serverTimeZoneId: str | None = None
systemId: str | None = None
systemMetadataVersion: str | None = None
systemMonitoringUrl: str | None = None
systemName: str | None = None
userAgent: str | None = None
version: str | None = None
|
DhisCalendar
Bases: StrEnum
Canonical DHIS2 calendar names (the values DHIS2 accepts on keyCalendar).
Matches the @Component name() of every calendar implementation under
org.hisp.dhis.calendar.impl on dhis2/dhis2w-core 2.42 — iso8601 is
the server default. Pass any of these to SystemModule.set_calendar().
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| class DhisCalendar(StrEnum):
"""Canonical DHIS2 calendar names (the values DHIS2 accepts on `keyCalendar`).
Matches the `@Component` `name()` of every calendar implementation under
`org.hisp.dhis.calendar.impl` on `dhis2/dhis2w-core` 2.42 — `iso8601` is
the server default. Pass any of these to `SystemModule.set_calendar()`.
"""
COPTIC = "coptic"
ETHIOPIAN = "ethiopian"
GREGORIAN = "gregorian"
ISLAMIC = "islamic"
ISO8601 = "iso8601"
JULIAN = "julian"
NEPALI = "nepali"
PERSIAN = "persian"
THAI = "thai"
|
DisplayRef
Bases: BaseModel
Minimal DHIS2 object reference carrying id + displayName for CLI rendering.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| class DisplayRef(BaseModel):
"""Minimal DHIS2 object reference carrying id + displayName for CLI rendering."""
model_config = ConfigDict(extra="allow")
id: str | None = None
displayName: str | None = None
|
Me
Bases: BaseModel
Shape of /api/me for the authenticated user (common fields; unknown preserved).
Hand-written: /api/me doesn't appear in the OpenAPI spec as a component
schema, so the emitter can't generate it. This is a stable-enough subset
of the real response — extra="allow" preserves anything else DHIS2 ships.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| class Me(BaseModel):
"""Shape of `/api/me` for the authenticated user (common fields; unknown preserved).
Hand-written: `/api/me` doesn't appear in the OpenAPI spec as a component
schema, so the emitter can't generate it. This is a stable-enough subset
of the real response — `extra="allow"` preserves anything else DHIS2 ships.
"""
model_config = ConfigDict(extra="allow")
id: str | None = None
username: str | None = None
displayName: str | None = None
email: str | None = None
firstName: str | None = None
surname: str | None = None
lastLogin: str | None = None
created: str | None = None
authorities: list[str] | None = None
organisationUnits: list[DisplayRef] | None = None
dataViewOrganisationUnits: list[DisplayRef] | None = None
userGroups: list[DisplayRef] | None = None
programs: list[DisplayRef] | None = None
@field_validator(
"organisationUnits",
"dataViewOrganisationUnits",
"userGroups",
"programs",
mode="before",
)
@classmethod
def _coerce_ref_lists(cls, value: Any) -> Any:
"""Accept bare UID strings as well as `{id, displayName}` dicts on every ref-list field."""
return _coerce_refs(value)
|
SystemModule
Accessor bound to a Dhis2Client exposing system-level endpoints.
info(), default_category_combo_uid(), and setting() read through
the client's SystemCache (5-minute TTL by default — see
Dhis2Client(system_cache_ttl=...)). Pass use_cache=False to force a
fresh fetch; call invalidate_cache() when you know the upstream
changed (settings update through another process, default category
combo renamed via Admin, etc.).
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| class SystemModule:
"""Accessor bound to a `Dhis2Client` exposing system-level endpoints.
`info()`, `default_category_combo_uid()`, and `setting()` read through
the client's `SystemCache` (5-minute TTL by default — see
`Dhis2Client(system_cache_ttl=...)`). Pass `use_cache=False` to force a
fresh fetch; call `invalidate_cache()` when you know the upstream
changed (settings update through another process, default category
combo renamed via Admin, etc.).
"""
def __init__(self, client: Dhis2Client) -> None:
"""Bind to the sharing client."""
self._client = client
async def info(self, *, use_cache: bool = True) -> SystemInfo:
"""Fetch `/api/system/info` and return a typed `SystemInfo` (cached by default)."""
cache = self._client.system_cache
if cache is None or not use_cache:
return await self._client.get("/api/system/info", model=SystemInfo)
return await cache.get_or_fetch(
"info",
lambda: self._client.get("/api/system/info", model=SystemInfo),
)
async def me(self) -> Me:
"""Fetch `/api/me` and return the typed authenticated user profile.
Not cached — `/api/me` is per-authenticated-user state and the
typical use case (scripts impersonating a specific user) benefits
from fresh reads.
"""
return await self._client.get("/api/me", model=Me)
async def default_category_combo_uid(self, *, use_cache: bool = True) -> str:
"""Return the UID of the DHIS2 default category combo (cached by default).
DHIS2 stamps every data element / data set with a categoryCombo; the
built-in `default` combo is what every unspecified reference points
at. Fetching it once per script is a small but real bootstrap cost.
"""
cache = self._client.system_cache
if cache is None or not use_cache:
return await self._fetch_default_category_combo_uid()
return await cache.get_or_fetch(
"default_category_combo_uid",
self._fetch_default_category_combo_uid,
)
async def _fetch_default_category_combo_uid(self) -> str:
"""Look up the default categoryCombo UID via `/api/categoryCombos?filter=name:eq:default`."""
raw = await self._client.get_raw(
"/api/categoryCombos",
params={"filter": "name:eq:default", "fields": "id", "paging": "false"},
)
combos = raw.get("categoryCombos") or []
if not combos or not isinstance(combos, list):
raise RuntimeError(
"DHIS2 returned no categoryCombo named 'default' — "
"every DHIS2 instance ships one; check the base URL + auth.",
)
first = combos[0]
if not isinstance(first, dict) or "id" not in first:
raise RuntimeError(f"malformed categoryCombos response: {raw!r}")
return str(first["id"])
async def calendar(self, *, use_cache: bool = True) -> str:
"""Return the active DHIS2 calendar name (the `keyCalendar` setting).
DHIS2 ships nine calendar implementations — see `DhisCalendar` for the
canonical set. The server default is `iso8601`, so when the setting is
unset this returns `iso8601`. Cached per the same TTL as other system
reads; pass `use_cache=False` to force a fresh fetch.
"""
value = await self.setting("keyCalendar", use_cache=use_cache)
return value or DhisCalendar.ISO8601.value
async def set_calendar(self, calendar: DhisCalendar | str) -> None:
"""Write `keyCalendar` so the server uses the named calendar going forward.
Accepts either a `DhisCalendar` member or a raw string for forward
compatibility with calendars that may ship after this client. The
change takes effect on the next request that resolves periods —
rarely something a script needs to do, hence no convenience wrappers
per calendar.
"""
await self.set_setting("keyCalendar", str(calendar))
async def setting(self, key: str, *, use_cache: bool = True) -> str | None:
"""Fetch `/api/systemSettings/{key}` and return its value (cached per key by default).
DHIS2 returns a plain value when `Accept: text/plain` or a
`{key: value}` envelope on JSON. This helper requests the JSON
shape + pulls the value. Returns `None` when the key is unset.
"""
cache = self._client.system_cache
cache_key = f"setting:{key}"
if cache is None or not use_cache:
return await self._fetch_setting(key)
return await cache.get_or_fetch(cache_key, lambda: self._fetch_setting(key))
async def _fetch_setting(self, key: str) -> str | None:
"""Read one system setting via `/api/systemSettings/{key}`; returns `None` when absent."""
try:
raw = await self._client.get_raw(f"/api/systemSettings/{key}")
except Exception: # noqa: BLE001 — treat any 4xx/5xx as 'unset' to keep the cache flow linear
return None
if not raw:
return None
value = raw.get(key)
if value is None:
return None
return str(value)
async def set_setting(self, key: str, value: str | None) -> None:
"""Write one system setting via `/api/systemSettings/{key}` (or DELETE when value is None).
DHIS2's systemSettings endpoint takes the new value as a
`text/plain` body on POST. Passing `None` sends a DELETE instead
so the key reverts to whatever hard-coded default the server
ships with. Invalidates the in-memory cache entry for `key`.
"""
if value is None:
await self._client._request("DELETE", f"/api/systemSettings/{key}") # noqa: SLF001
else:
await self._client._request( # noqa: SLF001
"POST",
f"/api/systemSettings/{key}",
content=value.encode("utf-8"),
extra_headers={"Content-Type": "text/plain"},
)
self.invalidate_cache(key=f"setting:{key}")
def invalidate_cache(self, *, key: str | None = None) -> None:
"""Drop one cache key (e.g. `"info"`, `"setting:keyFlag"`) or every cached value."""
cache = self._client.system_cache
if cache is None:
return
cache.invalidate(key)
|
Functions
__init__(client)
Bind to the sharing client.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| def __init__(self, client: Dhis2Client) -> None:
"""Bind to the sharing client."""
self._client = client
|
info(*, use_cache=True)
async
Fetch /api/system/info and return a typed SystemInfo (cached by default).
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def info(self, *, use_cache: bool = True) -> SystemInfo:
"""Fetch `/api/system/info` and return a typed `SystemInfo` (cached by default)."""
cache = self._client.system_cache
if cache is None or not use_cache:
return await self._client.get("/api/system/info", model=SystemInfo)
return await cache.get_or_fetch(
"info",
lambda: self._client.get("/api/system/info", model=SystemInfo),
)
|
me()
async
Fetch /api/me and return the typed authenticated user profile.
Not cached — /api/me is per-authenticated-user state and the
typical use case (scripts impersonating a specific user) benefits
from fresh reads.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def me(self) -> Me:
"""Fetch `/api/me` and return the typed authenticated user profile.
Not cached — `/api/me` is per-authenticated-user state and the
typical use case (scripts impersonating a specific user) benefits
from fresh reads.
"""
return await self._client.get("/api/me", model=Me)
|
default_category_combo_uid(*, use_cache=True)
async
Return the UID of the DHIS2 default category combo (cached by default).
DHIS2 stamps every data element / data set with a categoryCombo; the
built-in default combo is what every unspecified reference points
at. Fetching it once per script is a small but real bootstrap cost.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def default_category_combo_uid(self, *, use_cache: bool = True) -> str:
"""Return the UID of the DHIS2 default category combo (cached by default).
DHIS2 stamps every data element / data set with a categoryCombo; the
built-in `default` combo is what every unspecified reference points
at. Fetching it once per script is a small but real bootstrap cost.
"""
cache = self._client.system_cache
if cache is None or not use_cache:
return await self._fetch_default_category_combo_uid()
return await cache.get_or_fetch(
"default_category_combo_uid",
self._fetch_default_category_combo_uid,
)
|
calendar(*, use_cache=True)
async
Return the active DHIS2 calendar name (the keyCalendar setting).
DHIS2 ships nine calendar implementations — see DhisCalendar for the
canonical set. The server default is iso8601, so when the setting is
unset this returns iso8601. Cached per the same TTL as other system
reads; pass use_cache=False to force a fresh fetch.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def calendar(self, *, use_cache: bool = True) -> str:
"""Return the active DHIS2 calendar name (the `keyCalendar` setting).
DHIS2 ships nine calendar implementations — see `DhisCalendar` for the
canonical set. The server default is `iso8601`, so when the setting is
unset this returns `iso8601`. Cached per the same TTL as other system
reads; pass `use_cache=False` to force a fresh fetch.
"""
value = await self.setting("keyCalendar", use_cache=use_cache)
return value or DhisCalendar.ISO8601.value
|
set_calendar(calendar)
async
Write keyCalendar so the server uses the named calendar going forward.
Accepts either a DhisCalendar member or a raw string for forward
compatibility with calendars that may ship after this client. The
change takes effect on the next request that resolves periods —
rarely something a script needs to do, hence no convenience wrappers
per calendar.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def set_calendar(self, calendar: DhisCalendar | str) -> None:
"""Write `keyCalendar` so the server uses the named calendar going forward.
Accepts either a `DhisCalendar` member or a raw string for forward
compatibility with calendars that may ship after this client. The
change takes effect on the next request that resolves periods —
rarely something a script needs to do, hence no convenience wrappers
per calendar.
"""
await self.set_setting("keyCalendar", str(calendar))
|
setting(key, *, use_cache=True)
async
Fetch /api/systemSettings/{key} and return its value (cached per key by default).
DHIS2 returns a plain value when Accept: text/plain or a
{key: value} envelope on JSON. This helper requests the JSON
shape + pulls the value. Returns None when the key is unset.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def setting(self, key: str, *, use_cache: bool = True) -> str | None:
"""Fetch `/api/systemSettings/{key}` and return its value (cached per key by default).
DHIS2 returns a plain value when `Accept: text/plain` or a
`{key: value}` envelope on JSON. This helper requests the JSON
shape + pulls the value. Returns `None` when the key is unset.
"""
cache = self._client.system_cache
cache_key = f"setting:{key}"
if cache is None or not use_cache:
return await self._fetch_setting(key)
return await cache.get_or_fetch(cache_key, lambda: self._fetch_setting(key))
|
set_setting(key, value)
async
Write one system setting via /api/systemSettings/{key} (or DELETE when value is None).
DHIS2's systemSettings endpoint takes the new value as a
text/plain body on POST. Passing None sends a DELETE instead
so the key reverts to whatever hard-coded default the server
ships with. Invalidates the in-memory cache entry for key.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| async def set_setting(self, key: str, value: str | None) -> None:
"""Write one system setting via `/api/systemSettings/{key}` (or DELETE when value is None).
DHIS2's systemSettings endpoint takes the new value as a
`text/plain` body on POST. Passing `None` sends a DELETE instead
so the key reverts to whatever hard-coded default the server
ships with. Invalidates the in-memory cache entry for `key`.
"""
if value is None:
await self._client._request("DELETE", f"/api/systemSettings/{key}") # noqa: SLF001
else:
await self._client._request( # noqa: SLF001
"POST",
f"/api/systemSettings/{key}",
content=value.encode("utf-8"),
extra_headers={"Content-Type": "text/plain"},
)
self.invalidate_cache(key=f"setting:{key}")
|
invalidate_cache(*, key=None)
Drop one cache key (e.g. "info", "setting:keyFlag") or every cached value.
Source code in packages/dhis2w-client/src/dhis2w_client/system.py
| def invalidate_cache(self, *, key: str | None = None) -> None:
"""Drop one cache key (e.g. `"info"`, `"setting:keyFlag"`) or every cached value."""
cache = self._client.system_cache
if cache is None:
return
cache.invalidate(key)
|
System cache
Dhis2Client ships a per-client TTL-bounded cache for system-level reads.
system_cache
TTL-bounded in-memory cache for system-level reads on a single Dhis2Client.
Scoped to one client instance (not shared across clients). For scripts that
reuse one client for many reads, the cache kicks in on the second call of
each cached endpoint — /api/system/info, the default categoryCombo UID,
and per-key system-setting reads.
Concurrency: a per-key asyncio.Lock dedupes in-flight fetches so a fan-out
of 100 asyncio.gather tasks triggers one network call, not 100.
Classes
SystemCache
Per-client TTL cache for system-level reads.
ttl is the max age (seconds) of any cached entry before the next read
triggers a refetch. Entries never evict on their own — on a long-lived
client, call invalidate() when you know the upstream changed
(rename via the Admin UI, settings update through another process, etc.).
Source code in packages/dhis2w-client/src/dhis2w_client/system_cache.py
| class SystemCache:
"""Per-client TTL cache for system-level reads.
`ttl` is the max age (seconds) of any cached entry before the next read
triggers a refetch. Entries never evict on their own — on a long-lived
client, call `invalidate()` when you know the upstream changed
(rename via the Admin UI, settings update through another process, etc.).
"""
def __init__(self, ttl: float) -> None:
"""Bind the TTL; start with an empty store."""
self._ttl = ttl
self._store: dict[str, _CacheEntry] = {}
self._locks: dict[str, asyncio.Lock] = {}
@property
def ttl(self) -> float:
"""Max age (seconds) for any cached entry before the next read refetches."""
return self._ttl
def _lock_for(self, key: str) -> asyncio.Lock:
"""Lazily create one lock per key — dedupes in-flight fetches for the same key."""
lock = self._locks.get(key)
if lock is None:
lock = asyncio.Lock()
self._locks[key] = lock
return lock
async def get_or_fetch[T](self, key: str, fetcher: Callable[[], Awaitable[T]]) -> T:
"""Return a fresh cached value, or run `fetcher()` and cache its result."""
now = time.monotonic()
entry = self._store.get(key)
if entry is not None and entry.expires_at > now:
return entry.value # type: ignore[no-any-return]
async with self._lock_for(key):
# Re-check under the lock — a concurrent task may have populated it.
entry = self._store.get(key)
now = time.monotonic()
if entry is not None and entry.expires_at > now:
return entry.value # type: ignore[no-any-return]
value = await fetcher()
self._store[key] = _CacheEntry(value=value, expires_at=now + self._ttl)
return value
def set(self, key: str, value: Any) -> None:
"""Prime the cache — used by `connect()` to avoid a second round-trip to `/api/system/info`."""
self._store[key] = _CacheEntry(value=value, expires_at=time.monotonic() + self._ttl)
def invalidate(self, key: str | None = None) -> None:
"""Drop one key (when `key` is set) or every key (when `key is None`)."""
if key is None:
self._store.clear()
else:
self._store.pop(key, None)
|
Attributes
ttl
property
Max age (seconds) for any cached entry before the next read refetches.
Functions
__init__(ttl)
Bind the TTL; start with an empty store.
Source code in packages/dhis2w-client/src/dhis2w_client/system_cache.py
| def __init__(self, ttl: float) -> None:
"""Bind the TTL; start with an empty store."""
self._ttl = ttl
self._store: dict[str, _CacheEntry] = {}
self._locks: dict[str, asyncio.Lock] = {}
|
get_or_fetch(key, fetcher)
async
Return a fresh cached value, or run fetcher() and cache its result.
Source code in packages/dhis2w-client/src/dhis2w_client/system_cache.py
| async def get_or_fetch[T](self, key: str, fetcher: Callable[[], Awaitable[T]]) -> T:
"""Return a fresh cached value, or run `fetcher()` and cache its result."""
now = time.monotonic()
entry = self._store.get(key)
if entry is not None and entry.expires_at > now:
return entry.value # type: ignore[no-any-return]
async with self._lock_for(key):
# Re-check under the lock — a concurrent task may have populated it.
entry = self._store.get(key)
now = time.monotonic()
if entry is not None and entry.expires_at > now:
return entry.value # type: ignore[no-any-return]
value = await fetcher()
self._store[key] = _CacheEntry(value=value, expires_at=now + self._ttl)
return value
|
set(key, value)
Prime the cache — used by connect() to avoid a second round-trip to /api/system/info.
Source code in packages/dhis2w-client/src/dhis2w_client/system_cache.py
| def set(self, key: str, value: Any) -> None:
"""Prime the cache — used by `connect()` to avoid a second round-trip to `/api/system/info`."""
self._store[key] = _CacheEntry(value=value, expires_at=time.monotonic() + self._ttl)
|
invalidate(key=None)
Drop one key (when key is set) or every key (when key is None).
Source code in packages/dhis2w-client/src/dhis2w_client/system_cache.py
| def invalidate(self, key: str | None = None) -> None:
"""Drop one key (when `key` is set) or every key (when `key is None`)."""
if key is None:
self._store.clear()
else:
self._store.pop(key, None)
|
Calendar
DHIS2 ships nine canonical calendars — the names are the values DHIS2 accepts on the keyCalendar system setting. DhisCalendar enumerates them; iso8601 is the server default.
async with Dhis2Client(...) as client:
name = await client.system.calendar() # "iso8601" by default
await client.system.set_calendar(DhisCalendar.ETHIOPIAN)
The CLI mirrors this:
dhis2 system calendar # print current value
dhis2 system calendar ethiopian # set new value, with interactive y/N confirmation
dhis2 system calendar ethiopian --yes # skip the prompt (CI / scripts)
WARNING: only change the calendar when it is genuinely required. Switching keyCalendar after data collection has started can leave existing periods unreadable and break analytics. The CLI prints the current value, the new value, and a warning, then prompts Change calendar? [y/N] with N as the default. Calling dhis2 system calendar <same-value> is a no-op and does not prompt. Library callers do not get this confirmation — client.system.set_calendar() writes immediately.
NOTE: a local single-replica infra/ stack (DHIS2 2.42.4) round-trips the write end-to-end for all nine values. On the shared play.im.dhis2.org/dev-2-42 instance the same call returns 200 OK with a confirming message but the value does not persist on the next read — a deployment-topology issue, not a dhis2w-core regression. See BUGS.md entry 32.