pyphoenix.message_queue

Message queue implementation with delivery guarantees for PyPhoenix.

Classes

MessageQueue([delivery_guarantee])	Message queue with configurable delivery guarantees.
MessageStatus(value[, names, module, ...])	Status of a message in the queue.
QueuedMessage(id, topic, payload, created_at)	A message in the queue with metadata.

class pyphoenix.message_queue.MessageStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Status of a message in the queue.

PENDING = 'pending'

PROCESSING = 'processing'

DELIVERED = 'delivered'

FAILED = 'failed'

EXPIRED = 'expired'

class pyphoenix.message_queue.QueuedMessage(id, topic, payload, created_at, expires_at=None, attempts=0, max_attempts=3, status=MessageStatus.PENDING, last_attempt_at=None, error=None, serialized_payload=None)

A message in the queue with metadata.

Parameters:

• id (str)

• topic (str)

• payload (Any)

• created_at (float)

• expires_at (Optional[float])

• attempts (int)

• max_attempts (int)

• status (MessageStatus)

• last_attempt_at (Optional[float])

• error (Optional[str])

• serialized_payload (Optional[bytes])

id: str

topic: str

payload: Any

created_at: float

expires_at: Optional[float] = None

attempts: int = 0

max_attempts: int = 3

status: MessageStatus = 'pending'

last_attempt_at: Optional[float] = None

error: Optional[str] = None

serialized_payload: Optional[bytes] = None

is_expired()

Check if the message has expired.

Return type:

bool

can_retry()

Check if the message can be retried.

Return type:

bool

mark_attempt(error=None)

Mark an attempt to deliver the message.

Parameters:

error (Optional[str])

Return type:

None

mark_delivered()

Mark the message as successfully delivered.

Return type:

None

mark_failed(error)

Mark the message as failed.

Parameters:

error (str)

Return type:

None

__init__(id, topic, payload, created_at, expires_at=None, attempts=0, max_attempts=3, status=MessageStatus.PENDING, last_attempt_at=None, error=None, serialized_payload=None)

Parameters:

• id (str)

• topic (str)

• payload (Any)

• created_at (float)

• expires_at (Optional[float])

• attempts (int)

• max_attempts (int)

• status (MessageStatus)

• last_attempt_at (Optional[float])

• error (Optional[str])

• serialized_payload (Optional[bytes])

class pyphoenix.message_queue.MessageQueue(delivery_guarantee='at_least_once')

Message queue with configurable delivery guarantees.

Supports different delivery modes: - at_most_once: Fire and forget, no retries - at_least_once: Retry until delivered, may deliver duplicates - exactly_once: Deliver exactly once (requires deduplication)

Parameters:

delivery_guarantee (str)

__init__(delivery_guarantee='at_least_once')

Parameters:

delivery_guarantee (str)

async start()

Start the message processing loop.

Return type:

None

async stop()

Stop the message processing loop.

Return type:

None

async enqueue(topic, payload, ttl=None, max_attempts=None)

Enqueue a message for delivery.

Parameters:

• topic (str) – The topic to publish to

• payload (Any) – The message payload

• ttl (Optional[float]) – Time to live in seconds (None for no expiry)

• max_attempts (Optional[int]) – Maximum delivery attempts (None for default)

Return type:

str

Returns:

Message ID

async get_pending_messages(topic=None)

Get pending messages for processing.

Parameters:

topic (Optional[str])

Return type:

list[QueuedMessage]

async mark_delivered(message_id)

Mark a message as successfully delivered.

Parameters:

message_id (str)

Return type:

bool

async mark_failed(message_id, error)

Mark a message delivery attempt as failed.

Parameters:

• message_id (str)

• error (str)

Return type:

bool

async is_duplicate(message_id)

Check if a message has already been delivered (for exactly_once).

Parameters:

message_id (str)

Return type:

bool

async cleanup_expired()

Clean up expired and completed messages.

Return type:

int

async get_stats()

Get queue statistics.

Return type:

dict[str, Any]