-""""Connect o a remote Music Assistant Server using the default Websocket API."""
+"""Connect o a remote Music Assistant Server using the default Websocket API."""
from __future__ import annotations
"""Models for providers and plugins in the MA ecosystem."""
+
from __future__ import annotations
import asyncio
DB_TABLE_PROVIDER_MAPPINGS: Final[str] = "provider_mappings"
# all other
-MASS_LOGO_ONLINE: Final[
- str
-] = "https://github.com/home-assistant/brands/raw/master/custom_integrations/mass/icon%402x.png"
+MASS_LOGO_ONLINE: Final[str] = (
+ "https://github.com/home-assistant/brands/raw/master/custom_integrations/mass/icon%402x.png"
+)
ENCRYPT_SUFFIX = "_encrypted_"
SECURE_STRING_SUBSTITUTE = "this_value_is_encrypted"
CONFIGURABLE_CORE_CONTROLLERS = (
@overload
def catch_log_exception(
func: Callable[..., Coroutine[Any, Any, Any]], format_err: Callable[..., Any]
-) -> Callable[..., Coroutine[Any, Any, None]]:
- ...
+) -> Callable[..., Coroutine[Any, Any, None]]: ...
@overload
def catch_log_exception(
func: Callable[..., Any], format_err: Callable[..., Any]
-) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]:
- ...
+) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]: ...
def catch_log_exception(
"""Airplay Player provider for Music Assistant."""
+
from __future__ import annotations
import asyncio
@overload
def soco_error(
errorcodes: None = ...,
-) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]:
- ...
+) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]: ...
@overload
def soco_error(
errorcodes: list[str],
-) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
- ...
+) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: ...
def soco_error(
if tokeninfo is None:
tokeninfo = await self.login()
headers = {"Authorization": f'Bearer {tokeninfo["accessToken"]}'}
- async with self._throttler, self.mass.http_session.get(
- url, headers=headers, params=kwargs, ssl=True, timeout=120
- ) as response:
+ async with (
+ self._throttler,
+ self.mass.http_session.get(
+ url, headers=headers, params=kwargs, ssl=True, timeout=120
+ ) as response,
+ ):
# handle spotify rate limiter
if response.status == 429:
backoff_time = int(response.headers["Retry-After"])
"pytest==8.1.1",
"pytest-aiohttp==1.0.5",
"pytest-cov==4.1.0",
- "ruff==0.2.2",
+ "ruff==0.3.3",
"safety==3.0.1",
]