Add type hints to transmission (#99117)

* Add type hints

* Apply suggestions
This commit is contained in:
Rami Mosleh 2023-08-27 13:43:30 +03:00 committed by GitHub
parent 54cd0e8183
commit 2dd6b26fbc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 12 deletions

View File

@ -1,7 +1,8 @@
"""Support for the Transmission BitTorrent client API.""" """Support for the Transmission BitTorrent client API."""
from __future__ import annotations from __future__ import annotations
from datetime import timedelta from collections.abc import Callable
from datetime import datetime, timedelta
from functools import partial from functools import partial
import logging import logging
import re import re
@ -27,7 +28,11 @@ from homeassistant.const import (
Platform, Platform,
) )
from homeassistant.core import HomeAssistant, ServiceCall, callback from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from homeassistant.helpers import ( from homeassistant.helpers import (
config_validation as cv, config_validation as cv,
entity_registry as er, entity_registry as er,
@ -159,7 +164,9 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return unload_ok return unload_ok
async def get_api(hass, entry): async def get_api(
hass: HomeAssistant, entry: dict[str, Any]
) -> transmission_rpc.Client:
"""Get Transmission client.""" """Get Transmission client."""
host = entry[CONF_HOST] host = entry[CONF_HOST]
port = entry[CONF_PORT] port = entry[CONF_PORT]
@ -205,24 +212,26 @@ def _get_client(hass: HomeAssistant, data: dict[str, Any]) -> TransmissionClient
class TransmissionClient: class TransmissionClient:
"""Transmission Client Object.""" """Transmission Client Object."""
def __init__(self, hass, config_entry): def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Initialize the Transmission RPC API.""" """Initialize the Transmission RPC API."""
self.hass = hass self.hass = hass
self.config_entry = config_entry self.config_entry = config_entry
self.tm_api: transmission_rpc.Client = None self.tm_api: transmission_rpc.Client = None
self._tm_data: TransmissionData = None self._tm_data: TransmissionData | None = None
self.unsub_timer = None self.unsub_timer: Callable[[], None] | None = None
@property @property
def api(self) -> TransmissionData: def api(self) -> TransmissionData:
"""Return the TransmissionData object.""" """Return the TransmissionData object."""
if self._tm_data is None:
raise HomeAssistantError("data not initialized")
return self._tm_data return self._tm_data
async def async_setup(self) -> None: async def async_setup(self) -> None:
"""Set up the Transmission client.""" """Set up the Transmission client."""
try: try:
self.tm_api = await get_api(self.hass, self.config_entry.data) self.tm_api = await get_api(self.hass, dict(self.config_entry.data))
except CannotConnect as error: except CannotConnect as error:
raise ConfigEntryNotReady from error raise ConfigEntryNotReady from error
except (AuthenticationError, UnknownError) as error: except (AuthenticationError, UnknownError) as error:
@ -328,12 +337,12 @@ class TransmissionClient:
self.config_entry, options=options self.config_entry, options=options
) )
def set_scan_interval(self, scan_interval): def set_scan_interval(self, scan_interval: float) -> None:
"""Update scan interval.""" """Update scan interval."""
def refresh(event_time): def refresh(event_time: datetime):
"""Get the latest data from Transmission.""" """Get the latest data from Transmission."""
self._tm_data.update() self.api.update()
if self.unsub_timer is not None: if self.unsub_timer is not None:
self.unsub_timer() self.unsub_timer()

View File

@ -57,7 +57,9 @@ class TransmissionFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return TransmissionOptionsFlowHandler(config_entry) return TransmissionOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None): async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user.""" """Handle a flow initialized by the user."""
errors = {} errors = {}
@ -141,7 +143,9 @@ class TransmissionOptionsFlowHandler(config_entries.OptionsFlow):
"""Initialize Transmission options flow.""" """Initialize Transmission options flow."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input=None): async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the Transmission options.""" """Manage the Transmission options."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)