def get_event_loop()

in elastic_agent_client/util/async_tools.py [0:0]

• 14 lines of code
• 5 McCabe index (conditional complexity)

def get_event_loop():
    # activate uvloop if lib is present
    try:
        asyncio.set_event_loop_policy(_get_uvloop().EventLoopPolicy())
    except Exception as e:
        logger.warning(f"Unable to enable uvloop: {e}. Running with default event loop")
        pass
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.get_event_loop_policy().get_event_loop()
        if loop is None:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
    return loop