archastro.platform.channels.api_object_channel

  1# Copyright (c) 2026 ArchAstro Inc. Licensed under the MIT License.
  2# This file is auto-generated by @archastro/sdk-generator. Do not edit.
  3# Content hash: 5e56b0c728e8
  4
  5from collections.abc import Callable
  6from typing import TYPE_CHECKING, Any, Required, TypedDict
  7
  8if TYPE_CHECKING:
  9    from archastro.phx_channel.socket import Socket
 10
 11
 12class UpdateFieldsInput(TypedDict, total=False):
 13    fields: Required[dict[str, Any]]
 14    operation_id: str
 15
 16
 17class PresenceUpdateInput(TypedDict):
 18    presence: dict[str, Any]
 19
 20
 21class ObjectUpdatedPayload(TypedDict):
 22    fields: dict[str, Any]
 23    id: str
 24    operation_id: str
 25    partial: bool
 26
 27
 28class ObjectCreatedPayload(TypedDict):
 29    connection_id: str
 30    fields: dict[str, Any]
 31    id: str
 32    presence: list[dict[str, Any]]
 33    readonly: bool
 34
 35
 36class ObjectDeletedPayload(TypedDict):
 37    id: str
 38    reason: str
 39
 40
 41class PresenceUpdatedPayload(TypedDict):
 42    presence: dict[str, Any]
 43
 44
 45class PresenceLeftPayload(TypedDict):
 46    connection_id: str
 47
 48
 49class AccessRevokedPayload(TypedDict):
 50    reason: str
 51
 52
 53# Channel for real-time custom object collaboration.
 54# Clients join `api:object:{object_id}` to receive the current object state
 55# and subscribe to field-level updates. Mutations are sent as key:value maps.
 56class ApiObjectChannel:
 57    def __init__(self, channel, join_response=None):
 58        self._channel = channel
 59        self.join_response = join_response
 60
 61    @staticmethod
 62    def topic_by_id(object_id: str) -> str:
 63        return f"api:object:{object_id}"
 64
 65    @classmethod
 66    async def join_by_id(
 67        cls,
 68        socket: "Socket",
 69        object_id: str,
 70        *,
 71        connection_id: str | None = None,
 72        partial_updates: bool | None = None,
 73    ) -> "ApiObjectChannel":
 74        topic = cls.topic_by_id(object_id)
 75        channel = socket.channel(topic)
 76        payload: dict[str, object] = {}
 77        if connection_id is not None:
 78            payload["connection_id"] = connection_id
 79        if partial_updates is not None:
 80            payload["partial_updates"] = partial_updates
 81        join_response = await channel.join(payload)
 82        return cls(channel, join_response)
 83
 84    @staticmethod
 85    def topic_by_row_key(schema_type: str, row_key: str) -> str:
 86        return f"api:object:{schema_type}:{row_key}"
 87
 88    @classmethod
 89    async def join_by_row_key(
 90        cls,
 91        socket: "Socket",
 92        schema_type: str,
 93        row_key: str,
 94        *,
 95        connection_id: str | None = None,
 96        partial_updates: bool | None = None,
 97    ) -> "ApiObjectChannel":
 98        topic = cls.topic_by_row_key(schema_type, row_key)
 99        channel = socket.channel(topic)
100        payload: dict[str, object] = {}
101        if connection_id is not None:
102            payload["connection_id"] = connection_id
103        if partial_updates is not None:
104            payload["partial_updates"] = partial_updates
105        join_response = await channel.join(payload)
106        return cls(channel, join_response)
107
108    # Leave the underlying channel.
109    async def leave(self):
110        await self._channel.leave()
111
112    async def update_fields(self, payload: UpdateFieldsInput) -> dict[str, Any]:
113        return await self._channel.push("update_fields", payload)
114
115    async def save(self, payload: dict) -> dict[str, Any]:
116        return await self._channel.push("save", payload)
117
118    async def presence_update(self, payload: PresenceUpdateInput) -> dict[str, Any]:
119        return await self._channel.push("presence_update", payload)
120
121    def on_object_updated(
122        self, callback: Callable[[ObjectUpdatedPayload], None]
123    ) -> Callable[[], None]:
124        return self._channel.on("object_updated", callback)
125
126    def on_object_created(
127        self, callback: Callable[[ObjectCreatedPayload], None]
128    ) -> Callable[[], None]:
129        return self._channel.on("object_created", callback)
130
131    def on_object_deleted(
132        self, callback: Callable[[ObjectDeletedPayload], None]
133    ) -> Callable[[], None]:
134        return self._channel.on("object_deleted", callback)
135
136    def on_presence_updated(
137        self, callback: Callable[[PresenceUpdatedPayload], None]
138    ) -> Callable[[], None]:
139        return self._channel.on("presence_updated", callback)
140
141    def on_presence_left(
142        self, callback: Callable[[PresenceLeftPayload], None]
143    ) -> Callable[[], None]:
144        return self._channel.on("presence_left", callback)
145
146    def on_access_revoked(
147        self, callback: Callable[[AccessRevokedPayload], None]
148    ) -> Callable[[], None]:
149        return self._channel.on("access_revoked", callback)