pyphoenix.types

Core types and data structures for PyPhoenix.

Classes

`ChannelState`(value[, names, module, ...])	Channel connection states.
`JoinResponse`(status, response)	Response from joining a channel.
`Message`(topic, event, payload[, ref, join_ref])	Phoenix message structure.
`PresenceMeta`(online_at[, user_id, metadata])	Metadata for presence tracking.
`PresenceState`(metas)	State of a presence entry.
`PushResponse`(status, response[, ref])	Response from pushing a message to a channel.

class pyphoenix.types.ChannelState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Channel connection states.

CLOSED = 'closed'

ERRORED = 'errored'

JOINED = 'joined'

JOINING = 'joining'

LEAVING = 'leaving'

class pyphoenix.types.JoinResponse(status, response)[source]

Response from joining a channel.

Parameters:

status (Literal['ok', 'error'])
response (dict[str, Any])

status: Literal['ok', 'error']

response: dict[str, Any]

property is_ok: bool

property is_error: bool

__init__(status, response)

Parameters:

status (Literal['ok', 'error'])
response (dict[str, Any])

class pyphoenix.types.PushResponse(status, response, ref=None)[source]

Response from pushing a message to a channel.

Parameters:

status (Literal['ok', 'error', 'timeout'])
response (dict[str, Any])
ref (Optional[str])

status: Literal['ok', 'error', 'timeout']

response: dict[str, Any]

ref: Optional[str] = None

property is_ok: bool

property is_error: bool

property is_timeout: bool

__init__(status, response, ref=None)

Parameters:

status (Literal['ok', 'error', 'timeout'])
response (dict[str, Any])
ref (Optional[str])

class pyphoenix.types.Message(topic, event, payload, ref=None, join_ref=None)[source]

Phoenix message structure.

Parameters:

topic (str)
event (str)
payload (dict[str, Any])
ref (Optional[str])
join_ref (Optional[str])

topic: str

event: str

payload: dict[str, Any]

ref: Optional[str] = None

join_ref: Optional[str] = None

to_list()[source]

Convert to Phoenix wire format [join_ref, ref, topic, event, payload].

Return type:: list

classmethod from_list(data)[source]

Create message from Phoenix wire format.

Parameters:: data (list)
Return type:: Message

__init__(topic, event, payload, ref=None, join_ref=None)

Parameters:

topic (str)
event (str)
payload (dict[str, Any])
ref (Optional[str])
join_ref (Optional[str])

class pyphoenix.types.PresenceMeta(online_at, user_id=None, metadata=None)[source]

Metadata for presence tracking.

Parameters:

online_at (float)
user_id (Optional[str])
metadata (Optional[dict[str, Any]])

online_at: float

user_id: Optional[str] = None

metadata: Optional[dict[str, Any]] = None

__init__(online_at, user_id=None, metadata=None)

Parameters:

online_at (float)
user_id (Optional[str])
metadata (Optional[dict[str, Any]])

class pyphoenix.types.PresenceState(metas)[source]

State of a presence entry.

Parameters:: metas (list[PresenceMeta])

metas: list[PresenceMeta]

property first_meta: PresenceMeta | None: Get the first (oldest) meta entry.

__init__(metas)

Parameters:: metas (list[PresenceMeta])