harnest_postgres

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`PostgresMemoryStore`	Class
`PostgresStore`	Class
`PostgresTaskStore`	Class

PostgresMemoryStore

Persist deliberate memories in a Harnest-owned table in your database.

PostgresMemoryStore(
    dsn: str,
    *,
    pool_options: Optional[Mapping[str, Any]] = None,
    setup_schema: bool = True,
) -> None

Parameters and example

Parameter	Type	Default
`dsn`	`str`	Required
`pool_options`	`Optional[Mapping[str, Any]]`	`None`
`setup_schema`	`bool`	`True`

from harnest_postgres import PostgresMemoryStore

store = PostgresMemoryStore(dsn=...)

PostgresMemoryStore.delete_all

async PostgresMemoryStore.delete_all(scope: harnest.memory.MemoryScope) -> int

Erase one explicit namespace through trusted application code only.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
result = await store.delete_all(scope=...)

PostgresMemoryStore.delete_user

async PostgresMemoryStore.delete_user(application_id: str, user_id: str) -> int

Erase an explicitly authorized owner without registering any endpoint.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
result = await store.delete_user(application_id=..., user_id=...)

PostgresMemoryStore.purge_expired

async PostgresMemoryStore.purge_expired(
    scope: harnest.memory.MemoryScope,
    *,
    limit: int = 100,
    after: str | None = None,
) -> harnest.memory.MemoryCleanupPage

Sweep a bounded key page without exposing another user’s collection.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`limit`	`int`	`100`
`after`	`str \| None`	`None`

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
page = await store.purge_expired(scope=...)

PostgresMemoryStore.put

async PostgresMemoryStore.put(
    record: harnest.memory.MemoryRecord,
    *,
    expected_revision: str | None = None,
) -> harnest.memory.MemoryRecord

Validate before atomically persisting a deliberate memory write.

Parameters and example

Parameter	Type	Default
`record`	`harnest.memory.MemoryRecord`	Required
`expected_revision`	`str \| None`	`None`

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
record = await store.put(record=...)

PostgresMemoryStore.get

async PostgresMemoryStore.get(scope: harnest.memory.MemoryScope, key: str) -> harnest.memory.MemoryRecord | None

Read only an explicitly scoped live record.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`key`	`str`	Required

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
record = await store.get(scope=..., key=...)

PostgresMemoryStore.list

async PostgresMemoryStore.list(
    scope: harnest.memory.MemoryScope,
    *,
    limit: int = 20,
    after: str | None = None,
) -> harnest.memory.MemoryPage

Page records without loading an unbounded provider collection.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`limit`	`int`	`20`
`after`	`str \| None`	`None`

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
page = await store.list(scope=...)

PostgresMemoryStore.search

async PostgresMemoryStore.search(
    scope: harnest.memory.MemoryScope,
    query: str,
    *,
    limit: int = 20,
    after: str | None = None,
) -> harnest.memory.MemoryPage

Retrieve explicit literal text matches without model or embedding calls.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`query`	`str`	Required
`limit`	`int`	`20`
`after`	`str \| None`	`None`

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
page = await store.search(scope=..., query=...)

PostgresMemoryStore.delete

async PostgresMemoryStore.delete(
    scope: harnest.memory.MemoryScope,
    key: str,
    *,
    expected_revision: str | None = None,
) -> bool

Remove memory and its indexes with an optional atomic revision fence.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`key`	`str`	Required
`expected_revision`	`str \| None`	`None`

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
result = await store.delete(scope=..., key=...)

PostgresMemoryStore.start

async PostgresMemoryStore.start() -> None

Open an owned pool and serialize schema setup across replicas.

Example

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
await store.start()

PostgresMemoryStore.close

async PostgresMemoryStore.close() -> None

Close only this provider’s connection pool.

Example

from harnest_postgres import PostgresMemoryStore

store: PostgresMemoryStore
await store.close()

PostgresStore

Store sessions, checkpoints, continuations, tasks and cron together.

PostgresStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    lease_pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Parameters and example

Parameter	Type	Default
`dsn`	`str`	Required
`pool_options`	`Mapping[str, Any] \| None`	`None`
`lease_pool_options`	`Mapping[str, Any] \| None`	`None`
`setup_schema`	`bool`	`True`

from harnest_postgres import PostgresStore

store = PostgresStore(dsn=...)

PostgresStore.start

async PostgresStore.start() -> None

Provision both storage contracts using the combined store’s pool.

Example

from harnest_postgres import PostgresStore

store: PostgresStore
await store.start()

PostgresStore.close

async PostgresStore.close() -> None

Close only pools created and owned by this store.

Example

from harnest_postgres import PostgresStore

store: PostgresStore
await store.close()

PostgresStore.create_cron

async PostgresStore.create_cron(record: CronRecord) -> CronRecord

Create a schedule once without reactivating an existing cancelled key.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.create_cron(record=...)

PostgresStore.get_cron

async PostgresStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None

Read a schedule only within its application and owner scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)

PostgresStore.list_crons

async PostgresStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Apply owner filtering and keyset pagination in the database.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list_crons(application_id=..., user_id=...)

PostgresStore.update_cron

async PostgresStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord

Replace one schedule conditionally, preserving its immutable identity.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required
`expected_revision`	`int`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.update_cron(record=..., expected_revision=...)

PostgresStore.delete_cron

async PostgresStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool

Delete a schedule using the same owner predicate as reads.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)

PostgresStore.list_due_crons

async PostgresStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Page due schedules in database order without loading inactive rows.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`now`	`float`	Required
`after`	`tuple[float, str] \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list_due_crons(application_id=..., now=...)

PostgresStore.commit_cron_occurrence

async PostgresStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None

Atomically enqueue an occurrence and advance its schedule revision.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required
`expected_revision`	`int`	Required
`due_at`	`float`	Required
`next_run_at`	`float`	Required
`task`	`TaskRecord`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.commit_cron_occurrence(
    application_id=...,
    user_id=...,
    schedule_id=...,
    expected_revision=...,
    due_at=...,
    next_run_at=...,
    task=...,
)

PostgresStore.enqueue_task

async PostgresStore.enqueue_task(record: TaskRecord) -> TaskRecord

Persist an immutable task identity, returning exact idempotent retries.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.enqueue_task(record=...)

PostgresStore.get_task

async PostgresStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Read a scoped job; omitted user scope is reserved for trusted workers.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_task(application_id=..., job_id=...)

PostgresStore.claim_tasks

async PostgresStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Claim disjoint batches and retire exhausted attempts after crashes.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

PostgresStore.renew_task_lease

async PostgresStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Renew only an unexpired attempt still owned by the calling worker.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

PostgresStore.finish_task

async PostgresStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Fence completion and retries; terminal transitions erase private input.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

PostgresStore.cancel_task

async PostgresStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Cancel an owned nonterminal job and revoke any running attempt.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

PostgresStore.list_task_metadata

async PostgresStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Push owner, cursor, projection, ordering, and limit into one database query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list_task_metadata(application_id=..., user_id=...)

PostgresStore.list_cron_metadata

async PostgresStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Read a bounded metadata projection with owner filtering in one SQL query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list_cron_metadata(application_id=..., user_id=...)

PostgresStore.as_langgraph_checkpointer

PostgresStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest_postgres import PostgresStore

store: PostgresStore
result = store.as_langgraph_checkpointer()

PostgresStore.create

async PostgresStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one tenant-scoped session without overwriting conflicts.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.create(session_id=..., user_id=..., state=...)

PostgresStore.get

async PostgresStore.get(*, session_id: str, user_id: str) -> SessionRecord | None

Read one tenant-scoped session from PostgreSQL.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get(session_id=..., user_id=...)

PostgresStore.list

async PostgresStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

Apply tenant filtering and optional keyset pagination in PostgreSQL.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list(user_id=...)

PostgresStore.put_a2a_task

async PostgresStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Upsert an owner-scoped protobuf snapshot with immutable context.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.put_a2a_task(record=...)

PostgresStore.get_a2a_task

async PostgresStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one A2A task using its complete application and owner scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)

PostgresStore.list_a2a_tasks

async PostgresStore.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Apply A2A filters, ordering, and keyset pagination in PostgreSQL.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest_postgres import PostgresStore

store: PostgresStore
page = await store.list_a2a_tasks(application_id=..., user_id=...)

PostgresStore.delete_a2a_task

async PostgresStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete one exact A2A owner record without revealing foreign tasks.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)

PostgresStore.update

async PostgresStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Atomically merge a state delta in PostgreSQL.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.update(session_id=..., user_id=..., state_delta=...)

PostgresStore.delete

async PostgresStore.delete(*, session_id: str, user_id: str) -> bool

Delete one tenant-scoped session and report whether it existed.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.delete(session_id=..., user_id=...)

PostgresStore.acquire

PostgresStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]

Hold a cross-replica advisory lock for one session execution.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
async with store.acquire(session_id=..., user_id=...):
    ...

PostgresStore.begin_run

async PostgresStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Create an idempotent run while allowing one active run per session.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

PostgresStore.get_run

async PostgresStore.get_run(*, scope: RunScope) -> RunRecord | None

Read one PostgreSQL checkpoint run through its ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_run(scope=...)

PostgresStore.get_checkpoint

async PostgresStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None

Load an exact or newest checkpoint entirely in PostgreSQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_checkpoint(scope=...)

PostgresStore.list_checkpoints

PostgresStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Push checkpoint filtering, ordering, cursor, and limit into SQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest_postgres import PostgresStore

store: PostgresStore
async for item in store.list_checkpoints(scope=...):
    ...

PostgresStore.put

async PostgresStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Allocate a serialized revision and write only on matching CAS state.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

PostgresStore.put_writes

async PostgresStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None

Insert pending writes as one set and ignore existing identities.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

PostgresStore.get_writes

async PostgresStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read pending writes for one owned PostgreSQL checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.get_writes(scope=..., checkpoint_id=...)

PostgresStore.get_writes_batch

async PostgresStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Fetch pending writes for several checkpoints in one query.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

PostgresStore.suspend_continuation

async PostgresStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord

Atomically persist a provider wait and mark its owned run waiting.

Parameters and example

Parameter	Type	Default
`record`	`ContinuationRecord`	Required
`external_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.suspend_continuation(record=..., external_id=...)

PostgresStore.get_continuation

async PostgresStore.get_continuation(*, scope: RunScope, continuation_id: str) -> ContinuationRecord | None

Load one continuation with its complete ownership predicate in SQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`continuation_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.get_continuation(scope=..., continuation_id=...)

PostgresStore.get_provider_continuation

async PostgresStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> ProviderPendingContinuation | None

Load one exact provider wait and its private external identity.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
continuation = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

PostgresStore.get_continuation_by_external_id

async PostgresStore.get_continuation_by_external_id(
    *,
    application_id: str,
    provider: str,
    external_id: str,
) -> ProviderPendingContinuation | None

Use the unique provider key so replica callbacks never scan waits.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`external_id`	`str`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
continuation = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

PostgresStore.list_pending_continuations

async PostgresStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[ProviderPendingContinuation]

Push provider reconciliation filtering and pagination into PostgreSQL.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresStore

store: PostgresStore
result = await store.list_pending_continuations(application_id=..., provider=...)

PostgresStore.resolve_continuation

async PostgresStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: ContinuationFailure | None = None,
) -> ContinuationRecord

Commit one validated provider outcome with an indexed SQL CAS.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`external_id`	`str`	Required
`schema_id`	`str`	Required
`result`	`Any`	`None`
`failure`	`ContinuationFailure \| None`	`None`

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

PostgresStore.claim_continuation

async PostgresStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Claim an outcome and resume its run in one PostgreSQL statement.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

PostgresStore.cancel_continuation

async PostgresStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: ContinuationFailure,
) -> ContinuationRecord

Fail a pending continuation and cancel its waiting run atomically.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required
`failure`	`ContinuationFailure`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

PostgresStore.arm_continuation

async PostgresStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Arm one wait with a SQL CAS after native checkpoint persistence.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

PostgresStore.transition

async PostgresStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply a validated status transition with one SQL compare-and-swap.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest_postgres import PostgresStore

store: PostgresStore
record = await store.transition(scope=..., expected_status=..., status=...)

PostgresStore.delete_run

async PostgresStore.delete_run(*, scope: RunScope) -> None

Delete a run and cascade its private checkpoint data.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest_postgres import PostgresStore

store: PostgresStore
await store.delete_run(scope=...)

PostgresTaskStore

Persist jobs and schedules with transactional claims and fenced leases. _pool supports composing this provider with the combined session store; externally supplied pools remain owned by their caller.

PostgresTaskStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None

Parameters and example

Parameter	Type	Default
`dsn`	`str`	Required
`pool_options`	`Mapping[str, Any] \| None`	`None`
`setup_schema`	`bool`	`True`

from harnest_postgres import PostgresTaskStore

store = PostgresTaskStore(dsn=...)

PostgresTaskStore.start

async PostgresTaskStore.start() -> None

Open one pool and install or validate the durable schema.

Example

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
await store.start()

PostgresTaskStore.close

async PostgresTaskStore.close() -> None

Close only the pool opened by this instance.

Example

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
await store.close()

PostgresTaskStore.create_cron

async PostgresTaskStore.create_cron(record: CronRecord) -> CronRecord

Create a schedule once without reactivating an existing cancelled key.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.create_cron(record=...)

PostgresTaskStore.get_cron

async PostgresTaskStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None

Read a schedule only within its application and owner scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)

PostgresTaskStore.list_crons

async PostgresTaskStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Apply owner filtering and keyset pagination in the database.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_crons(application_id=..., user_id=...)

PostgresTaskStore.update_cron

async PostgresTaskStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord

Replace one schedule conditionally, preserving its immutable identity.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required
`expected_revision`	`int`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.update_cron(record=..., expected_revision=...)

PostgresTaskStore.delete_cron

async PostgresTaskStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool

Delete a schedule using the same owner predicate as reads.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)

PostgresTaskStore.list_due_crons

async PostgresTaskStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Page due schedules in database order without loading inactive rows.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`now`	`float`	Required
`after`	`tuple[float, str] \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_due_crons(application_id=..., now=...)

PostgresTaskStore.commit_cron_occurrence

async PostgresTaskStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None

Atomically enqueue an occurrence and advance its schedule revision.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required
`expected_revision`	`int`	Required
`due_at`	`float`	Required
`next_run_at`	`float`	Required
`task`	`TaskRecord`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.commit_cron_occurrence(
    application_id=...,
    user_id=...,
    schedule_id=...,
    expected_revision=...,
    due_at=...,
    next_run_at=...,
    task=...,
)

PostgresTaskStore.enqueue_task

async PostgresTaskStore.enqueue_task(record: TaskRecord) -> TaskRecord

Persist an immutable task identity, returning exact idempotent retries.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.enqueue_task(record=...)

PostgresTaskStore.get_task

async PostgresTaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Read a scoped job; omitted user scope is reserved for trusted workers.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
record = await store.get_task(application_id=..., job_id=...)

PostgresTaskStore.claim_tasks

async PostgresTaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Claim disjoint batches and retire exhausted attempts after crashes.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

PostgresTaskStore.renew_task_lease

async PostgresTaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Renew only an unexpired attempt still owned by the calling worker.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

PostgresTaskStore.finish_task

async PostgresTaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Fence completion and retries; terminal transitions erase private input.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

PostgresTaskStore.cancel_task

async PostgresTaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Cancel an owned nonterminal job and revoke any running attempt.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

PostgresTaskStore.list_task_metadata

async PostgresTaskStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Push owner, cursor, projection, ordering, and limit into one database query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_task_metadata(application_id=..., user_id=...)

PostgresTaskStore.list_cron_metadata

async PostgresTaskStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Read a bounded metadata projection with owner filtering in one SQL query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest_postgres import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_cron_metadata(application_id=..., user_id=...)

​Exports

​PostgresMemoryStore

​PostgresMemoryStore.delete_all

​PostgresMemoryStore.delete_user

​PostgresMemoryStore.purge_expired

​PostgresMemoryStore.put

​PostgresMemoryStore.get

​PostgresMemoryStore.list

​PostgresMemoryStore.search

​PostgresMemoryStore.delete

​PostgresMemoryStore.start

​PostgresMemoryStore.close

​PostgresStore

​PostgresStore.start

​PostgresStore.close

​PostgresStore.create_cron

​PostgresStore.get_cron

​PostgresStore.list_crons

​PostgresStore.update_cron

​PostgresStore.delete_cron

​PostgresStore.list_due_crons

​PostgresStore.commit_cron_occurrence

​PostgresStore.enqueue_task

​PostgresStore.get_task

​PostgresStore.claim_tasks

​PostgresStore.renew_task_lease

​PostgresStore.finish_task

​PostgresStore.cancel_task

​PostgresStore.list_task_metadata

​PostgresStore.list_cron_metadata

​PostgresStore.as_langgraph_checkpointer

​PostgresStore.create

​PostgresStore.get

​PostgresStore.list

​PostgresStore.put_a2a_task

​PostgresStore.get_a2a_task

​PostgresStore.list_a2a_tasks

​PostgresStore.delete_a2a_task

​PostgresStore.update

​PostgresStore.delete

​PostgresStore.acquire

​PostgresStore.begin_run

​PostgresStore.get_run

​PostgresStore.get_checkpoint

​PostgresStore.list_checkpoints

​PostgresStore.put

​PostgresStore.put_writes

​PostgresStore.get_writes

​PostgresStore.get_writes_batch

​PostgresStore.suspend_continuation

​PostgresStore.get_continuation

​PostgresStore.get_provider_continuation

​PostgresStore.get_continuation_by_external_id

​PostgresStore.list_pending_continuations

​PostgresStore.resolve_continuation

​PostgresStore.claim_continuation

​PostgresStore.cancel_continuation

​PostgresStore.arm_continuation

​PostgresStore.transition

​PostgresStore.delete_run

​PostgresTaskStore

​PostgresTaskStore.start

​PostgresTaskStore.close

​PostgresTaskStore.create_cron

​PostgresTaskStore.get_cron

​PostgresTaskStore.list_crons

​PostgresTaskStore.update_cron

​PostgresTaskStore.delete_cron

​PostgresTaskStore.list_due_crons

​PostgresTaskStore.commit_cron_occurrence

​PostgresTaskStore.enqueue_task

​PostgresTaskStore.get_task

​PostgresTaskStore.claim_tasks

​PostgresTaskStore.renew_task_lease

​PostgresTaskStore.finish_task

​PostgresTaskStore.cancel_task

​PostgresTaskStore.list_task_metadata

​PostgresTaskStore.list_cron_metadata

Exports

PostgresMemoryStore

PostgresMemoryStore.delete_all

PostgresMemoryStore.delete_user

PostgresMemoryStore.purge_expired

PostgresMemoryStore.put

PostgresMemoryStore.get

PostgresMemoryStore.list

PostgresMemoryStore.search

PostgresMemoryStore.delete

PostgresMemoryStore.start

PostgresMemoryStore.close

PostgresStore

PostgresStore.start

PostgresStore.close

PostgresStore.create_cron

PostgresStore.get_cron

PostgresStore.list_crons

PostgresStore.update_cron

PostgresStore.delete_cron

PostgresStore.list_due_crons

PostgresStore.commit_cron_occurrence

PostgresStore.enqueue_task

PostgresStore.get_task

PostgresStore.claim_tasks

PostgresStore.renew_task_lease

PostgresStore.finish_task

PostgresStore.cancel_task

PostgresStore.list_task_metadata

PostgresStore.list_cron_metadata

PostgresStore.as_langgraph_checkpointer

PostgresStore.create

PostgresStore.get

PostgresStore.list

PostgresStore.put_a2a_task

PostgresStore.get_a2a_task

PostgresStore.list_a2a_tasks

PostgresStore.delete_a2a_task

PostgresStore.update

PostgresStore.delete

PostgresStore.acquire

PostgresStore.begin_run

PostgresStore.get_run

PostgresStore.get_checkpoint

PostgresStore.list_checkpoints

PostgresStore.put

PostgresStore.put_writes

PostgresStore.get_writes

PostgresStore.get_writes_batch

PostgresStore.suspend_continuation

PostgresStore.get_continuation

PostgresStore.get_provider_continuation

PostgresStore.get_continuation_by_external_id

PostgresStore.list_pending_continuations

PostgresStore.resolve_continuation

PostgresStore.claim_continuation

PostgresStore.cancel_continuation

PostgresStore.arm_continuation

PostgresStore.transition

PostgresStore.delete_run

PostgresTaskStore

PostgresTaskStore.start

PostgresTaskStore.close

PostgresTaskStore.create_cron

PostgresTaskStore.get_cron

PostgresTaskStore.list_crons

PostgresTaskStore.update_cron

PostgresTaskStore.delete_cron

PostgresTaskStore.list_due_crons

PostgresTaskStore.commit_cron_occurrence

PostgresTaskStore.enqueue_task

PostgresTaskStore.get_task

PostgresTaskStore.claim_tasks

PostgresTaskStore.renew_task_lease

PostgresTaskStore.finish_task

PostgresTaskStore.cancel_task

PostgresTaskStore.list_task_metadata

PostgresTaskStore.list_cron_metadata