notification_queue

Queue management for Streamlit notifications.

streamlit_notify.notification_queue.default_sort_func(x)[source]

Sort notifications by priority (highest first).

Parameters:: x (StatusElementNotification)
Return type:: int

class streamlit_notify.notification_queue.NotificationPriorityQueue(queue_name, sort_func=None)[source]

Bases: object

A Priority queue for managing Streamlit notifications.

Initialize the queue.

Parameters:

queue_name (str)
sort_func (Callable[[StatusElementNotification], int] | None)

__init__(queue_name, sort_func=None)[source]

Initialize the queue.

Parameters:

queue_name (str)
sort_func (Callable[[StatusElementNotification], int] | None)

Return type:

None

property queue: List[StatusElementNotification]: Get the current queue.

property queue_name: str: Get the name of the queue.

property sort_func: Callable[[StatusElementNotification], int]: Get the sorting function for the queue.

ensure_queue()[source]

Ensure the queue exists in session state.

Return type:: None

_sort()[source]

Sort the queue by priority.

Return type:: None

has_items()[source]

Check if the queue has items.

Return type:: bool

is_empty()[source]

Check if the queue is empty.

Return type:: bool

append(item)[source]

Add an item to the queue.

Parameters:: item (StatusElementNotification)
Return type:: None

extend(items)[source]

Add multiple items to the queue.

Parameters:: items (Iterable[StatusElementNotification])
Return type:: None

remove(item)[source]

Remove an item from the queue.

Parameters:: item (StatusElementNotification | int)
Return type:: None

contains(item)[source]

Check if an item is in the queue.

Parameters:: item (StatusElementNotification)
Return type:: bool

_get_all_less_than(priority)[source]

Get all items in the queue with priority less than the specified value.

Parameters:: priority (int)
Return type:: List[StatusElementNotification]

_get_all_less_than_equal_to(priority)[source]

Get all items in the queue with priority less than or equal to the specified value.

Parameters:: priority (int)
Return type:: List[StatusElementNotification]

_get_all_greater_than(priority)[source]

Get all items in the queue with priority greater than the specified value.

Parameters:: priority (int)
Return type:: List[StatusElementNotification]

_get_all_greater_than_equal_to(priority)[source]

Get all items in the queue with priority greater than or equal to the specified value.

Parameters:: priority (int)
Return type:: List[StatusElementNotification]

_get_all_equal_to(priority)[source]

Get all items in the queue with priority equal to the specified value.

Parameters:: priority (int)
Return type:: List[StatusElementNotification]

get_all(priority=None, priority_type='eq')[source]

Get all items in the queue.

Parameters:

priority (int | None)
priority_type (Literal['le', 'lt', 'ge', 'gt', 'eq'])

Return type:

List[StatusElementNotification]

clear()[source]

Clear the queue.

Return type:: None

pop(index=0)[source]

Pop an item from the queue.

Parameters:: index (int)
Return type:: StatusElementNotification

get(index=0)[source]

Get an item from the queue without removing it.

Parameters:: index (int)
Return type:: StatusElementNotification

size()[source]

Get the size of the queue.

Return type:: int

__len__()[source]

Get the size of the queue.

Return type:: int

__repr__()[source]

String representation of the queue.

Return type:: str

__str__()[source]

String representation of the queue.

Return type:: str

__bool__()[source]

Boolean representation of the queue.

Return type:: bool

__contains__(item)[source]

Check if an item is in the queue.

Parameters:: item (StatusElementNotification)
Return type:: bool

__getitem__(index)[source]

Get an item by index.

Parameters:: index (int)
Return type:: StatusElementNotification

__setitem__(index, value)[source]

Set an item by index.

Parameters:

index (int)
value (StatusElementNotification)

Return type:

None

__delitem__(index)[source]

Delete an item by index.

Parameters:: index (int)
Return type:: None

__hash__()[source]

Hash of the queue based on its name.

Return type:: int

__eq__(other)[source]

Check if this queue is equal to another.

Parameters:: other (object)
Return type:: bool

__ne__(other)[source]

Check if this queue is not equal to another.

Parameters:: other (object)
Return type:: bool

__lt__(other)[source]

Check if this queue is less than another.

Parameters:: other (object)
Return type:: bool

__iter__()[source]

Iterate over the notifications in the queue.

Return type:: Generator[StatusElementNotification, None, None]

__reversed__()[source]

Iterate over the notifications in reverse order.

Return type:: Generator[StatusElementNotification, None, None]

__copy__()[source]

Create a shallow copy of the queue.

Return type:: NotificationPriorityQueue

__deepcopy__(memo)[source]

Create a deep copy of the queue.

Parameters:: memo (Dict[int, object])
Return type:: NotificationPriorityQueue