datarhei MediaCore Python Client

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
API models
Error handling
Developing & testing
Changelog
Contributing
Licence

Features

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

Install

Latest

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

Specific version

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

{release_tag} like 1.0.0

Usage

Preferred import path

from datarhei.mediacore import MediaCoreClient, AsyncMediaCoreClient

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

Init arguments

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

Sync

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)

Async

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())

API definitions

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.

General

GET /api
```
about()
```
POST /api/graph/query
```
graph_query(query: GraphQuery)
```
Model: GraphQuery

Token refresh (long-running applications)

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().

Events

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

Event streaming (async)

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, or frame=False for raw lines.

Streaming methods (async only):

v3_events_stream(filters=…, frame=True, model=None) → POST /api/v3/events
v3_cluster_events_stream(filters=…) → POST /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())

Notes:

Filter models: LogEventFilter for the log-event streams (/api/v3/events, /api/v3/cluster/events); ProcessEventFilter (filter by type, domain, pid, core_id) for /api/v3/cluster/events/process. Each filter value is a case-insensitive, unanchored regex (e.g. type="progress", type="progress|report"); multiple fields are AND-combined. Raw dict filters are also accepted. An empty filter delivers all events (a firehose — filter tightly).
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.

Cluster

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()
```

Cluster Database

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)
```

Cluster IAM

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()
```

Cluster Node

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)
```

Cluster Process

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 = "")

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

Config

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()
```

Filesystem

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)

IAM

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)
```

Log

GET /api/v3/log
```
v3_log_get(format: str = "console")
```

Metadata

GET /api/v3/metadata/{key}
```
v3_metadata_get(key: str)
```

PUT /api/v3/metadata/{key}

v3_metadata_put(key: str, data: Metadata)

Model: Metadata

Metrics

GET /api/v3/metrics
```
v3_metrics_get()
```
POST /api/v3/metrics
```
v3_metrics_post(config: Metrics)
```
Model: Metrics

Process

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 = "")

Process Playout (commercial extention)

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)

Report

GET /api/v3/report/process

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

RTMP

GET /api/v3/rtmp
```
v3_rtmp_get()
```

Session

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

Skills

GET /api/v3/skills
```
v3_skills_get()
```
GET /api/v3/skills/reload
```
v3_skills_reload()
```

SRT

GET /api/v3/srt
```
v3_srt_get()
```

Widget

GET /api/v3/widget/process/{id}
```
v3_widget_get_process(id: str)
```

Misc

GET /ping
```
ping()
```

Additional options per request:

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

Examples

GET token data

from core_client import Client

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

token = client.login()
print(token)

GET processes

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)

POST 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", "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)

GET 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)

PUT 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)

DELETE 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)

API models

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)

Pydantic model exports:

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

Pydantic parse object as model:

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

More details and options in the pydantic docs.

Error handling

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}.")

Developing & testing

Clone

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

Testing

Quick start with Docker (recommended)

$ 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.

Unit tests (no backend required)

$ pytest

Only tests/unit is collected by default.

Start a Core backend:

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

Integration tests (require a running Core)

$ 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

$ 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.

Code checks

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

Requires pip install -r requirements-dev.txt.

Changelog

Contributing

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

Licence

MIT

Name		Name	Last commit message	Last commit date
Latest commit History 94 Commits
.github/workflows		.github/workflows
core_client		core_client
datarhei		datarhei
tests		tests
.editorconfig		.editorconfig
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
Makefile		Makefile
README.md		README.md
pyproject.toml		pyproject.toml
pytest.ini		pytest.ini
requirements-dev.txt		requirements-dev.txt
requirements.txt		requirements.txt
setup.cfg		setup.cfg
setup.py		setup.py

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

datarhei MediaCore Python Client

Features

Install

Latest

Specific version

Usage

Preferred import path

Init arguments

Sync

Async

API definitions

General

Token refresh (long-running applications)

Events

Event streaming (async)

Cluster

Cluster Database

Cluster IAM

Cluster Node

Cluster Process

Config

Filesystem

IAM

Log

Metadata

Metrics

Process

Process Playout (commercial extention)

Report

RTMP

Session

Skills

SRT

Widget

Misc

Examples

GET token data

GET processes

POST process

GET process

PUT process

DELETE process

API models

Pydantic model exports:

Pydantic parse object as model:

Error handling

Developing & testing

Clone

Testing

Quick start with Docker (recommended)

Unit tests (no backend required)

Start a Core backend:

Integration tests (require a running Core)

Docker

Code checks

Changelog

Contributing

Licence

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages