Add type hints for google calendar integration (#65353)
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
parent
390d32c71b
commit
d3374ecd8e
2 changed files with 88 additions and 50 deletions
|
@ -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")
|
||||
|
|
|
@ -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")
|
||||
|
||||
|
|
Loading…
Add table
Reference in a new issue