"""
This type stub file was generated by pyright.
"""

import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Final, NamedTuple, TYPE_CHECKING
from streamlit.runtime.media_file_manager import MediaFileManager
from streamlit.runtime.session_manager import SessionClient, SessionManager, SessionStorage
from streamlit.runtime.stats import StatsManager
from collections.abc import Awaitable
from streamlit.components.types.base_component_registry import BaseComponentRegistry
from streamlit.proto.BackMsg_pb2 import BackMsg
from streamlit.runtime.caching.storage import CacheStorageManager
from streamlit.runtime.media_file_storage import MediaFileStorage
from streamlit.runtime.uploaded_file_manager import UploadedFileManager

if TYPE_CHECKING:
    ...
SCRIPT_RUN_CHECK_TIMEOUT: Final = ...
_LOGGER: Final = ...
class RuntimeStoppedError(Exception):
    """Raised by operations on a Runtime instance that is stopped."""
    ...


@dataclass(frozen=True)
class RuntimeConfig:
    """Config options for StreamlitRuntime."""
    script_path: str
    command_line: str | None
    media_file_storage: MediaFileStorage
    uploaded_file_manager: UploadedFileManager
    cache_storage_manager: CacheStorageManager = ...
    component_registry: BaseComponentRegistry = ...
    session_manager_class: type[SessionManager] = ...
    session_storage: SessionStorage = ...
    is_hello: bool = ...


class RuntimeState(Enum):
    INITIAL = ...
    NO_SESSIONS_CONNECTED = ...
    ONE_OR_MORE_SESSIONS_CONNECTED = ...
    STOPPING = ...
    STOPPED = ...


class AsyncObjects(NamedTuple):
    """Container for all asyncio objects that Runtime manages.
    These cannot be initialized until the Runtime's eventloop is assigned.
    """
    eventloop: asyncio.AbstractEventLoop
    must_stop: asyncio.Event
    has_connection: asyncio.Event
    need_send_data: asyncio.Event
    started: asyncio.Future[None]
    stopped: asyncio.Future[None]
    ...


class Runtime:
    _instance: Runtime | None = ...
    @classmethod
    def instance(cls) -> Runtime:
        """Return the singleton Runtime instance. Raise an Error if the
        Runtime hasn't been created yet.
        """
        ...
    
    @classmethod
    def exists(cls) -> bool:
        """True if the singleton Runtime instance has been created.

        When a Streamlit app is running in "raw mode" - that is, when the
        app is run via `python app.py` instead of `streamlit run app.py` -
        the Runtime will not exist, and various Streamlit functions need
        to adapt.
        """
        ...
    
    def __init__(self, config: RuntimeConfig) -> None:
        """Create a Runtime instance. It won't be started yet.

        Runtime is *not* thread-safe. Its public methods are generally
        safe to call only on the same thread that its event loop runs on.

        Parameters
        ----------
        config
            Config options.
        """
        ...
    
    @property
    def state(self) -> RuntimeState:
        ...
    
    @property
    def component_registry(self) -> BaseComponentRegistry:
        ...
    
    @property
    def uploaded_file_mgr(self) -> UploadedFileManager:
        ...
    
    @property
    def cache_storage_manager(self) -> CacheStorageManager:
        ...
    
    @property
    def media_file_mgr(self) -> MediaFileManager:
        ...
    
    @property
    def stats_mgr(self) -> StatsManager:
        ...
    
    @property
    def stopped(self) -> Awaitable[None]:
        """A Future that completes when the Runtime's run loop has exited."""
        ...
    
    def get_client(self, session_id: str) -> SessionClient | None:
        """Get the SessionClient for the given session_id, or None
        if no such session exists.

        Notes
        -----
        Threading: SAFE. May be called on any thread.
        """
        ...
    
    def clear_user_info_for_session(self, session_id: str) -> None:
        """Clear the user_info for the given session_id.

        Notes
        -----
        Threading: SAFE. May be called on any thread.
        """
        ...
    
    async def start(self) -> None:
        """Start the runtime. This must be called only once, before
        any other functions are called.

        When this coroutine returns, Streamlit is ready to accept new sessions.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    def stop(self) -> None:
        """Request that Streamlit close all sessions and stop running.
        Note that Streamlit won't stop running immediately.

        Notes
        -----
        Threading: SAFE. May be called from any thread.
        """
        ...
    
    def is_active_session(self, session_id: str) -> bool:
        """True if the session_id belongs to an active session.

        Notes
        -----
        Threading: SAFE. May be called on any thread.
        """
        ...
    
    def connect_session(self, client: SessionClient, user_info: dict[str, str | bool | None], existing_session_id: str | None = ..., session_id_override: str | None = ...) -> str:
        """Create a new session (or connect to an existing one) and return its unique ID.

        Parameters
        ----------
        client
            A concrete SessionClient implementation for communicating with
            the session's client.
        user_info
            A dict that contains information about the session's user. For now,
            it only (optionally) contains the user's email address.

            {
                "email": "example@example.com"
            }
        existing_session_id
            The ID of an existing session to reconnect to. If one is not provided, a new
            session is created. Note that whether the Runtime's SessionManager supports
            reconnecting to an existing session depends on the SessionManager that this
            runtime is configured with.
        session_id_override
            The ID to assign to a new session being created with this method. Setting
            this can be useful when the service that a Streamlit Runtime is running in
            wants to tie the lifecycle of a Streamlit session to some other session-like
            object that it manages. Only one of existing_session_id and
            session_id_override should be set.

        Returns
        -------
        str
            The session's unique string ID.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    def create_session(self, client: SessionClient, user_info: dict[str, str | bool | None], existing_session_id: str | None = ..., session_id_override: str | None = ...) -> str:
        """Create a new session (or connect to an existing one) and return its unique ID.

        Notes
        -----
        This method is simply an alias for connect_session added for backwards
        compatibility.
        """
        ...
    
    def close_session(self, session_id: str) -> None:
        """Close and completely shut down a session.

        This differs from disconnect_session in that it always completely shuts down a
        session, permanently losing any associated state (session state, uploaded files,
        etc.).

        This function may be called multiple times for the same session,
        which is not an error. (Subsequent calls just no-op.)

        Parameters
        ----------
        session_id
            The session's unique ID.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    def disconnect_session(self, session_id: str) -> None:
        """Disconnect a session. It will stop producing ForwardMsgs.

        Differs from close_session because disconnected sessions can be reconnected to
        for a brief window (depending on the SessionManager/SessionStorage
        implementations used by the runtime).

        This function may be called multiple times for the same session,
        which is not an error. (Subsequent calls just no-op.)

        Parameters
        ----------
        session_id
            The session's unique ID.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    def handle_backmsg(self, session_id: str, msg: BackMsg) -> None:
        """Send a BackMsg to an active session.

        Parameters
        ----------
        session_id
            The session's unique ID.
        msg
            The BackMsg to deliver to the session.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    def handle_backmsg_deserialization_exception(self, session_id: str, exc: BaseException) -> None:
        """Handle an Exception raised during deserialization of a BackMsg.

        Parameters
        ----------
        session_id
            The session's unique ID.
        exc
            The Exception.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    
    @property
    async def is_ready_for_browser_connection(self) -> tuple[bool, str]:
        ...
    
    async def does_script_run_without_error(self) -> tuple[bool, str]:
        """Load and execute the app's script to verify it runs without an error.

        Returns
        -------
        (True, "ok") if the script completes without error, or (False, err_msg)
        if the script raises an exception.

        Notes
        -----
        Threading: UNSAFE. Must be called on the eventloop thread.
        """
        ...
    


