pyphoenix.presence

Presence tracking implementation for PyPhoenix.

Functions

get_presence()

Get or create the global Presence instance.

Classes

Presence()

Distributed presence tracking system.

class pyphoenix.presence.Presence

Distributed presence tracking system.

Tracks which processes/users are present on which topics across the system. This is a local implementation that will later be extended for distribution.

__init__()

Initialize the Presence tracker.

async track(pid, topic, meta)

Track a process’s presence on a topic.

Parameters:

• pid (str) – Process/user identifier

• topic (str) – The topic to track presence on

• meta (dict[str, Any]) – Metadata about the presence (e.g., user info, online_at timestamp)

Return type:

None

async untrack(pid, topic)

Stop tracking a process’s presence.

Parameters:

• pid (str) – Process/user identifier

• topic (str) – The topic to stop tracking on

Return type:

None

async list(topic)

List all presences for a topic.

Parameters:

topic (str) – The topic to list presences for

Return type:

dict[str, PresenceState]

Returns:

Dictionary mapping process IDs to their presence states

async get_presence(pid, topic)

Get presence state for a specific process on a topic.

Parameters:

• pid (str) – Process/user identifier

• topic (str) – The topic to check

Return type:

Optional[PresenceState]

Returns:

PresenceState if present, None otherwise

on_join(callback)

Register callback for when a presence joins.

Parameters:

callback (Callable) – Function called with (topic, pid, presence_state)

Return type:

None

on_leave(callback)

Register callback for when a presence leaves.

Parameters:

callback (Callable) – Function called with (topic, pid, presence_state)

Return type:

None

async sync_presence_state(topic)

Get the current presence state for synchronization.

This method returns the full presence state in a format suitable for sending to new joiners or for synchronization between nodes.

Parameters:

topic (str) – The topic to get state for

Return type:

dict[str, Any]

Returns:

Serializable presence state

async topic_count()

Get the number of topics with active presences.

Return type:

int

Returns:

Number of topics

async total_presence_count()

Get the total number of presences across all topics.

Return type:

int

Returns:

Total presence count

async presence_count_for_topic(topic)

Get the number of presences for a specific topic.

Parameters:

topic (str) – The topic to count presences for

Return type:

int

Returns:

Number of presences on the topic

async cleanup_stale_presences(max_age_seconds=3600)

Clean up stale presences that haven’t been updated recently.

Parameters:

max_age_seconds (float) – Maximum age in seconds before considering a presence stale

Return type:

int

Returns:

Number of stale presences cleaned up

pyphoenix.presence.get_presence()

Get or create the global Presence instance.

Return type:

Presence