# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2026)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

from collections.abc import Iterator, MutableMapping
from typing import Any, Final

from streamlit import logger as _logger
from streamlit import runtime
from streamlit.elements.lib.utils import Key
from streamlit.runtime.metrics_util import gather_metrics
from streamlit.runtime.state.common import require_valid_user_key
from streamlit.runtime.state.safe_session_state import SafeSessionState
from streamlit.runtime.state.session_state import SessionState

_LOGGER: Final = _logger.get_logger(__name__)


_state_use_warning_already_displayed: bool = False
# The mock session state is used as a fallback if the script is run without `streamlit run`
_mock_session_state: SafeSessionState | None = None


def get_session_state() -> SafeSessionState:
    """Get the SessionState object for the current session.

    Note that in streamlit scripts, this function should not be called
    directly. Instead, SessionState objects should be accessed via
    st.session_state.
    """
    global _state_use_warning_already_displayed  # noqa: PLW0603
    from streamlit.runtime.scriptrunner_utils.script_run_context import (
        get_script_run_ctx,
    )

    ctx = get_script_run_ctx()

    # If there is no script run context because the script is run bare, we
    # use a global mock session state version to allow bare script execution (via python script.py)
    if ctx is None:
        if not _state_use_warning_already_displayed:
            _state_use_warning_already_displayed = True
            if not runtime.exists():
                _LOGGER.warning(
                    "Session state does not function when running a script without `streamlit run`"
                )

        global _mock_session_state  # noqa: PLW0603

        if _mock_session_state is None:
            # Lazy initialize the mock session state
            _mock_session_state = SafeSessionState(SessionState(), lambda: None)
        return _mock_session_state
    return ctx.session_state


# The proxy does not allow getting, setting, or iterating over keyless
# widgets (widgets created without a user_key, which have autogenerated keys).
class SessionStateProxy(MutableMapping[Key, Any]):
    """A dictionary-like object for persisting values across reruns.

    ``st.session_state`` stores per-tab app state. Use it to initialize
    values, share data between widgets and logic, and read or update keyed
    widget values.

    Access values with bracket or attribute notation. Bracket notation
    works for every key. Attribute notation works only when the key is a
    valid Python identifier and does not collide with a dictionary method
    name such as ``items`` or ``get``.

    Widget values live under the widget's ``key``. Assign
    ``st.session_state[key]`` **before** that widget is created, or from an
    ``on_change`` / ``on_click`` callback (callbacks run before widgets are
    instantiated). Assigning after the widget is created on the same run
    raises ``StreamlitWidgetAlreadyInstantiatedError``.

    Some widgets are read-only in session state (buttons and button-like
    triggers, file and media inputs, ``st.data_editor``, ``st.form`` keys,
    and chart or ``ButtonColumn`` selections). Creating one after assigning
    to its key raises ``StreamlitValueAssignmentNotAllowedError``. Store
    values you need to set in a different session state key.

    Examples
    --------
    **Example 1: Initialize and update a value**

    >>> import streamlit as st
    >>>
    >>> st.session_state.setdefault("count", 0)
    >>> if st.button("Increment"):
    ...     st.session_state.count += 1
    >>> st.write(st.session_state.count)

    **Example 2: Set a widget's value before creating it**

    >>> import streamlit as st
    >>>
    >>> st.session_state.setdefault("volume", 50)
    >>> st.slider("Volume", 0, 100, key="volume")

    **Example 3: Store button-driven state in a separate key**

    >>> import streamlit as st
    >>>
    >>> def mark_done():
    ...     st.session_state["done"] = True
    >>>
    >>> st.button("Done", key="done_btn", on_click=mark_done)
    """

    def __iter__(self) -> Iterator[Any]:
        """Iterator over user state and keyed widget values."""
        # TODO: this is unsafe if fastReruns is true! Let's deprecate/remove.
        return iter(get_session_state().filtered_state)

    def __len__(self) -> int:
        """Number of user state and keyed widget values in session_state."""
        return len(get_session_state().filtered_state)

    def __str__(self) -> str:
        """String representation of user state and keyed widget values."""
        return str(get_session_state().filtered_state)

    def __getitem__(self, key: Key) -> Any:
        """Return the state or widget value with the given key.

        Raises
        ------
        StreamlitAPIException
            If the key is not a valid SessionState user key.
        """
        key = str(key)
        require_valid_user_key(key)
        return get_session_state()[key]

    @gather_metrics("session_state.set_item")
    def __setitem__(self, key: Key, value: Any) -> None:
        """Set the value of the given key.

        Raises
        ------
        StreamlitAPIException
            If the key is not a valid SessionState user key.
        StreamlitWidgetAlreadyInstantiatedError
            If the key belongs to a widget or form already instantiated
            during the current script run.
        """
        key = str(key)
        require_valid_user_key(key)
        get_session_state()[key] = value

    def __delitem__(self, key: Key) -> None:
        """Delete the value with the given key.

        Raises
        ------
        StreamlitAPIException
            If the key is not a valid SessionState user key.
        """
        key = str(key)
        require_valid_user_key(key)
        del get_session_state()[key]

    def __getattr__(self, key: str) -> Any:
        try:
            return self[key]
        except KeyError:
            raise AttributeError(_missing_attr_error_message(key))

    @gather_metrics("session_state.set_attr")
    def __setattr__(self, key: str, value: Any) -> None:
        self[key] = value

    def __delattr__(self, key: str) -> None:
        try:
            del self[key]
        except KeyError:
            raise AttributeError(_missing_attr_error_message(key))

    def to_dict(self) -> dict[str, Any]:
        """Return a dict containing all session_state and keyed widget values."""
        return get_session_state().filtered_state


def _missing_attr_error_message(attr_name: str) -> str:
    return (
        f'st.session_state has no attribute "{attr_name}". Did you forget to initialize it? '
        f"More info: https://docs.streamlit.io/develop/concepts/architecture/session-state#initialization"
    )
