Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
websocket = WebsocketEchoClient(login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email),
websocket)
await websocket.async_run()
except BaseException as exception_:
_LOGGER.debug("%s: Websocket creation failed: %s",
hide_email(email),
exception_)
return
return websocket
def ws_connect():
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
from alexapy import WebsocketEchoClient
websocket = None
try:
websocket = WebsocketEchoClient(login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email),
websocket)
except BaseException as exception_:
_LOGGER.debug("%s: Websocket creation failed: %s",
hide_email(email),
exception_)
return websocket
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
if login_obj.session.closed:
_LOGGER.debug(
"%s: Websocket creation aborted. Session is closed.",
hide_email(email),
)
return
websocket = WebsocketEchoClient(
login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler,
)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email), websocket)
await websocket.async_run()
except BaseException as exception_: # pylint: disable=broad-except
_LOGGER.debug(
"%s: Websocket creation failed: %s", hide_email(email), exception_
)
return
return websocket
def ws_connect():
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
from alexapy import WebsocketEchoClient
try:
websocket = WebsocketEchoClient(login_obj,
ws_handler,
ws_close_handler,
ws_error_handler)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email),
websocket)
except BaseException as exception_:
_LOGGER.debug("%s: Websocket failed: %s falling back to polling",
hide_email(email),
exception_)
websocket = None
return websocket
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
if login_obj.session.closed:
_LOGGER.debug(
"%s: Websocket creation aborted. Session is closed.",
hide_email(email),
)
return
websocket = WebsocketEchoClient(
login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler,
)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email), websocket)
await websocket.async_run()
except BaseException as exception_: # pylint: disable=broad-except
_LOGGER.debug(
"%s: Websocket creation failed: %s", hide_email(email), exception_
)
return
return websocket
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
websocket = WebsocketEchoClient(login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email),
websocket)
await websocket.async_run()
except BaseException as exception_:
_LOGGER.debug("%s: Websocket creation failed: %s",
hide_email(email),
exception_)
return websocket
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
if login_obj.session.closed:
_LOGGER.debug(
"%s: Websocket creation aborted. Session is closed.",
hide_email(email),
)
return
websocket = WebsocketEchoClient(
login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler,
)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email), websocket)
await websocket.async_run()
except BaseException as exception_: # pylint: disable=broad-except
_LOGGER.debug(
"%s: Websocket creation failed: %s", hide_email(email), exception_
)
return
return websocket
async def ws_connect() -> WebsocketEchoClient:
"""Open WebSocket connection.
This will only attempt one login before failing.
"""
websocket: Optional[WebsocketEchoClient] = None
try:
websocket = WebsocketEchoClient(login_obj,
ws_handler,
ws_open_handler,
ws_close_handler,
ws_error_handler)
_LOGGER.debug("%s: Websocket created: %s", hide_email(email),
websocket)
await websocket.async_run()
except BaseException as exception_:
_LOGGER.debug("%s: Websocket creation failed: %s",
hide_email(email),
exception_)
return
return websocket