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: 97417bfbd8b3
  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
 39
 40class PresenceUpdatedPayload(TypedDict):
 41    presence: dict[str, Any]
 42
 43
 44class PresenceLeftPayload(TypedDict):
 45    connection_id: str
 46
 47
 48class AccessRevokedPayload(TypedDict):
 49    reason: str
 50
 51
 52# Channel for real-time custom object collaboration.
 53# Clients join `api:object:{object_id}` to receive the current object state
 54# and subscribe to field-level updates. Mutations are sent as key:value maps.
 55class ApiObjectChannel:
 56    def __init__(self, channel, join_response=None):
 57        self._channel = channel
 58        self.join_response = join_response
 59
 60    @staticmethod
 61    def topic_by_id(object_id: str) -> str:
 62        return f"api:object:{object_id}"
 63
 64    @classmethod
 65    async def join_by_id(
 66        cls,
 67        socket: "Socket",
 68        object_id: str,
 69        *,
 70        connection_id: str | None = None,
 71        partial_updates: bool | None = None,
 72    ) -> "ApiObjectChannel":
 73        topic = cls.topic_by_id(object_id)
 74        channel = socket.channel(topic)
 75        payload: dict[str, object] = {}
 76        if connection_id is not None:
 77            payload["connection_id"] = connection_id
 78        if partial_updates is not None:
 79            payload["partial_updates"] = partial_updates
 80        join_response = await channel.join(payload)
 81        return cls(channel, join_response)
 82
 83    @staticmethod
 84    def topic_by_row_key(schema_type: str, row_key: str) -> str:
 85        return f"api:object:{schema_type}:{row_key}"
 86
 87    @classmethod
 88    async def join_by_row_key(
 89        cls,
 90        socket: "Socket",
 91        schema_type: str,
 92        row_key: str,
 93        *,
 94        connection_id: str | None = None,
 95        partial_updates: bool | None = None,
 96    ) -> "ApiObjectChannel":
 97        topic = cls.topic_by_row_key(schema_type, row_key)
 98        channel = socket.channel(topic)
 99        payload: dict[str, object] = {}
100        if connection_id is not None:
101            payload["connection_id"] = connection_id
102        if partial_updates is not None:
103            payload["partial_updates"] = partial_updates
104        join_response = await channel.join(payload)
105        return cls(channel, join_response)
106
107    # Leave the underlying channel.
108    async def leave(self):
109        await self._channel.leave()
110
111    async def update_fields(self, payload: UpdateFieldsInput) -> dict[str, Any]:
112        return await self._channel.push("update_fields", payload)
113
114    async def save(self, payload: dict) -> dict[str, Any]:
115        return await self._channel.push("save", payload)
116
117    async def presence_update(self, payload: PresenceUpdateInput) -> dict[str, Any]:
118        return await self._channel.push("presence_update", payload)
119
120    def on_object_updated(
121        self, callback: Callable[[ObjectUpdatedPayload], None]
122    ) -> Callable[[], None]:
123        return self._channel.on("object_updated", callback)
124
125    def on_object_created(
126        self, callback: Callable[[ObjectCreatedPayload], None]
127    ) -> Callable[[], None]:
128        return self._channel.on("object_created", callback)
129
130    def on_object_deleted(
131        self, callback: Callable[[ObjectDeletedPayload], None]
132    ) -> Callable[[], None]:
133        return self._channel.on("object_deleted", callback)
134
135    def on_presence_updated(
136        self, callback: Callable[[PresenceUpdatedPayload], None]
137    ) -> Callable[[], None]:
138        return self._channel.on("presence_updated", callback)
139
140    def on_presence_left(
141        self, callback: Callable[[PresenceLeftPayload], None]
142    ) -> Callable[[], None]:
143        return self._channel.on("presence_left", callback)
144
145    def on_access_revoked(
146        self, callback: Callable[[AccessRevokedPayload], None]
147    ) -> Callable[[], None]:
148        return self._channel.on("access_revoked", callback)