Repository: gradio-app/gradio
File: gradio/utils.py (Python)

Code before:
    from gradio.state_holder import SessionState

P = ParamSpec("P")
T = TypeVar("T")


def get_package_version() -> str:
    try:
        package_json_data = (
            pkgutil.get_data(__name__, "package.json").decode("utf-8").strip()  # type: ignore
        )
        package_data = json.loads(package_json_data)
        version = package_data.get("version", "")
        return version
    except Exception:
        return ""


def safe_get_lock() -> asyncio.Lock:
    """Get asyncio.Lock() without fear of getting an Exception.

    Needed because in reload mode we import the Blocks object outside
    the main thread.
    """
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        return asyncio.Lock()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        return asyncio.Lock()


def safe_get_stop_event() -> asyncio.Event:
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        return asyncio.Event()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        return asyncio.Event()


class DynamicBoolean(int):
    def __init__(self, value: int):

Review comment (suggestion):
I think we want to keep the `.is_closed()` check because of https://github.com/gradio-app/gradio/issues/11280, fixed with: https://github.com/gradio-app/gradio/pull/11408:

something like this?

```py
import asyncio

def get_lock():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    if loop.is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    return asyncio.Lock()
```

Code after:
    from gradio.state_holder import SessionState

P = ParamSpec("P")
T = TypeVar("T")


def get_package_version() -> str:
    try:
        package_json_data = (
            pkgutil.get_data(__name__, "package.json").decode("utf-8").strip()  # type: ignore
        )
        package_data = json.loads(package_json_data)
        version = package_data.get("version", "")
        return version
    except Exception:
        return ""


def safe_get_lock() -> asyncio.Lock:
    """Get asyncio.Lock() without fear of getting an Exception.

    Needed because in reload mode we import the Blocks object outside
    the main thread.
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return asyncio.Lock()


def safe_get_stop_event() -> asyncio.Event:
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return asyncio.Event()


class DynamicBoolean(int):
    def __init__(self, value: int):
        self.value = bool(value)

    def __bool__(self):
        return self.value

    def set(self, value: int):
        self.value = bool(value)