For rapid development of Python applications around datarhei Core / MediaCore. Requires Python 3.11+ and datarhei Core v16.10+.

• Features
• Install
• Usage
• API definitions
  • General
  • Cluster
  • Cluster Database
  • Cluster IAM
  • Cluster Node
  • Cluster Process
  • Config
  • Events
  • Filesystem
  • IAM
  • Log
  • Metadata
  • Metrics
  • Process
  • Process Playout (commercial extention)
  • Report
  • RTMP
  • Session
  • Skills
  • SRT
  • Widget
  • Misc
• Examples
  • GET token data
  • GET processes
  • GET/POST/PUT/DELETE process
• API models
• Error handling
• Developing & testing
• Changelog
• Contributing
• Licence

• Async & sync support
• Async event streaming (SSE / NDJSON) as endless async generators
• Request & response validation
• Retries and timeout settings per request
• Automatic JWT renewal (+ public refresh()/arefresh(), one-shot 401 auto-retry)
• pydantic Models
• HTTPX

pip install https://github.com/datarhei/core-client-python/archive/refs/heads/main.tar.gz

pip install https://github.com/datarhei/core-client-python/archive/refs/tags/{release_tag}.tar.gz

{release_tag} like 1.0.0

from datarhei.mediacore import MediaCoreClient, AsyncMediaCoreClient

core_client is still supported as a backward-compatible alias during migration.

• base_url: str
• optional: basic auth jwt username: str = None, password: str = None
• optional: token injection access_token: str = None, refresh_token: str = None
• optional: httpx global settings retries: int = 3, timeout: float = 10.0

from datarhei.mediacore import MediaCoreClient

client = MediaCoreClient(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

about = client.about_get()
print(about)

import asyncio
from datarhei.mediacore import AsyncMediaCoreClient

client = AsyncMediaCoreClient(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

async def main():
    about = await client.about_get()
    print(about)

asyncio.run(main())

Note on domain vs. domainpattern (IAM domains / cluster)

• List endpoints (v3_process_get_list, v3_cluster_process_get_list): domain means "act in the context of this domain" and does not filter the result. To filter a list by domain, use domainpattern (a glob).
• Detail endpoints (v3_process_get, …_get_probe, …_get_state, …): a domain-scoped process returns 404 unless the matching domain is passed — even when the id was taken from a list response.

• GET /api

  about()

• POST /api/graph/query

  graph_query(query: GraphQuery)

  Model: GraphQuery

client.refresh()          # sync: refresh the access token now (re-login fallback)
await client.arefresh()   # async

Tokens are also renewed lazily before each request. In addition, every request transparently retries once on a 401 (access token invalidated server-side by a restart or secret rotation). Streaming methods are exempt — handle 401 in your reconnect loop via arefresh().

• POST /api/v3/events

  v3_events_post(filters: EventFilters)

• POST /api/v3/events/media/{type}

  v3_events_post_media(type: str, glob: str = "")

  Model: EventFilters

The event endpoints are endless streams. On the AsyncClient they are exposed as async generators that yield event by event (no read timeout). The default delivery is a lightweight (event_type, data_str) tuple — no per-event validation (a busy cluster emits thousands of events per second). Pass model= for typed events (LogEvent for the log streams, ProcessEvent for the process streams), or frame=False for raw lines.

Streaming methods (async only):

• v3_events_stream(filters=…, frame=True, model=None) → POST /api/v3/events
• v3_events_log_stream(filters=…) → POST /api/v3/events/log (Core alias of /api/v3/events)
• v3_events_process_stream(filters=…) → POST /api/v3/events/process
• v3_cluster_events_stream(filters=…) → POST /api/v3/cluster/events
• v3_cluster_events_log_stream(filters=…) → POST /api/v3/cluster/events/log (Core alias of /api/v3/cluster/events)
• v3_cluster_events_process_stream(filters=…) → POST /api/v3/cluster/events/process

import asyncio
from datarhei.mediacore import AsyncMediaCoreClient
from datarhei.mediacore.base.models.v3 import EventFilters, ProcessEventFilter

async def main():
    client = AsyncMediaCoreClient(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
    await client.alogin()

    filters = EventFilters(filters=[ProcessEventFilter(type="progress")])
    async for event_type, data in client.v3_cluster_events_process_stream(filters=filters):
        print(event_type, data)   # data is the raw JSON line (str)

asyncio.run(main())

For the log streams, event must name an existing component exactly (lowercased); the other fields are regexes:

from datarhei.mediacore.base.models.v3 import EventFilters, LogEventFilter

filters = EventFilters(filters=[LogEventFilter(event="iam", level="debug")])
async for _, data in client.v3_cluster_events_log_stream(filters=filters):
    print(json.loads(data))

Notes:

• Filter models: LogEventFilter for the log-event streams (/api/v3/events, /api/v3/events/log, /api/v3/cluster/events, /api/v3/cluster/events/log); ProcessEventFilter (filter by type, domain, pid, core_id) for /api/v3/events/process and /api/v3/cluster/events/process. Omitting filters delivers all events (a firehose — filter tightly). Raw dict filters are also accepted.
• Filter values are case-insensitive, unanchored regexes (e.g. type="progress", type="progress|report"), AND-combined across fields — except LogEventFilter.event, which Core never compiles as a regex. It is matched as an exact, lowercased component name (filter[strings.ToLower(f.Component)], then a plain map lookup on the event's component), so a name that does not exist silently drops every event. The components a Core emits are values like iam, http, filesystem, processlimiter, cluster — not process or report names. A LogEventFilter without event builds the key "" and matches nothing, so always set it. data additionally requires the key to exist on the event.
• Wire format: the streams request NDJSON (accept: application/x-json-stream) and therefore yield ("message", data_str), with the component inside the JSON as event. Core's log endpoints would default to SSE, but its SSE branch is broken for a cluster — it serializes an empty LogEvent for every event. Pass accept=SSE (from core_client.base.api._stream) for the SSE framing, where event_type is the component. Keepalives are dropped in framed mode and passed through with frame=False.
• Connect errors (e.g. 401) are raised as CoreAPIError on connection, so you can refresh and reconnect; network errors propagate as httpx exceptions.
• Reconnect is the caller's responsibility. The generator ends normally on EOF; it does not retry. Wrap it in your own reconnect/backoff loop for long-running use, and call await client.arefresh() on a 401 before reconnecting.
• Cancelling the task or calling aclose() on the generator closes the connection cleanly.

• GET /api/v3/cluster

  v3_cluster_get()

• GET /api/v3/cluster/healthy

  v3_cluster_get_healthy()

• GET /api/v3/cluster/deployments

  v3_cluster_get_deployments()

• GET /api/v3/cluster/fs/{storage}

  v3_cluster_fs_get_file_list(storage: str, glob: str = "", sort: str = "", order: str = "")

• GET /api/v3/cluster/snapshot

  v3_cluster_get_snapshot()

• POST /api/v3/cluster/events

  v3_cluster_post_events(filters: EventFilters)

  Model: EventFilters

• PUT /api/v3/cluster/transfer/{id}

  v3_cluster_put_transfer(id: str)

• PUT /api/v3/cluster/leave

  v3_cluster_put_leave()

• GET /api/v3/cluster/db/locks

  v3_cluster_db_get_lock_list()

• GET /api/v3/cluster/db/policies

  v3_cluster_db_get_policy_list()

• GET /api/v3/cluster/db/process

  v3_cluster_db_get_process_list()

• GET /api/v3/cluster/db/map/process

  v3_cluster_db_get_process_map()

• GET /api/v3/cluster/db/map/reallocate

  v3_cluster_db_get_reallocate_map()

• GET /api/v3/cluster/db/process/{id}

  v3_cluster_db_get_process(id: str, domain: str = "")

• GET /api/v3/cluster/db/kv

  v3_cluster_db_get_kv()

• GET /api/v3/cluster/db/node

  v3_cluster_db_get_node_list()

• GET /api/v3/cluster/db/user

  v3_cluster_db_get_user_list()

• GET /api/v3/cluster/db/user/{name}

  v3_cluster_db_get_user(name: str)

• GET /api/v3/cluster/iam/user

  v3_cluster_iam_get_user_list()

• POST /api/v3/cluster/iam/user

  v3_cluster_iam_post_user(config: IamUser)

  Model: IamUser

• GET /api/v3/cluster/iam/user/{name}

  v3_cluster_iam_get_user(name: str)

• PUT /api/v3/cluster/iam/user/{name}

  v3_cluster_iam_put_user(name: str, domain: str = "", config: IamUser)

  Model: IamUser

• PUT /api/v3/cluster/iam/user/{name}/policies

  v3_cluster_iam_put_user_policy_list(name: str, domain: str = "", config: IamUserPolicyList)

  Model: IamUserPolicyList

• DELETE /api/v3/cluster/iam/user/{name}

  v3_cluster_iam_delete_user(name: str)

• GET /api/v3/cluster/iam/policies

  v3_cluster_iam_get_policy_list()

• GET /api/v3/cluster/iam/reload

  v3_cluster_iam_get_reload()

• GET /api/v3/cluster/node

  v3_cluster_node_get_list()

• GET /api/v3/cluster/node/{id}

  v3_cluster_node_get(id: str)

• GET /api/v3/cluster/node/{id}/files

  v3_cluster_node_get_files(id: str)

• GET /api/v3/cluster/node/{id}/process

  v3_cluster_node_get_process_list(id: str)

• GET /api/v3/cluster/node/{id}/version

  v3_cluster_node_get_version(id: str)

• GET /api/v3/cluster/process

  v3_cluster_process_get_list(domain: str = "", filter: str = "", reference: str = "", id: str = "", idpattern: str = "", refpattern: str = "", domainpattern: str = "")

• POST /api/v3/cluster/process

  v3_cluster_process_post(config: ProcessConfig)

  Model: ProcessConfig

• GET /api/v3/cluster/process/{id}

  v3_cluster_process_get(id: str, domain: str = "", filter: str = "")

• GET /api/v3/cluster/process/{id}/config

  v3_cluster_process_get_config(id: str, domain: str = "")

  Model: ProcessConfig

• GET /api/v3/cluster/process/{id}/state

  v3_cluster_process_get_state(id: str, domain: str = "")

  Model: ProcessState

• GET /api/v3/cluster/process/{id}/report

  v3_cluster_process_get_report(id: str, created_at: int = "", exited_at: int = "", domain: str = "")

  Model: ProcessReport

• PUT /api/v3/cluster/process/{id}

  v3_cluster_process_put(id: str, config: ProcessConfig)

  Model: ProcessConfig

• DELETE /api/v3/cluster/process/{id}

  v3_cluster_process_delete(id: str)

• PUT /api/v3/cluster/process/{id}/command

  v3_cluster_process_put_command(id: str, command: ProcessCommandAction)

  Model: ProcessCommandAction

• GET /api/v3/cluster/process/{id}/metadata/{key}

  v3_cluster_process_get_metadata(id: str, key: str, domain: str = "")

• PUT /api/v3/cluster/process/{id}/metadata/{key}

  v3_cluster_process_put_metadata(id: str, domain: str = "", key: str, data: Metadata)

  Model: Metadata

• GET /api/v3/cluster/process/{id}/probe

  v3_cluster_process_get_probe(id: str, domain: str = "")

• POST /api/v3/cluster/process/probe

  v3_cluster_process_post_probe(config: ProcessConfig, coreid: str = "")

  Model: ProcessConfig

• GET /api/v3/config

  v3_config_get()

• PUT /api/v3/config

  v3_config_put(config: Config)

  Model: Config

• GET /api/v3/config/reload

  v3_config_reload()

• GET /api/v3/fs

  v3_fs_get_list()

• PUT /api/v3/fs

  v3_fs_put(source: str, target: str, operation: FilesystemOperationOrder, bandwidth_limit_kbit: int = None)

  Model: FilesystemOperationOrder

• GET /api/v3/fs/{storage}

  v3_fs_get_file_list(storage=str, glob: str = "", size_min: str = "", size_max: str = "", lastmod_start: str = "", lastmod_end: str = "", sort: str = "", order: str = "")

• GET /api/v3/fs/{storage}/{path}

  v3_fs_get_file(storage: str, path: str)

• PUT /api/v3/fs/{storage}/{path}

  v3_fs_put_file(storage: str, path: str, data: bytes)

• DELETE /api/v3/fs/{storage}

  v3_fs_delete_file_list(storage: str, glob: str = "", size_min: str = "", size_max: str = "", lastmod_start: str = "", lastmod_end: str = "")

  Deletes multiple files by glob (requires Core v16.20.0+).

• DELETE /api/v3/fs/{storage}/{path}

  v3_fs_delete_file(storage: str, path: str)

• GET /api/v3/iam/user

  v3_iam_get_user_list()

• POST /api/v3/iam/user

  v3_iam_post_user(config: IamUser)

  Model: IamUser

• GET /api/v3/iam/user/{name}

  v3_iam_get_user(name: str, domain: str = "")

• PUT /api/v3/iam/user/{name}

  v3_iam_put_user(name: str, domain: str = "", config: IamUser)

  Model: IamUser

• PUT /api/v3/iam/user/{name}/policies

  v3_iam_put_user_policy_list(name: str, domain: str = "", config: IamUserPolicyList)

  Model: IamUser

• DELETE /api/v3/iam/user/{name}

  v3_iam_delete_user(name: str)

• GET /api/v3/log

  v3_log_get(format: str = "console")

• GET /api/v3/metadata/{key}

  v3_metadata_get(key: str)

• PUT /api/v3/metadata/{key}

  v3_metadata_put(key: str, data: Metadata)

  Model: Metadata

• GET /api/v3/metrics

  v3_metrics_get()

• POST /api/v3/metrics

  v3_metrics_post(config: Metrics)

  Model: Metrics

• GET /api/v3/process

  v3_process_get_list(domain: str = "", filter: str = "", reference: str = "", id: str = "", idpattern: str = "", refpattern: str = "", domainpattern: str = "")

• POST /api/v3/process

  v3_process_post(config: ProcessConfig)

  Model: ProcessConfig

• GET /api/v3/process/{id}

  v3_process_get(id: str, domain: str = "", filter: str = "")

• PUT /api/v3/process/{id}

  v3_process_put(id: str, domain: str = "", config: ProcessConfig)

  Model: ProcessConfig

• DELETE /api/v3/process/{id}

  v3_process_delete(id: str, domain: str = "")

• PUT /api/v3/process/{id}/command

  v3_process_put_command(id: str, domain: str = "", command: ProcessCommandAction)

  Model: ProcessCommandAction

• GET /api/v3/process/{id}/config

  v3_process_get_config(id: str, domain: str = "")

• GET /api/v3/process/{id}/metadata/{key}

  v3_process_get_metadata(id: str, domain: str = "", key: str)

• PUT /api/v3/process/{id}/metadata/{key}

  v3_process_put_metadata(id: str, domain: str = "", key: str, data: Metadata)

  Model: Metadata

• GET /api/v3/process/{id}/probe

  v3_process_get_probe(id: str, domain: str = "")

• POST /api/v3/process/probe

  v3_process_post_probe(config: ProcessConfig)

  Model: ProcessConfig Probes a process config directly (requires Core v16.20.0+).

• POST /api/v3/process/validate

  v3_process_post_validate(config: ProcessConfig)

  Model: ProcessConfig Validates a process config (requires Core v16.20.0+).

• PUT /api/v3/process/{id}/report

  v3_process_put_report(id: str, report: ProcessReport, domain: str = "")

  Model: ProcessReport

• GET /api/v3/process/{id}/report

  v3_process_get_report_list(id: str, domain: str = "", created_at: str = "", exited_at: str = "")

• GET /api/v3/process/{id}/state

  v3_process_get_state(id: str, domain: str = "")

• GET /api/v3/process/{id}/playout/{input_id}/errorframe/encode

  v3_process_get_playout_input_errorframe_encode(id: str, input_id: str)

• POST /api/v3/process/{id}/playout/{input_id}/errorframe/{input_name}

  v3_process_post_playout_input_errorframe_name(id: str, input_id: str, input_name: str)

• GET /api/v3/process/{id}/playout/{input_id}/keyframe/{input_name}

  v3_process_get_playout_input_keyframe(id: str, input_id: str, input_name: str)

• GET /api/v3/process/{id}/playout/{input_id}/reopen

  v3_process_get_playout_input_reopen(id: str, input_id: str)

• GET /api/v3/process/{id}/playout/{input_id}/status

  v3_process_get_playout_input_status(id: str, input_id: str)

• PUT /api/v3/process/{id}/playout/{input_id}/stream

  v3_process_put_playout_input_stream(id: str, input_id: str)

• GET /api/v3/report/process

  v3_process_get_report(idpattern: str = "", refpattern: str = "", state: str = "", fromdate: int = "", todate: int = "")

• GET /api/v3/rtmp

  v3_rtmp_get()

• GET /api/v3/rtmp/channels

  v3_rtmp_get_channels()

  Model: RtmpChannel — the detailed listing; v3_rtmp_get returns names only

• PUT /api/v3/rtmp/disconnect

  v3_rtmp_put_disconnect(path: str = "")

  Disconnects the sessions of path; * disconnects all. Returns a string.

• GET /api/v3/session

  v3_session_get(collectors: str)

• GET /api/v3/session/active

  v3_session_get_active(collectors: str)

• PUT /api/v3/session/token

  v3_session_put_token(username: str, config: SessionToken)

  Model: SessionToken

• GET /api/v3/skills

  v3_skills_get()

• GET /api/v3/skills/reload

  v3_skills_reload()

• GET /api/v3/srt

  v3_srt_get()

• GET /api/v3/srt/channels

  v3_srt_get_channels()

  Model: SrtChannel — marked EXPERIMENTAL by Core

• PUT /api/v3/srt/disconnect

  v3_srt_put_disconnect()

  Disconnects all SRT sessions. Returns a string.

• GET /api/v3/widget/process/{id}

  v3_widget_get_process(id: str)

• GET /ping

  ping()

Additional options per request:

• retries: int = default of class
• timeout: float = default of class

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")

token = client.login()
print(token)

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

process_list = client.v3_process_get_list()
for process in process_list:
    print(process.id)

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

process_example = {
    "id": "my_proc",
    "reference": "my_ref",
    "input": [
        {
            "address": "testsrc=size=1280x720:rate=25",
            "id": "input_0",
            "options": ["-f", "lavfi", "-re"],
        }
    ],
    "options": ["-loglevel", "info"],
    "output": [
        {
            "address": "-",
            "id": "output_0",
            "options": ["-codec:v", "libx264", "-r", "25", "-f", "null"]
        }
    ]
}

post_process = client.v3_process_post(config=process_example)
print(post_process)

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

get_process = client.v3_process_get(id="my_proc")
print(get_process)

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

process_example = {
    "id": "my_proc",
    "reference": "my_ref",
    "input": [
        {
            "address": "testsrc=size=1280x720:rate=25",
            "id": "input_0",
            "options": ["-f", "lavfi", "-re"],
        }
    ],
    "options": ["-loglevel", "debug"],
    "output": [
        {
            "address": "-",
            "id": "output_0",
            "options": ["-codec:v", "libx264", "-r", "25", "-f", "null"]
        }
    ]
}

put_process = client.v3_process_put(id="testproc", config=process_example)
print(put_process)

from core_client import Client

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

delete_process = client.v3_process_delete(id="testproc")
print(delete_process)

Models are located here:

• core_client/base/models/
• core_client/base/models/v3

from core_client import Client
from core_client.base.models.v3 import ProcessConfig, ProcessConfigIO

client = Client(base_url="http://127.0.0.1:8080", username="admin", password="datarhei")
client.login()

put_process = client.v3_process_put(id="my_proc", config=ProcessConfig(
    id="my_proc",
    reference="my_ref",
    input=[
        ProcessConfigIO(
            address="testsrc2=rate=25:size=1280x720",
            id="input_0",
            options=["-re", "-f", "lavfi"]
        )
    ],
    options=["-loglevel", "error"],
    output=[
        ProcessConfigIO(
            address="-",
            id="output_0",
            options=["-codec:v", "libx264", "-r", "25", "-f", "null"]
        )
    ]
))
print(put_process)

• model.dict() exports model to dict
• model.json() exports model to json

• parse_obj_as(ModelName, obj)
• ModelName.parse_obj(obj)
• ModelName(**obj)

More details and options in the pydantic docs.

raise_for_status() is unused, but the exceptions are still available:

try:
    process = client.v3_process_get_list()
except httpx.HTTPError as exc:
    print(f"Error while requesting {exc.request.url!r}.")

More in the HTTPX docs.

$ git clone datarhei/core-client-python
$ cd core-client-python && \
    pip install -r requirements-dev.txt

$ make test              # unit tests only
$ make test-integration  # starts a fresh Core in Docker, runs integration tests, tears it down
$ make test-all          # both of the above

make test-integration uses container core-client-test on port 8088 by default; override with e.g. make test-integration CORE_PORT=8080 CORE_IMAGE=datarhei/core:latest.

$ pytest

Only tests/unit is collected by default.

$ docker run -d --name core -p 8080:8080 datarhei/core:latest

$ RUN_INTEGRATION_TESTS=1 CORE_URL=http://127.0.0.1:8080 \
    pytest tests/integration

Cluster integration tests additionally need RUN_CLUSTER_TESTS=1 (and two nodes). The suite expects a fresh Core with auth disabled; it enables JWT auth for the run and restores the original state afterwards. Use coverage html to create an html report.

$ docker build --build-arg PYTHON_VERSION=3.11 \
    -f tests/Dockerfile -t core_test .

$ docker run -it --rm \
    -e CORE_URL=http://192.168.1.1:8080 core_test

Notice: 127.0.0.1 is the container itself.

$ ruff check .
$ pre-commit run --all-files

Requires pip install -r requirements-dev.txt.

Changelog

Found a mistake or misconduct? Create a issue or send a pull-request. Suggestions for improvement are welcome.

MIT