streamlit_notify

Initialization module for the st_notify package.

class streamlit_notify.RerunnableStatusElement(base_widget)[source]

Bases: object

A wrapper for Streamlit widgets to enable notification queueing.

Initialize the wrapper.

Parameters:: base_widget (Callable[[...], Any])

__init__(base_widget)[source]

Initialize the wrapper.

Parameters:: base_widget (Callable[[...], Any])
Return type:: None

property session_state_key: str: Get the session state key for the notification queue.

property base_widget: Callable[[...], Any]: Get the base widget function.

property name: str: Get the name of the base widget.

property notifications: NotificationPriorityQueue: Get the notification queue.

setup_queue()[source]

Ensure the notification queue is set up in session state.

Return type:: None

__call__(*args, **kwargs)[source]

Add a notification to the queue.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

create_notification(*args, **kwargs)[source]

Create a notification without adding it to the queue.

Parameters:

args (Any)
kwargs (Any)

Return type:

StatusElementNotification

notify(remove=True, priority=None, priority_type='eq')[source]

Display all queued notifications.

Parameters:

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

Return type:

None

__repr__()[source]

String representation of the wrapper.

Return type:: str

__str__()[source]

String representation of the wrapper.

Return type:: str

streamlit_notify.create_notification(*args, **kwargs)[source]

Create a notification without adding it to the queue.

Parameters:

args (Any)
kwargs (Any)

Return type:

StatusElementNotification

streamlit_notify.notify(notification_type=None, remove=True, priority=None, priority_type='ge')[source]

Display queued notifications.

Parameters:

notification_type (Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception'] | ~typing.Iterable[~typing.Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception']] | None)
remove (bool)
priority (int | None)
priority_type (Literal['le', 'ge', 'eq'])

Return type:

None

streamlit_notify.get_notifications(notification_type=None, priority=None, priority_type='ge')[source]

Retrieve all notifications for a specific type(s).

Parameters:

notification_type (Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception'] | ~typing.Iterable[~typing.Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception']] | None)
priority (int | None)
priority_type (Literal['le', 'ge', 'eq'])

Return type:

List[StatusElementNotification]

streamlit_notify.clear_notifications(notification_type=None)[source]

Clear notifications for a specific type(s).

Parameters:: notification_type (Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception'] | ~typing.Iterable[~typing.Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception']] | None)
Return type:: None

streamlit_notify.get_notification_queue(notification_type)[source]

Retrieve notification priority queue.

Parameters:: notification_type (Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception'])
Return type:: NotificationPriorityQueue

streamlit_notify.has_notifications(notification_type=None)[source]

Check if there are any notifications.

Parameters:: notification_type (Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception'] | ~typing.Iterable[~typing.Literal['toast', 'balloons', 'snow', 'success', 'info', 'error', 'warning', 'exception']] | None)
Return type:: bool

streamlit_notify.remove_notifications(notifications)[source]

Remove notification(s) from the queue.

Parameters:: notifications (StatusElementNotification | Iterable[StatusElementNotification])
Return type:: None

streamlit_notify.add_notifications(notifications)[source]

Add notification(s) to the queue.

Parameters:: notifications (StatusElementNotification | Iterable[StatusElementNotification])
Return type:: None

streamlit_notify.toast_stn(*args, **kwargs)[source]

Display a toast notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.balloons_stn(*args, **kwargs)[source]

Display a balloons notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.snow_stn(*args, **kwargs)[source]

Display a snow notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.success_stn(*args, **kwargs)[source]

Display a success notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.info_stn(*args, **kwargs)[source]

Display an info notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.error_stn(*args, **kwargs)[source]

Display an error notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.warning_stn(*args, **kwargs)[source]

Display a warning notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

streamlit_notify.exception_stn(*args, **kwargs)[source]

Display an exception notification.

Parameters:

args (Any)
kwargs (Any)

Return type:

None

class streamlit_notify.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

class streamlit_notify.StatusElementNotification(base_widget, args, priority=0, data=None)[source]

Bases: object

Represents a notification for a Streamlit widget.

Parameters:

base_widget (Callable[[...], Any])
args (OrderedDict[str, Any])
priority (int)
data (Any)

base_widget: Callable[[...], Any]

args: OrderedDict[str, Any]

priority: int = 0

data: Any = None

notify()[source]

Display the notification using the widget.

Return type:: None

property name: str: Get the name of the widget function.

__init__(base_widget, args, priority=0, data=None)

Parameters:

base_widget (Callable[[...], Any])
args (OrderedDict[str, Any])
priority (int)
data (Any | None)

Return type:

None

class streamlit_notify.NotificationType(value)[source]

Bases: Enum

Enum class for notification types.

TOAST = 'toast'

BALLOONS = 'balloons'

SNOW = 'snow'

SUCCESS = 'success'

INFO = 'info'

ERROR = 'error'

WARNING = 'warning'

EXCEPTION = 'exception'

streamlit_notify.get_status_element(notification_type)[source]

Get the Streamlit status element corresponding to the notification type.

Parameters:: notification_type (NotificationType | str)
Return type:: RerunnableStatusElement

streamlit_notify.init_session_state()[source]

Initialize session state for all notification elements.

Return type:: None

streamlit_notify.__getattr__(name)[source]

Delegate attribute access to Streamlit if not found in this module.

This function is not included in type checking to prevent it from appearing in stub files.

Args:: name: Name of the attribute to get.
Returns:: The requested attribute from Streamlit.
Raises:: AttributeError: If the attribute is not found in Streamlit.

Parameters:: name (str)
Return type:: Any