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 connection(self):
"""Return a connection to the server, establishing it if necessary."""
if self._connection is None:
try:
self._connection = aioimaplib.IMAP4_SSL(self._server, self._port)
await self._connection.wait_hello_from_server()
await self._connection.login(self._user, self._password)
await self._connection.select(self._folder)
self._does_push = self._connection.has_capability("IDLE")
except (aioimaplib.AioImapException, asyncio.TimeoutError):
self._connection = None
return self._connection
import aioimaplib
while True:
try:
if await self.connection():
await self.refresh_email_count()
await self.async_update_ha_state()
idle = await self._connection.idle_start()
await self._connection.wait_server_push()
self._connection.idle_done()
with async_timeout.timeout(10):
await idle
else:
await self.async_update_ha_state()
except (aioimaplib.AioImapException, asyncio.TimeoutError):
self.disconnected()