From d3374ecd8e0b07884713af77e539ebbf36d53881 Mon Sep 17 00:00:00 2001 From: Allen Porter Date: Tue, 1 Feb 2022 08:28:32 -0800 Subject: [PATCH] Add type hints for google calendar integration (#65353) Co-authored-by: Martin Hjelmare --- homeassistant/components/google/__init__.py | 69 ++++++++++++--------- homeassistant/components/google/calendar.py | 69 ++++++++++++++------- 2 files changed, 88 insertions(+), 50 deletions(-) diff --git a/homeassistant/components/google/__init__.py b/homeassistant/components/google/__init__.py index 05bfb490cb8..9f85d41774b 100644 --- a/homeassistant/components/google/__init__.py +++ b/homeassistant/components/google/__init__.py @@ -1,8 +1,10 @@ """Support for Google - Calendar Event Devices.""" +from collections.abc import Mapping from datetime import datetime, timedelta, timezone from enum import Enum import logging import os +from typing import Any from googleapiclient import discovery as google_discovery import httplib2 @@ -158,7 +160,9 @@ ADD_EVENT_SERVICE_SCHEMA = vol.Schema( ) -def do_authentication(hass, hass_config, config): +def do_authentication( + hass: HomeAssistant, hass_config: ConfigType, config: ConfigType +) -> bool: """Notify user of actions and authenticate. Notify user of user_code and verification_url then poll @@ -192,7 +196,7 @@ def do_authentication(hass, hass_config, config): notification_id=NOTIFICATION_ID, ) - def step2_exchange(now): + def step2_exchange(now: datetime) -> None: """Keep trying to validate the user_code until it expires.""" _LOGGER.debug("Attempting to validate user code") @@ -261,7 +265,9 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool: return True -def check_correct_scopes(hass, token_file, config): +def check_correct_scopes( + hass: HomeAssistant, token_file: str, config: ConfigType +) -> bool: """Check for the correct scopes in file.""" creds = Storage(token_file).get() if not creds or not creds.scopes: @@ -270,9 +276,30 @@ def check_correct_scopes(hass, token_file, config): return target_scope in creds.scopes +class GoogleCalendarService: + """Calendar service interface to Google.""" + + def __init__(self, token_file: str) -> None: + """Init the Google Calendar service.""" + self.token_file = token_file + + def get(self) -> google_discovery.Resource: + """Get the calendar service from the storage file token.""" + credentials = Storage(self.token_file).get() + http = credentials.authorize(httplib2.Http()) + service = google_discovery.build( + "calendar", "v3", http=http, cache_discovery=False + ) + return service + + def setup_services( - hass, hass_config, config, track_new_found_calendars, calendar_service -): + hass: HomeAssistant, + hass_config: ConfigType, + config: ConfigType, + track_new_found_calendars: bool, + calendar_service: GoogleCalendarService, +) -> None: """Set up the service listeners.""" def _found_calendar(call: ServiceCall) -> None: @@ -359,10 +386,9 @@ def setup_services( hass.services.register( DOMAIN, SERVICE_ADD_EVENT, _add_event, schema=ADD_EVENT_SERVICE_SCHEMA ) - return True -def do_setup(hass, hass_config, config): +def do_setup(hass: HomeAssistant, hass_config: ConfigType, config: ConfigType) -> None: """Run the setup after we have everything configured.""" _LOGGER.debug("Setting up integration") # Load calendars the user has configured @@ -372,6 +398,7 @@ def do_setup(hass, hass_config, config): track_new_found_calendars = convert( config.get(CONF_TRACK_NEW), bool, DEFAULT_CONF_TRACK_NEW ) + assert track_new_found_calendars is not None setup_services( hass, hass_config, config, track_new_found_calendars, calendar_service ) @@ -381,29 +408,13 @@ def do_setup(hass, hass_config, config): # Look for any new calendars hass.services.call(DOMAIN, SERVICE_SCAN_CALENDARS, None) - return True -class GoogleCalendarService: - """Calendar service interface to Google.""" - - def __init__(self, token_file): - """Init the Google Calendar service.""" - self.token_file = token_file - - def get(self): - """Get the calendar service from the storage file token.""" - credentials = Storage(self.token_file).get() - http = credentials.authorize(httplib2.Http()) - service = google_discovery.build( - "calendar", "v3", http=http, cache_discovery=False - ) - return service - - -def get_calendar_info(hass, calendar): +def get_calendar_info( + hass: HomeAssistant, calendar: Mapping[str, Any] +) -> dict[str, Any]: """Convert data from Google into DEVICE_SCHEMA.""" - calendar_info = DEVICE_SCHEMA( + calendar_info: dict[str, Any] = DEVICE_SCHEMA( { CONF_CAL_ID: calendar["id"], CONF_ENTITIES: [ @@ -420,7 +431,7 @@ def get_calendar_info(hass, calendar): return calendar_info -def load_config(path): +def load_config(path: str) -> dict[str, Any]: """Load the google_calendar_devices.yaml.""" calendars = {} try: @@ -439,7 +450,7 @@ def load_config(path): return calendars -def update_config(path, calendar): +def update_config(path: str, calendar: dict[str, Any]) -> None: """Write the google_calendar_devices.yaml.""" with open(path, "a", encoding="utf8") as out: out.write("\n") diff --git a/homeassistant/components/google/calendar.py b/homeassistant/components/google/calendar.py index 6db9c810ad9..90b5bbb3d89 100644 --- a/homeassistant/components/google/calendar.py +++ b/homeassistant/components/google/calendar.py @@ -2,9 +2,11 @@ from __future__ import annotations import copy -from datetime import timedelta +from datetime import datetime, timedelta import logging +from typing import Any +from googleapiclient import discovery as google_discovery from httplib2 import ServerNotFoundError from homeassistant.components.calendar import ( @@ -72,40 +74,48 @@ def setup_platform( class GoogleCalendarEventDevice(CalendarEventDevice): """A calendar event device.""" - def __init__(self, calendar_service, calendar, data, entity_id): + def __init__( + self, + calendar_service: GoogleCalendarService, + calendar_id: str, + data: dict[str, Any], + entity_id: str, + ) -> None: """Create the Calendar event device.""" self.data = GoogleCalendarData( calendar_service, - calendar, + calendar_id, data.get(CONF_SEARCH), - data.get(CONF_IGNORE_AVAILABILITY), + data.get(CONF_IGNORE_AVAILABILITY, False), ) - self._event = None - self._name = data[CONF_NAME] + self._event: dict[str, Any] | None = None + self._name: str = data[CONF_NAME] self._offset = data.get(CONF_OFFSET, DEFAULT_CONF_OFFSET) self._offset_reached = False self.entity_id = entity_id @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, bool]: """Return the device state attributes.""" return {"offset_reached": self._offset_reached} @property - def event(self): + def event(self) -> dict[str, Any] | None: """Return the next upcoming event.""" return self._event @property - def name(self): + def name(self) -> str: """Return the name of the entity.""" return self._name - async def async_get_events(self, hass, start_date, end_date): + async def async_get_events( + self, hass: HomeAssistant, start_date: datetime, end_date: datetime + ) -> list[dict[str, Any]]: """Get all events in a specific time frame.""" return await self.data.async_get_events(hass, start_date, end_date) - def update(self): + def update(self) -> None: """Update event data.""" self.data.update() event = copy.deepcopy(self.data.event) @@ -120,15 +130,23 @@ class GoogleCalendarEventDevice(CalendarEventDevice): class GoogleCalendarData: """Class to utilize calendar service object to get next event.""" - def __init__(self, calendar_service, calendar_id, search, ignore_availability): + def __init__( + self, + calendar_service: GoogleCalendarService, + calendar_id: str, + search: str | None, + ignore_availability: bool, + ) -> None: """Set up how we are going to search the google calendar.""" self.calendar_service = calendar_service self.calendar_id = calendar_id self.search = search self.ignore_availability = ignore_availability - self.event = None + self.event: dict[str, Any] | None = None - def _prepare_query(self): + def _prepare_query( + self, + ) -> tuple[google_discovery.Resource | None, dict[str, Any] | None]: try: service = self.calendar_service.get() except ServerNotFoundError as err: @@ -143,17 +161,19 @@ class GoogleCalendarData: return service, params - async def async_get_events(self, hass, start_date, end_date): + async def async_get_events( + self, hass: HomeAssistant, start_date: datetime, end_date: datetime + ) -> list[dict[str, Any]]: """Get all events in a specific time frame.""" service, params = await hass.async_add_executor_job(self._prepare_query) - if service is None: + if service is None or params is None: return [] params["timeMin"] = start_date.isoformat("T") params["timeMax"] = end_date.isoformat("T") - event_list = [] + event_list: list[dict[str, Any]] = [] events = await hass.async_add_executor_job(service.events) - page_token = None + page_token: str | None = None while True: page_token = await self.async_get_events_page( hass, events, params, page_token, event_list @@ -162,7 +182,14 @@ class GoogleCalendarData: break return event_list - async def async_get_events_page(self, hass, events, params, page_token, event_list): + async def async_get_events_page( + self, + hass: HomeAssistant, + events: google_discovery.Resource, + params: dict[str, Any], + page_token: str | None, + event_list: list[dict[str, Any]], + ) -> str | None: """Get a page of events in a specific time frame.""" params["pageToken"] = page_token result = await hass.async_add_executor_job(events.list(**params).execute) @@ -177,10 +204,10 @@ class GoogleCalendarData: return result.get("nextPageToken") @Throttle(MIN_TIME_BETWEEN_UPDATES) - def update(self): + def update(self) -> None: """Get the latest data.""" service, params = self._prepare_query() - if service is None: + if service is None or params is None: return params["timeMin"] = dt.now().isoformat("T")