src/nominatim_db/utils/asyncio_utils.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2026 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Helper functions to get appropriate asyncio loop factory based on platform.
   9 """
  10 import sys
  11 import asyncio
  12 import selectors
  13 from typing import TypeVar, Coroutine
  14
  15 _T = TypeVar('_T')
  16
  17
  18 def asyncio_run(main: Coroutine[None, None, _T]) -> _T:
  19     """ Execute the coroutine with compatible loop factory and return the result.
  20     """
  21     if sys.platform == "win32":
  22         if sys.version_info >= (3, 12):
  23             def loop_factory() -> asyncio.AbstractEventLoop:
  24                 return asyncio.SelectorEventLoop(selectors.DefaultSelector())
  25             return asyncio.run(main, loop_factory=loop_factory)  # type: ignore[call-arg]
  26         else:
  27             if (asyncio.get_event_loop_policy() != asyncio.WindowsSelectorEventLoopPolicy()):
  28                 asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
  29             return asyncio.run(main)
  30     else:
  31         return asyncio.run(main)