archastro.platform.channels.api_object_channel 

 1# Copyright (c) 2026 ArchAstro Inc. All Rights Reserved.
 2# This file is auto-generated by @archastro/sdk-generator. Do not edit.
 3# Content hash: 24dfaffabee5
 4
 5from collections.abc import Callable
 6from typing import TYPE_CHECKING, Any, TypedDict
 7
 8if TYPE_CHECKING:
 9    from archastro.phx_channel.socket import Socket
10
11
12class UpdateFieldsInput(TypedDict):
13    fields: dict[str, Any]
14
15
16class ObjectUpdatedPayload(TypedDict, total=False):
17    fields: dict[str, Any] | None
18    id: str | None
19
20
21class ObjectCreatedPayload(TypedDict, total=False):
22    fields: dict[str, Any] | None
23    id: str | None
24
25
26# Channel for real-time custom object collaboration.
27# Clients join `api:object:{object_id}` to receive the current object state
28# and subscribe to field-level updates. Mutations are sent as key:value maps.
29class ApiObjectChannel:
30    def __init__(self, channel, join_response=None):
31        self._channel = channel
32        self.join_response = join_response
33
34    @staticmethod
35    def topic_by_id(object_id: str) -> str:
36        return f"api:object:{object_id}"
37
38    @classmethod
39    async def join_by_id(cls, socket: "Socket", object_id: str) -> "ApiObjectChannel":
40        topic = cls.topic_by_id(object_id)
41        channel = socket.channel(topic)
42        join_response = await channel.join()
43        return cls(channel, join_response)
44
45    @staticmethod
46    def topic_by_row_key(schema_type: str, row_key: str) -> str:
47        return f"api:object:{schema_type}:{row_key}"
48
49    @classmethod
50    async def join_by_row_key(
51        cls, socket: "Socket", schema_type: str, row_key: str
52    ) -> "ApiObjectChannel":
53        topic = cls.topic_by_row_key(schema_type, row_key)
54        channel = socket.channel(topic)
55        join_response = await channel.join()
56        return cls(channel, join_response)
57
58    # Leave the underlying channel.
59    async def leave(self):
60        await self._channel.leave()
61
62    async def update_fields(self, payload: UpdateFieldsInput) -> dict[str, Any]:
63        return await self._channel.push("update_fields", payload)
64
65    async def save(self, payload: dict) -> dict[str, Any]:
66        return await self._channel.push("save", payload)
67
68    def on_object_updated(
69        self, callback: Callable[[ObjectUpdatedPayload], None]
70    ) -> Callable[[], None]:
71        return self._channel.on("object_updated", callback)
72
73    def on_object_created(
74        self, callback: Callable[[ObjectCreatedPayload], None]
75    ) -> Callable[[], None]:
76        return self._channel.on("object_created", callback)
class UpdateFieldsInput(typing.TypedDict): 
13class UpdateFieldsInput(TypedDict):
14    fields: dict[str, Any]
fields: dict[str, typing.Any]
class ObjectUpdatedPayload(typing.TypedDict): 
17class ObjectUpdatedPayload(TypedDict, total=False):
18    fields: dict[str, Any] | None
19    id: str | None
fields: dict[str, typing.Any] | None
id: str | None
class ObjectCreatedPayload(typing.TypedDict): 
22class ObjectCreatedPayload(TypedDict, total=False):
23    fields: dict[str, Any] | None
24    id: str | None
fields: dict[str, typing.Any] | None
id: str | None
class ApiObjectChannel: 
30class ApiObjectChannel:
31    def __init__(self, channel, join_response=None):
32        self._channel = channel
33        self.join_response = join_response
34
35    @staticmethod
36    def topic_by_id(object_id: str) -> str:
37        return f"api:object:{object_id}"
38
39    @classmethod
40    async def join_by_id(cls, socket: "Socket", object_id: str) -> "ApiObjectChannel":
41        topic = cls.topic_by_id(object_id)
42        channel = socket.channel(topic)
43        join_response = await channel.join()
44        return cls(channel, join_response)
45
46    @staticmethod
47    def topic_by_row_key(schema_type: str, row_key: str) -> str:
48        return f"api:object:{schema_type}:{row_key}"
49
50    @classmethod
51    async def join_by_row_key(
52        cls, socket: "Socket", schema_type: str, row_key: str
53    ) -> "ApiObjectChannel":
54        topic = cls.topic_by_row_key(schema_type, row_key)
55        channel = socket.channel(topic)
56        join_response = await channel.join()
57        return cls(channel, join_response)
58
59    # Leave the underlying channel.
60    async def leave(self):
61        await self._channel.leave()
62
63    async def update_fields(self, payload: UpdateFieldsInput) -> dict[str, Any]:
64        return await self._channel.push("update_fields", payload)
65
66    async def save(self, payload: dict) -> dict[str, Any]:
67        return await self._channel.push("save", payload)
68
69    def on_object_updated(
70        self, callback: Callable[[ObjectUpdatedPayload], None]
71    ) -> Callable[[], None]:
72        return self._channel.on("object_updated", callback)
73
74    def on_object_created(
75        self, callback: Callable[[ObjectCreatedPayload], None]
76    ) -> Callable[[], None]:
77        return self._channel.on("object_created", callback)
ApiObjectChannel(channel, join_response=None) 
31    def __init__(self, channel, join_response=None):
32        self._channel = channel
33        self.join_response = join_response
join_response
@staticmethod
def topic_by_id(object_id: str) -> str: 
35    @staticmethod
36    def topic_by_id(object_id: str) -> str:
37        return f"api:object:{object_id}"
@classmethod
async def join_by_id( cls, socket: archastro.phx_channel.Socket, object_id: str) -> ApiObjectChannel: 
39    @classmethod
40    async def join_by_id(cls, socket: "Socket", object_id: str) -> "ApiObjectChannel":
41        topic = cls.topic_by_id(object_id)
42        channel = socket.channel(topic)
43        join_response = await channel.join()
44        return cls(channel, join_response)
@staticmethod
def topic_by_row_key(schema_type: str, row_key: str) -> str: 
46    @staticmethod
47    def topic_by_row_key(schema_type: str, row_key: str) -> str:
48        return f"api:object:{schema_type}:{row_key}"
@classmethod
async def join_by_row_key( cls, socket: archastro.phx_channel.Socket, schema_type: str, row_key: str) -> ApiObjectChannel: 
50    @classmethod
51    async def join_by_row_key(
52        cls, socket: "Socket", schema_type: str, row_key: str
53    ) -> "ApiObjectChannel":
54        topic = cls.topic_by_row_key(schema_type, row_key)
55        channel = socket.channel(topic)
56        join_response = await channel.join()
57        return cls(channel, join_response)
async def leave(self): 
60    async def leave(self):
61        await self._channel.leave()
async def update_fields( self, payload: UpdateFieldsInput) -> dict[str, typing.Any]: 
63    async def update_fields(self, payload: UpdateFieldsInput) -> dict[str, Any]:
64        return await self._channel.push("update_fields", payload)
async def save(self, payload: dict) -> dict[str, typing.Any]: 
66    async def save(self, payload: dict) -> dict[str, Any]:
67        return await self._channel.push("save", payload)
def on_object_updated( self, callback: Callable[[ObjectUpdatedPayload], None]) -> Callable[[], None]: 
69    def on_object_updated(
70        self, callback: Callable[[ObjectUpdatedPayload], None]
71    ) -> Callable[[], None]:
72        return self._channel.on("object_updated", callback)
def on_object_created( self, callback: Callable[[ObjectCreatedPayload], None]) -> Callable[[], None]: 
74    def on_object_created(
75        self, callback: Callable[[ObjectCreatedPayload], None]
76    ) -> Callable[[], None]:
77        return self._channel.on("object_created", callback)