Memory Service ORM

datarobot[application-utils] ships a lightweight async ORM over the DataRobot Agentic Memory Service. Think of it as a typed, Pydantic v2-style document store that persists sessions (documents) and events (log entries) through a bearer-auth REST API, using your existing DATAROBOT_ENDPOINT and DATAROBOT_API_TOKEN credentials.

Install

pip install "datarobot[application-utils]"

Quick-start

import asyncio
from typing import Annotated

from datarobot.application_utils.persistence import (
    DRConcurrencyField,
    DRDeduplicationKey,
    DREvent,
    DRMemorySpace,
    DRRangeKey,
    DRSession,
    DRMemoryServiceClient,
    SYSTEM_PARTICIPANT,
)


# ── 1. Define your domain models ─────────────────────────────────────────────


class ChatSession(DRSession):
    """A chat session stored in the Memory Service."""

    __description_prefix__ = "chat"  # stable prefix; part of every query

    tenant: Annotated[str, DRRangeKey]  # description segment 1 (range queries)
    topic: Annotated[str, DRRangeKey]  # description segment 2 (range queries)
    chat_id: Annotated[str, DRDeduplicationKey]  # point-lookup / idempotent create
    rev: Annotated[int, DRConcurrencyField]  # mirrors server version integer
    title: str = ""  # plain metadata (payload only)


class ChatMessage(DREvent, session=ChatSession):
    """A single message in a chat session."""

    __event_type__ = "message"

    score: float = 0.0  # extra body field; round-trips through the wire


# ── 2. Use the ORM ────────────────────────────────────────────────────────────


async def demo() -> None:
    async with DRMemoryServiceClient() as client:
        # Create or adopt an existing space (idempotent via deduplication_key)
        space = await DRMemorySpace.post(
            client,
            description="My agent memory",
            deduplication_key="my-agent-space-v1",
        )

        # Create (or adopt) a session
        session = await ChatSession.post(
            space,
            tenant="acme",
            topic="billing",
            chat_id="billing-chat-001",
            title="Billing enquiry",
        )
        # `rev` is never passed in: it mirrors the server-assigned version.
        assert session.rev == 1

        # Append events
        msg = await ChatMessage.post(
            session,
            content="Hello, I need help with my invoice.",
            emitter_type="user",
            emitter_id="aabbccddeeff001122334455",  # 24-hex ObjectId
            score=0.9,
        )
        print(msg.sequence_id, msg.created_at)

        # List recent messages
        recent = await ChatMessage.last(session, n=10)

        # Fetch by range-key prefix (all billing sessions for "acme")
        billing_sessions = await ChatSession.list(space, tenant="acme", topic="billing")

        # Update session metadata (optimistic-concurrency guard via If-Match)
        await session.patch(title="Resolved billing enquiry")

        # Patch an event (guarded by its createdAt token)
        await msg.patch(score=0.5)


asyncio.run(demo())

Environment variables

Variable	Description
`DATAROBOT_ENDPOINT`	Full DataRobot API base URL, e.g., `https://app.datarobot.com/api/v2`.
`DATAROBOT_API_TOKEN`	DataRobot API bearer token.

Both are resolved automatically; pass them as constructor arguments to override.

Core concepts

Memory space (`DRMemorySpace`)

A namespace that owns sessions and events. Create one per-agent deployment or application context. Idempotent via deduplication_key.

space = await DRMemorySpace.post(client, deduplication_key="my-app-v1")
space2 = await DRMemorySpace.get(client, space.id)
spaces = await DRMemorySpace.list(client, deduplication_key="my-app-v1")
await space.patch(description="Updated description")
await space.delete()

Sessions (`DRSession`)

Documents stored in a memory space. Subclass DRSession and annotate fields with ORM markers:

Marker	Wire field	Purpose
`Annotated[T, DRDeduplicationKey]`	`deduplicationKey`	Unique key; idempotent create.
`Annotated[T, DRRangeKey]`	`description` segment	Range / prefix queries.
`Annotated[T, DRConcurrencyField]`	`version` mirror	User-visible version counter.
(plain field)	`metadata`	Arbitrary payload; not queryable.

Declare range-key fields in query order — a list query must specify a contiguous leading prefix (see §Range-key encoding below).

Lifecycle strategies (TTL)

By default, every DRSession subclass sends a single soft_delete lifecycle strategy on creation, triggered by a 2-year TTL (DEFAULT_SESSION_TTL_SECONDS, 63072000 seconds) — the Memory Service’s own maximum for a TTL trigger. Sessions therefore auto-clean unless you override this.

Override __lifecycle_strategies__ to use a shorter TTL (or a different strategy):

class ChatSession(DRSession):
    __description_prefix__ = "chat"
    __lifecycle_strategies__ = [
        {"type": "soft_delete", "trigger": {"ttl": 30 * 86400}},  # 30 days
    ]
    ...

Set it to an empty list to send no lifecycle strategies at all:

class ChatSession(DRSession):
    __lifecycle_strategies__ = []

Lifecycle strategies are sent on create and can be replaced afterwards with session.patch_lifecycle_strategies([...]): the stored list is replaced wholesale (any strategy not resent is deleted), guarded by If-Match optimistic concurrency. Read the current list first from the read-only session.lifecycle_strategies property, which is refreshed from every wire response. Replacement requires memory service 11.13+ (FLEET-7891) — an older service ignores the field, and the method raises DRMemoryServiceError instead of reporting a silent no-op. Changing a strategy type that has already executed for the session is rejected by the service (HTTP 409, surfaced as DRMemoryServiceError — the executor never runs a type twice, so re-reading and retrying cannot help); resending an executed type unchanged is allowed, so read-modify-write keeps working. session.patch(...) never touches strategies. Up to five strategy objects are allowed per session.

Field names id, created_at, version, lifecycle_strategies, and participants are reserved by the DRSession base (read-only properties and the participants field); declaring one on a subclass raises ValueError when the model is first used.

Events (`DREvent`)

An append-only log under a session. Bind to a session type with class MyEvent(DREvent, session=MySession). All plain declared fields map to the body dict.

class ChatMessage(DREvent, session=ChatSession):
    __event_type__ = "message"  # "message" | "tool_output" | "status"
    score: float = 0.0

Declared fields round-trip through any Pydantic-expressible type — nested models, list[Model], Optional[Model], enums and dicts all serialize into the event body (and session metadata) and validate back on read. This is what lets the chat-history layer nest typed ToolCall / Reasoning models inside a single message event.

Range-key encoding

Range keys are encoded in the session description field using a hierarchical path scheme:

description = "//" + esc(prefix) + "/" + esc(k1) + "/" + esc(k2) + "/"

esc(v) percent-encodes % (→ %25) then / (→ %2F), so values can contain arbitrary text including slashes. The leading // and trailing / after every segment create an anchored prefix — a substring-match on the service side is equivalent to a hierarchy prefix query.

Example — two sessions stored under chat/acme:

//chat/acme/billing/    ← tenant=acme, topic=billing
//chat/acme/support/    ← tenant=acme, topic=support

Query list(tenant="acme") sends description=//chat/acme/ which matches both. Query list(tenant="acme", topic="billing") sends description=//chat/acme/billing/ which matches only the first.

A subclass’ .list always sends at least its //<prefix>/ description filter — even with no range-key arguments — so ChatSession.list(space) returns only chat-prefixed sessions, never sessions of another DRSession subclass that happens to share the same space. (Trailing-slash anchoring keeps prefixes disjoint, so //chat/ never matches a //chatx/… or //loc/… session.) This subclass isolation is what lets the chat-history layer keep its Chat sessions and its EntityLocator index sessions (prefix loc) in one space without either bleeding into the other’s .list results.

⚠️ Case-insensitive caveat — the service performs a case-insensitive substring match, so Acme and acme are treated as the same tenant. Values that differ only in case will collide.

Optimistic concurrency

Sessions

Session PATCH sends an If-Match: <version> header. If the server’s version has advanced since you last fetched the session, the service returns HTTP 409 and the ORM raises DRMemoryVersionConflictError. Resolve by re-fetching (DRSession.get) and retrying.

try:
    await session.patch(title="New title")
except DRMemoryVersionConflictError:
    session = await ChatSession.get(space, id=session.id)
    await session.patch(title="New title")

Events

Event PATCH uses createdAt as a query-string concurrency token instead of a header. A stale token yields HTTP 422 and DRMemoryVersionConflictError.

try:
    await msg.patch(content="Corrected text")
except DRMemoryVersionConflictError:
    # Re-list to obtain fresh tokens
    events = await ChatMessage.list(session)
    msg = next(e for e in events if e.sequence_id == msg.sequence_id)
    await msg.patch(content="Corrected text")

Batch operations

# Atomic batch create (up to 200 events)
msgs = await ChatMessage.post_batch(
    session,
    events=[
        {"content": "First", "emitter_type": "agent"},
        {"content": "Second", "emitter_type": "agent", "score": 0.5},
    ],
)

# Atomic batch patch (up to 200 events)
await ChatMessage.patch_batch(
    session,
    updates=[
        (msgs[0], {"score": 0.9}),
        (msgs[1], {"content": "Updated second"}),
    ],
)

Emitter participant check

When emitter_type="user", the emitter’s ObjectId must be in the session’s participants list. The ORM raises DRMemoryBadRequestError early (before the HTTP call) if the emitter is not a participant.

The system sentinel SYSTEM_PARTICIPANT = "000000000000000000000000" is a client-side convention for agent-owned sessions. It is a valid ObjectId accepted by the service.

Error types

Exception	HTTP status	Cause
`DRMemoryBadRequestError`	400	Invalid request (emitter not a participant, etc.)
`DRMemoryNotFoundError`	404	Resource does not exist
`DRMemoryConflictError`	409	Deduplication conflict on create (ORM auto-adopts)
`DRMemoryVersionConflictError`	409 / 422	Stale If-Match or createdAt token
`DRMemoryValidationError`	422	Schema validation error
`DRMemoryRateLimitError`	429	Trial quota / rate limit; `retry_after` carries the `Retry-After` seconds (`None` for the storage cap, which sends no header)
`DRMemoryUnavailableError`	—	No response received: timeout or transport failure (original `httpx` exception as `__cause__`)
`DRMemoryServiceError`	other 4xx/5xx	Unexpected error

Public API

from datarobot.application_utils.persistence import (
    DRMemorySpace,
    DRSession,
    DREvent,
    DRDeduplicationKey,
    DRRangeKey,
    DRConcurrencyField,
    DRMemoryServiceClient,
    SYSTEM_PARTICIPANT,
    DEFAULT_SESSION_TTL_SECONDS,
    DRMemoryServiceError,
    DRMemoryNotFoundError,
    DRMemoryBadRequestError,
    DRMemoryValidationError,
    DRMemoryConflictError,
    DRMemoryVersionConflictError,
    DRMemoryRateLimitError,
    DRMemoryUnavailableError,
)

Client and models

class datarobot.application_utils.persistence.DRMemoryServiceClient

Async HTTP client for the DataRobot Agentic Memory Service.

Resolves DATAROBOT_ENDPOINT and DATAROBOT_API_TOKEN from the environment; constructor arguments take precedence. Owns an httpx.AsyncClient unless one is injected.

The instance itself is lightweight — the resolved base URL plus a headers dict. Applications that act on behalf of many principals (a different API token per end user) are supported by constructing one DRMemoryServiceClient per principal over a single shared http_client: the shared pool carries the connections, each instance carries only an identity.

That isolation holds only while the shared client stays identity-free: construct it with no auth (request() also sends auth=None as a guard, since identity always travels in this instance’s headers) and a cookie jar that stores nothing — httpx.AsyncClient persists Set-Cookie responses in a jar shared by every request through it, which would replay one principal’s cookies on another principal’s requests. The service authenticates by header, not cookies, so refusing cookies loses nothing (see the multi-principal example below).

Parameters:

endpoint (str | None) – DataRobot API endpoint (e.g. https://app.datarobot.com/api/v2). Defaults to the DATAROBOT_ENDPOINT environment variable.
api_token (str | None) – DataRobot API token. Defaults to the DATAROBOT_API_TOKEN environment variable. Multi-principal applications must pass this explicitly — the environment fallback is the application’s own credential, not the requesting user’s.
base_path (str) – Sub-path appended to the endpoint. Defaults to "memory" — the gateway mount path for the Memory Service (/api/v2/memory).
http_client (httpx.AsyncClient | None) – Injected async client. When supplied, the DRMemoryServiceClient will not close it on aclose() — the caller owns its lifetime. This is a supported production pattern (share one connection pool across many per-principal client instances) as well as the hook for testing with respx. A client shared across principals must be identity-free: no auth, and a non-storing cookie jar (see Examples). When omitted, the client creates and owns a pool of its own.
timeout (float) – Default per-request timeout in seconds. Ignored when http_client is injected — configure the timeout on the injected client instead.

Examples

One client, one credential (scripts, single-principal apps):

import asyncio
from datarobot.application_utils.persistence import (
    DRMemorySpace,
    DRMemoryServiceClient,
)

async def main() -> None:
    async with DRMemoryServiceClient() as client:
        space = await DRMemorySpace.post(client, description="my-space")
        print(space.id)

asyncio.run(main())

One shared pool, one thin client per principal (multi-user web apps):

import http.cookiejar

import httpx

# App startup.  The shared pool must stay identity-free: no auth
# (identity belongs on each DRMemoryServiceClient, not the
# transport), and a cookie jar that stores nothing — httpx would
# otherwise replay one principal's Set-Cookie on another's requests.
shared_http = httpx.AsyncClient(
    timeout=30.0,
    cookies=http.cookiejar.CookieJar(
        policy=http.cookiejar.DefaultCookiePolicy(allowed_domains=[])
    ),
)

def client_for(user_token: str) -> DRMemoryServiceClient:
    # Cheap: per-request construction, no new connections.
    return DRMemoryServiceClient(
        api_token=user_token,
        http_client=shared_http,
    )

async def shutdown() -> None:
    await shared_http.aclose()  # close the pool once, at app shutdown

property base_url: str: The resolved Memory Service base URL.

async request(method, path, *, params=None, json=None, extra_headers=None)

Send a request and return the response, mapping errors to typed exceptions.

Parameters:

method (str) – HTTP method ("GET", "POST", "PATCH", "DELETE").
path (str) – Path relative to base_url. Must end with "/".
params (dict | None) – Query parameters.
json (Any) – JSON-serializable request body.
extra_headers (dict | None) – Additional headers (e.g. {"If-Match": "3"}).

Returns:

On 2xx status codes.

Return type:

httpx.Response

Raises:

DRMemoryBadRequestError – HTTP 400 — bad request (e.g. emitter not a participant).
DRMemoryNotFoundError – HTTP 404 — resource not found.
DRMemoryConflictError – HTTP 409 — deduplication conflict on create.
DRMemoryVersionConflictError – HTTP 409 — stale If-Match on session patch(), or HTTP 422 with the event-version detail.
DRMemoryValidationError – HTTP 422 — schema validation error.
DRMemoryRateLimitError – HTTP 429 — quota or rate limit exceeded; carries retry_after seconds parsed from the Retry-After response header.
DRMemoryUnavailableError – No response received — request timeout or transport failure (connection refused, DNS, TLS). The original httpx exception is preserved as __cause__.
DRMemoryServiceError – Any other 4xx/5xx error.

async aclose()

Close the underlying HTTP client (only if owned by this instance).

Return type:: None

class datarobot.application_utils.persistence.DRMemorySpace

Represents a DataRobot Agentic Memory Service memory space.

Acts as the container for sessions; every session and event call is scoped to a space. Construct via the class methods post() or get() rather than directly instantiating.

Variables:

description (str | None) – Human-readable description of the space.
deduplication_key (str | None) – Unique client-assigned key; enables idempotent space creation.
llm_model_name (str | None) – LLM model name for extract_memories lifecycle strategies.
llm_base_url (str | None) – LLM base URL override.
custom_instructions (str | None) – Custom instructions passed to the LLM when extracting memories.
created_at (str) – ISO-8601 creation timestamp (server-assigned).

property id: str: Server-assigned memory space UUID.

property user_id: str: Owner user ID.

property tenant_id: str: Tenant UUID.

async classmethod post(client, *, description=None, deduplication_key=None, llm_model_name=None, llm_base_url=None, custom_instructions=None)

Create a new memory space, or adopt an existing one on a deduplication conflict.

If a deduplication_key is supplied and a space with that key already exists, the existing space is fetched and returned (409 → adopt).

Parameters:

client (DRMemoryServiceClient) – Transport client.
description (str | None) – Human-readable description (max 1000 chars).
deduplication_key (str | None) – Unique client key for idempotent creation (1–72 chars).
llm_model_name (str | None) – LLM model name for extract_memories strategies.
llm_base_url (str | None) – LLM base URL override.
custom_instructions (str | None) – Custom LLM instructions (max 10 000 chars).

Returns:

The newly created or adopted memory space.

Return type:

DRMemorySpace

async classmethod get(client, space_id)

Fetch a memory space by its server-assigned ID.

Parameters:

client (DRMemoryServiceClient) – Transport client.
space_id (str) – UUID of the memory space.

Return type:

DRMemorySpace

Raises:

DRMemoryNotFoundError – If no space with the given ID exists (or it belongs to another user).

async classmethod list(client, *, deduplication_key=None, offset=0, limit=100)

List memory spaces visible to the authenticated user.

Parameters:

client (DRMemoryServiceClient) – Transport client.
deduplication_key (str | None) – Exact-match filter on deduplicationKey.
offset (int) – Number of spaces to skip (for pagination).
limit (int) – Maximum number of spaces to return (1–100).

Return type:

list[DRMemorySpace]

async patch(*, description=None, llm_model_name=None, llm_base_url=None, custom_instructions=None)

Update this memory space in place.

Only the supplied keyword arguments are changed; omitted fields keep their current values on the server.

Parameters:

description (str | None) – New description.
llm_model_name (str | None) – New LLM model name.
llm_base_url (str | None) – New LLM base URL.
custom_instructions (str | None) – New custom instructions.

Raises:

ValueError – When no field is supplied, rather than sending a no-op request.

Return type:

None

async delete()

Soft-delete this memory space.

After deletion the space is no longer accessible via get or list.

Return type:: None

class datarobot.application_utils.persistence.DRSession

Abstract base class for Memory Service ORM session models.

Do not instantiate directly; subclass and declare fields with ORM markers.

Class variables

__description_prefix__str: Prefix injected at the start of the encoded description. Defaults to the subclass name. Keep it short and stable; it is part of every stored description and every list-query filter.
__lifecycle_strategies__list[dict[str, Any]]: Lifecycle strategy objects sent on session creation. Defaults to a single soft_delete strategy with a DEFAULT_SESSION_TTL_SECONDS (2 year) TTL trigger, so sessions auto-clean unless a subclass overrides this. Override with a different strategy list to change the TTL/strategy, or set to [] to send no lifecycle strategies at all.

Read-only properties

idstr: Server-assigned session UUID.
created_atstr: ISO-8601 creation timestamp.
versionint: Server-assigned version integer (optimistic-concurrency token).

property id: str: Server-assigned session UUID (read-only).

property created_at: str: ISO-8601 creation timestamp (read-only).

property version: int: Current server version integer (read-only; updated on every patch).

property lifecycle_strategies: list[dict[str, Any]]

The stored lifecycle strategies, as last seen on the wire (read-only).

Refreshed from every wire response (get/post/list/ patch/patch_lifecycle_strategies). Use it to read-modify-write with patch_lifecycle_strategies(), which replaces the stored list wholesale — any strategy not resent is deleted. Returns a copy; mutating it does not change the session.

async classmethod post(space, **kwargs)

Create a session, or adopt the existing one on a deduplication conflict.

Parameters:

space (DRMemorySpace) – Memory space to create the session in.
**kwargs (Any) – Session field values. Pass participants=["<objectid>"] to scope to a user; omit to use the system sentinel. A DRConcurrencyField may not be passed — it mirrors the server-assigned version and is populated from the response.

Returns:

The newly created or adopted session.

Return type:

DRSession

Raises:

ValueError – On undeclared kwargs, or when a DRConcurrencyField is supplied.

Examples

session = await ChatSession.post(
    space,
    tenant="acme",
    topic="billing",
    chat_id="chat-001",
    title="Billing enquiry",
)

async classmethod get(space, id=None, **kwargs)

Fetch a session by its server-assigned id or by deduplication_key.

Exactly one of id= or the subclass’s DRDeduplicationKey field name must be supplied.

Parameters:

space (DRMemorySpace) – Memory space containing the session.
id (str | None) – Server-assigned session UUID.
**kwargs (Any) – Pass the DRDeduplicationKey field name as a keyword argument for an exact-match point lookup (e.g. chat_id="billing-chat-001").

Return type:

DRSession

Raises:

DRMemoryNotFoundError – If no matching session is found.
ValueError – If neither id nor a dedup key is supplied, or the subclass has no DRDeduplicationKey field and a keyword arg is provided.

Examples

# By server id
session = await ChatSession.get(space, id="uuid-string")

# By dedup key
session = await ChatSession.get(space, chat_id="billing-chat-001")

async classmethod list(space, *, participant=None, **kwargs)

List sessions matching a range-key prefix and/or participant filter.

The query is always scoped to this subclass by its __description_prefix__ (so list() never returns sessions of a different subclass sharing the space); range-key kwargs narrow it further.

Range-key kwargs must form a contiguous leading prefix of the declared DRRangeKey fields (e.g. for fields [tenant, topic] you may filter on tenant= alone or on tenant= + topic=, but not on topic= alone).

Parameters:

space (DRMemorySpace) – Memory space to query.
participant (str | None) – Filter to sessions that include this ObjectId in participants.
**kwargs (Any) – Leading range-key field values for a prefix query.

Returns:

All matching sessions (auto-paginated).

Return type:

list[DRSession]

Raises:

ValueError – If range-key kwargs are not a contiguous leading prefix.

Examples

# All sessions for tenant "acme"
sessions = await ChatSession.list(space, tenant="acme")

# Scoped to a user
sessions = await ChatSession.list(space, participant=user_oid)

# Combined: user + range prefix
sessions = await ChatSession.list(
    space, participant=user_oid, tenant="acme", topic="billing"
)

async patch(**kwargs)

Update this session in place.

Pass any combination of metadata fields and/or DRRangeKey fields. participants and DRDeduplicationKey fields cannot be changed.

Parameters:: **kwargs (Any) – Fields to update.
Raises:: DRMemoryVersionConflictError – If the session was updated concurrently (stale If-Match).
Return type:: None

Examples

await session.patch(title="New title")
await session.patch(topic="support", title="Re: billing")

async patch_lifecycle_strategies(strategies)

Replace this session’s lifecycle strategies in place.

The service replaces the stored list wholesale with strategies — the same wire shape __lifecycle_strategies__ declares at create time, e.g. [{"type": "soft_delete", "trigger": {"ttl": 63_072_000}}]. A strategy’s retention deadline is recomputed from its stored anchor plus the new trigger duration.

Requires a non-empty list: the service treats an omitted field as “no change”, and this method never sends an empty replacement. Read the current list from lifecycle_strategies first — replacement is wholesale, so any strategy not resent is deleted.

Parameters:

strategies (list[dict[str, Any]]) – Replacement lifecycle strategies, in wire shape.

Raises:

DRMemoryVersionConflictError – If the session was updated concurrently (stale If-Match).
DRMemoryValidationError – When the service rejects a strategy — e.g. a {"never": true} trigger in an environment where the never-expire gate is off (unless that strategy type already stores one).
DRMemoryServiceError – With status_code=409 when the change touches a strategy type that has already executed for this session — the executor never runs a type twice, so the service rejects the change as one that can never take effect. Not retryable; resending an executed type unchanged (as read-modify-write does) is allowed. Also raised when the server accepted the PATCH but ignored lifecycleStrategies — services older than 11.13 (FLEET-7891) drop the unknown field and answer 200 with an unchanged version, which would otherwise be a silent no-op; session retention is unchanged in that case.

Examples

await session.patch_lifecycle_strategies(
    [{"type": "soft_delete", "trigger": {"idle": 63_072_000}}]
)

Return type:: None

async delete()

Soft-delete this session.

After deletion the session is no longer returned by get or list.

Return type:: None

class datarobot.application_utils.persistence.DREvent

Abstract base class for Memory Service ORM event models.

Do not instantiate directly; subclass with session=<SessionClass>.

Parameters:

content (str) – Event text (1–100 000 characters).
emitter_type (Literal[``”user”, ``"agent"]) – Who produced the event.
emitter_id (str | None) – ObjectId of the emitting user. Required when emitter_type="user" and must be a member of the session’s participants.
properties (Read-only)
--------------------
sequence_id (int) – Server-assigned monotonic integer address (−1 before the event is posted).
created_at (str) – ISO-8601 timestamp; also serves as the concurrency token for patch().

property sequence_id: int: Server-assigned event address (read-only; −1 before posting).

property created_at: str: ISO-8601 creation timestamp (read-only; also the concurrency token).

async classmethod post(session, *, content, emitter_type, emitter_id=None, **kwargs)

Append a single event to the session.

Parameters:

session (DRSession) – Session to append the event to.
content (str) – Event text (1–100 000 characters).
emitter_type (Literal[``”user”, ``"agent"]) – Who produced the event.
emitter_id (str | None) – ObjectId of the emitter. Required when emitter_type="user" and must be in session.participants.
**kwargs (Any) – Any declared body fields (e.g. score=0.9).

Return type:

DREvent

async classmethod post_batch(session, events)

Atomically append up to 200 events to the session.

All events are appended in list order. If any event fails validation the entire batch is rolled back.

Parameters:

session (DRSession) – Session to append the events to.
events (list[dict[str, Any]]) – Each dict contains the same keyword arguments as post() (content, emitter_type, optionally emitter_id and body fields).

Return type:

list[DREvent]

Raises:

ValueError – If the batch exceeds 200 events.

Examples

msgs = await ChatMessage.post_batch(
    session=my_session,
    events=[
        {"content": "Hi", "emitter_type": "user", "emitter_id": oid},
        {"content": "Hello", "emitter_type": "agent"},
    ],
)

async classmethod list(session, *, type=None, offset=0, limit=100)

List events under a session, optionally filtered by type.

Parameters:

session (DRSession) – Session to query.
type (str | None) – Event type filter: "message", "tool_output", or "status". None returns all types.
offset (int) – Number of events to skip.
limit (int) – Maximum number of events to return (1–100).

Return type:

list[DREvent]

async classmethod last(session, *, n, type=None)

Return the last n events in chronological order.

lastN and offset are mutually exclusive on the service API; this method never sends offset.

Parameters:

session (DRSession) – Session to query.
n (int) – Number of tail events to return (1–100).
type (str | None) – Optional event-type filter.

Return type:

list[DREvent]

async patch(*, content=None, emitter_type=None, emitter_id=None, **kwargs)

Update this event in place, guarded by its created_at token.

At least one field must be supplied.

Parameters:

content (str | None) – New event text.
emitter_type (Literal[``”user”, ``"agent"] | None) – New emitter type.
emitter_id (str | None) – New emitter ObjectId.
**kwargs (Any) – Any declared body fields to update.

Raises:

DRMemoryVersionConflictError – If the event was updated concurrently (stale createdAt token).

Return type:

None

async delete()

Soft-delete this event.

After deletion the event is no longer returned by list or last.

Return type:: None

async classmethod patch_batch(session, updates)

Atomically update up to 200 events, each guarded by its created_at token.

Parameters:

session (DRSession) – Session containing the events. All events must belong to this session.
updates (list[tuple[DREvent, dict[str, Any]]]) – Each tuple is (event_instance, kwargs). The kwargs accept the same arguments as patch().

Returns:

Updated event instances (in the order of the input list).

Return type:

list[DREvent]

Raises:

ValueError – If the batch exceeds 200 updates or no fields are provided for an item.
DRMemoryVersionConflictError – If any event was updated concurrently.

Examples

await ChatMessage.patch_batch(
    session=my_session,
    updates=[
        (event_a, {"score": 0.9}),
        (event_b, {"content": "Updated text"}),
    ],
)

Field markers and constants

class datarobot.application_utils.persistence.DRDeduplicationKey

Marker: field maps to the session deduplicationKey (point-lookup primary key).

At most one field per DRSession subclass may carry this marker. Enables MySession.get(space, my_key="value") exact-match point lookups and idempotent session creation (409 → adopt the existing session).

class datarobot.application_utils.persistence.DRRangeKey

Marker: field maps to a segment of the session description (range/prefix queries).

Segments are appended to description in declaration order, encoded with the //prefix/seg1/seg2/ scheme. Values must be non-empty strings.

Known limitation

The Memory Service description filter is case-insensitive, so values differing only in case (e.g. Foo vs foo) will collide when querying.

class datarobot.application_utils.persistence.DRConcurrencyField

Marker: field is kept in sync with the server version integer.

Enables user code to inspect the current optimistic-concurrency version without calling .version directly. At most one field per DRSession subclass may carry this marker.

Whether or not this marker is present, the ORM always tracks the server version internally for If-Match concurrency control on patch().

datarobot.application_utils.persistence.markers.SYSTEM_PARTICIPANT: str = '000000000000000000000000': Sentinel ObjectId for sessions that belong to no specific user. This is a client-side convention; the Memory Service treats it as any other valid 24-hex ObjectId.

datarobot.application_utils.persistence.markers.DEFAULT_SESSION_TTL_SECONDS: int = 63072000: Default session TTL in seconds (2 years). Also the Memory Service maximum for a TTL trigger; used to build the default soft_delete lifecycle strategy.

Errors

exception datarobot.application_utils.persistence.DRMemoryServiceError

Base exception for all Memory Service ORM errors.

Parameters:

detail (str) – Human-readable error detail.
status_code (int | None) – HTTP status code, if applicable.
payload (dict | None) – Raw response body, if available.

exception datarobot.application_utils.persistence.DRMemoryNotFoundError: Raised when the requested resource does not exist (HTTP 404).

exception datarobot.application_utils.persistence.DRMemoryBadRequestError

Raised on client-side validation errors from the service (HTTP 400).

Common causes: an event emitter is not a session participant, an invalid ObjectId is supplied for a participant filter.

exception datarobot.application_utils.persistence.DRMemoryValidationError

Raised on schema validation errors from the service (HTTP 422).

This typically means the request body or query parameters failed the service’s Pydantic validation (e.g. a field is too long, a required field is missing, or mutually exclusive parameters are both supplied).

exception datarobot.application_utils.persistence.DRMemoryConflictError

Raised on a deduplication conflict when creating a session or DRMemorySpace (HTTP 409).

The ORM automatically adopts the existing resource (by fetching it via existing_id) rather than propagating this exception to callers in the normal post() flow.

Parameters:

existing_id (str | None) – Server-assigned ID of the existing resource.
location (str | None) – URL from the service Location header pointing to the existing resource.

exception datarobot.application_utils.persistence.DRMemoryVersionConflictError

Raised on an optimistic-concurrency failure.

Surfaces as HTTP 409 for session patch() (stale If-Match header) and as HTTP 422 for event patch() (stale createdAt token).

Resolution: re-read the resource to get the current version, then retry.

exception datarobot.application_utils.persistence.DRMemoryRateLimitError

Raised when the service rejects a request due to quota or rate limits (HTTP 429).

The Memory Service enforces per-tenant trial quotas: monthly read/write counts answer 429 with a Retry-After header (seconds until the window resets), while the storage cap answers 429 without Retry-After — storage is a level, not a windowed quota, so freeing data (or upgrading) is the remedy rather than waiting.

Resolution: when retry_after is set, wait that many seconds before retrying or propagate the value to your own HTTP response so the caller can back off correctly. When it is None, retrying later will not help by itself — inspect detail for the limit that was hit.

Parameters:: retry_after (int | None) – Whole seconds to wait before retrying, parsed from the Retry-After response header (supports both delta-seconds and HTTP-date forms; integer so it can be propagated verbatim into another Retry-After header). None when the service did not send the header (e.g. the trial storage-cap 429).

exception datarobot.application_utils.persistence.DRMemoryUnavailableError

Raised when no HTTP response was received from the service.

Covers request timeouts and transport failures (connection refused, DNS resolution, TLS errors, protocol violations). The original httpx exception is preserved as __cause__.

status_code is always None: the failure happened before a status code existed. Catching DRMemoryServiceError therefore covers both service-side errors and an unreachable service, without importing httpx at call sites.

Running integration tests

Integration tests are skipped by default. To run them against a live endpoint:

export DATAROBOT_ENDPOINT="https://app.datarobot.com/api/v2"
export DATAROBOT_API_TOKEN="<your-token>"

pytest tests/application_utils/persistence/acceptance -m integration -vv

Memory Service ORM

Install

Quick-start

Environment variables

Core concepts

Memory space (DRMemorySpace)

Sessions (DRSession)

Lifecycle strategies (TTL)

Events (DREvent)

Range-key encoding

Optimistic concurrency

Sessions

Events

Batch operations

Emitter participant check

Error types

Public API

Client and models

Class variables

Read-only properties

Field markers and constants

Known limitation

Errors

Running integration tests

Memory space (`DRMemorySpace`)

Sessions (`DRSession`)

Events (`DREvent`)