Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
response = await self._session.request(
method,
url,
auth=auth,
data=data,
json=json_data,
params=params,
headers=headers,
ssl=self.verify_ssl,
)
except asyncio.TimeoutError as exception:
raise WLEDConnectionTimeoutError(
"Timeout occurred while connecting to WLED device."
) from exception
except (aiohttp.ClientError, socket.gaierror) as exception:
raise WLEDConnectionError(
"Error occurred while communicating with WLED device."
) from exception
content_type = response.headers.get("Content-Type", "")
if (response.status // 100) in [4, 5]:
contents = await response.read()
response.close()
if content_type == "application/json":
raise WLEDError(response.status, json.loads(contents.decode("utf8")))
raise WLEDError(response.status, {"message": contents.decode("utf8")})
if "application/json" in content_type:
data = await response.json()
if (
method == "POST"
async def test_timeout(aresponses):
"""Test request timeout from WLED."""
# Faking a timeout by sleeping
async def response_handler(_):
await asyncio.sleep(0.2)
return aresponses.Response(body="Goodmorning!")
# Backoff will try 3 times
aresponses.add("example.com", "/", "GET", response_handler)
aresponses.add("example.com", "/", "GET", response_handler)
aresponses.add("example.com", "/", "GET", response_handler)
async with aiohttp.ClientSession() as session:
wled = WLED("example.com", session=session, request_timeout=0.1)
with pytest.raises(WLEDConnectionError):
assert await wled._request("/")
"""Exceptions for WLED."""
class WLEDError(Exception):
"""Generic WLED exception."""
class WLEDEmptyResponseError(Exception):
"""WLED empty API response exception."""
class WLEDConnectionError(WLEDError):
"""WLED connection exception."""
class WLEDConnectionTimeoutError(WLEDConnectionError):
"""WLED connection Timeout exception."""
@backoff.on_exception(backoff.expo, WLEDConnectionError, max_tries=3, logger=None)
async def _request(
self,
uri: str = "",
method: str = "GET",
data: Optional[Any] = None,
json_data: Optional[dict] = None,
params: Optional[Mapping[str, str]] = None,
) -> Any:
"""Handle a request to a WLED device."""
scheme = "https" if self.tls else "http"
url = URL.build(
scheme=scheme, host=self.host, port=self.port, path=self.base_path
).join(URL(uri))
auth = None
if self.username and self.password: