Python Package API

This page covers the primadb-python package surface. It is generated directly from the public stub file shipped with the package.

Generated from packages/primadb-python/python/primadb/__init__.pyi.

PeerHookContext

Kind: class

class PeerHookContext(TypedDict, total=False):
    peerId: str
    replicaId: str
    transport: str
    capabilities: list[str]
    topics: list[str]
    metadata: dict[str, str]

ConnectHookContext

Kind: class

class ConnectHookContext(TypedDict, total=False):
    peer: PeerHookContext
    transport: str
    relayUrl: Optional[str]

RoomHookContext

Kind: class

class RoomHookContext(TypedDict, total=False):
    peerId: str
    room: str
    transport: str
    peer: Optional[PeerHookContext]

ServeRequestContext

Kind: class

class ServeRequestContext(TypedDict, total=False):
    peerId: str
    transport: str
    requestId: Optional[str]
    watchId: Optional[str]
    request: Any

ServeResultContext

Kind: class

class ServeResultContext(TypedDict, total=False):
    peerId: str
    transport: str
    requestId: Optional[str]
    watchId: Optional[str]
    request: Any
    initial: bool

NetworkHooks

Kind: class

class NetworkHooks(Protocol):
    def on_connect(self, context: ConnectHookContext, /) -> Any: ...
    def on_join_room(self, context: RoomHookContext, /) -> Any: ...
    def on_pull(self, context: ServeRequestContext, /) -> Any: ...
    def on_watch(self, context: ServeRequestContext, /) -> Any: ...
    def on_serve_result(self, context: ServeResultContext, result: Any, /) -> Any: ...

Primadb

Kind: class

class Primadb:
    def __init__(self, replica_id: Optional[str] = ...) -> None: ...
    def replica_id(self) -> str: ...
    def chain(self, root: str) -> Chain: ...
    def snapshot(self) -> Any: ...
    def snapshot_for_root(self, root: Optional[str] = ...) -> Any: ...
    def export_snapshot_json(self) -> str: ...
    def import_snapshot_json(self, payload: str) -> None: ...
    def merge_snapshot_json(self, payload: str) -> None: ...
    def pending_operations(self) -> Any: ...
    def pending_envelope(self) -> Any: ...
    def export_pending_operations_json(self) -> str: ...
    def drain_pending_operations(self) -> Any: ...
    def drain_pending_envelope(self) -> Any: ...
    def apply_operations(self, operations: Any) -> int: ...
    def apply_envelope(self, envelope: Any) -> int: ...
    def apply_operations_json(self, payload: str) -> int: ...
    def open_durable_storage(self, config: Any) -> Any: ...
    def open_blob_storage(self, config: Any) -> Any: ...
    def connect_relay(self, config: Any) -> WebSocketSync: ...
    def connect_mesh(self, config: Any) -> WebRtcMesh: ...
    def set_network_hooks(self, hooks: NetworkHooks | dict[str, Any] | object | None) -> None: ...
    def clear_network_hooks(self) -> None: ...

Chain

Kind: class

class Chain:
    def field(self, key: str) -> Chain: ...
    def path(self) -> str: ...
    def put(self, value: Any) -> None: ...
    def put_bytes(self, value: bytes) -> None: ...
    def put_signed(self, value: Any, certificate: Optional[str] = ...) -> None: ...
    def once(self) -> Any: ...
    def once_bytes(self) -> Optional[bytes]: ...
    def unset(self) -> None: ...
    def set(self, value: Any) -> str: ...
    def set_signed(self, value: Any, certificate: Optional[str] = ...) -> str: ...
    def remove(self, value: Any) -> str: ...
    def put_blob(self, value: bytes, media_type: Optional[str] = ...) -> Any: ...
    def blob_ref(self) -> Any: ...
    def get_blob(self) -> Optional[bytes]: ...
    def map(self) -> Any: ...
    def query(self, spec: Any) -> Any: ...
    def first_query(self, spec: Any) -> Any: ...
    def scan(self, spec: Any) -> Any: ...
    def subscribe(self) -> Subscription: ...

Subscription

Kind: class

class Subscription:
    def next(self) -> Any: ...
    def try_next(self) -> Any: ...
    def close(self) -> None: ...

RemoteWatch

Kind: class

class RemoteWatch:
    def next(self) -> Any: ...
    def try_next(self) -> Any: ...
    def close(self) -> None: ...

WebSocketSync

Kind: class

class WebSocketSync:
    def is_connected(self) -> bool: ...
    def pending_count(self) -> int: ...
    def inflight_count(self) -> int: ...
    def known_peer_count(self) -> int: ...
    def recommended_peers(self) -> Any: ...
    def remote_get(self, peer_id: str, path: Any) -> Any: ...
    def remote_query(self, peer_id: str, path: Any, spec: Any) -> Any: ...
    def remote_lex(self, peer_id: str, path: Any, spec: Any) -> Any: ...
    def remote_snapshot(self, peer_id: str, root: Optional[str] = ...) -> Any: ...
    def watch_remote_get(self, peer_id: str, path: Any) -> RemoteWatch: ...
    def watch_remote_map(self, peer_id: str, path: Any) -> RemoteWatch: ...
    def watch_remote_query(self, peer_id: str, path: Any, spec: Any) -> RemoteWatch: ...
    def watch_remote_lex(self, peer_id: str, path: Any, spec: Any) -> RemoteWatch: ...
    def watch_remote_snapshot(self, peer_id: str, root: Optional[str] = ...) -> RemoteWatch: ...
    def flush_pending(self) -> int: ...
    def retry_inflight(self) -> int: ...
    def close(self) -> None: ...

WebRtcMesh

Kind: class

class WebRtcMesh:
    def peer_id(self) -> str: ...
    def signaling_mode(self) -> str: ...
    def relay_url(self) -> Optional[str]: ...
    def relay_connected(self) -> bool: ...
    def peer_count(self) -> int: ...
    def open_peer_count(self) -> int: ...
    def inflight_count(self) -> int: ...
    def recommended_peers(self) -> Any: ...
    def watch_remote_get(self, peer_id: str, path: Any) -> RemoteWatch: ...
    def watch_remote_map(self, peer_id: str, path: Any) -> RemoteWatch: ...
    def watch_remote_query(self, peer_id: str, path: Any, spec: Any) -> RemoteWatch: ...
    def watch_remote_lex(self, peer_id: str, path: Any, spec: Any) -> RemoteWatch: ...
    def watch_remote_snapshot(self, peer_id: str, root: Optional[str] = ...) -> RemoteWatch: ...
    def flush_pending(self) -> int: ...
    def retry_inflight(self) -> int: ...
    def close(self) -> None: ...