Skip to content

Commit

Permalink
Remove deprecated code
Browse files Browse the repository at this point in the history
  • Loading branch information
@cgtobi
cgtobi committed Oct 19, 2023
1 parent 73903a4 commit a032919
Show file tree
Hide file tree
Showing 16 changed files with 4 additions and 5,143 deletions.
21 changes: 1 addition & 20 deletions src/pyatmo/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,44 +1,25 @@
"""Expose submodules."""
from pyatmo import const, modules
from pyatmo.account import AsyncAccount
from pyatmo.auth import AbstractAsyncAuth, ClientAuth, NetatmoOAuth2
from pyatmo.camera import AsyncCameraData, CameraData
from pyatmo.auth import AbstractAsyncAuth
from pyatmo.exceptions import ApiError, InvalidHome, InvalidRoom, NoDevice, NoSchedule
from pyatmo.home import Home
from pyatmo.home_coach import AsyncHomeCoachData, HomeCoachData
from pyatmo.modules import Module
from pyatmo.modules.device_types import DeviceType
from pyatmo.public_data import AsyncPublicData, PublicData
from pyatmo.room import Room
from pyatmo.thermostat import AsyncHomeData, AsyncHomeStatus, HomeData, HomeStatus
from pyatmo.weather_station import AsyncWeatherStationData, WeatherStationData

__all__ = [
"AbstractAsyncAuth",
"ApiError",
"AsyncAccount",
"AsyncCameraData",
"AsyncHomeCoachData",
"AsyncHomeData",
"AsyncHomeStatus",
"AsyncPublicData",
"AsyncWeatherStationData",
"CameraData",
"ClientAuth",
"HomeCoachData",
"HomeData",
"HomeStatus",
"InvalidHome",
"InvalidRoom",
"Home",
"Module",
"Room",
"DeviceType",
"NetatmoOAuth2",
"NoDevice",
"NoSchedule",
"PublicData",
"WeatherStationData",
"const",
"modules",
]
276 changes: 0 additions & 276 deletions src/pyatmo/auth.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,21 +3,13 @@

from abc import ABC, abstractmethod
import asyncio
from collections.abc import Callable
from json import JSONDecodeError
import logging
from time import sleep
from typing import Any

from aiohttp import ClientError, ClientResponse, ClientSession, ContentTypeError
from oauthlib.oauth2 import LegacyApplicationClient, TokenExpiredError
import requests
from requests_oauthlib import OAuth2Session

from pyatmo.const import (
ALL_SCOPES,
AUTH_REQ_ENDPOINT,
AUTH_URL_ENDPOINT,
AUTHORIZATION_HEADER,
DEFAULT_BASE_URL,
ERRORS,
Expand All @@ -29,274 +21,6 @@
LOG = logging.getLogger(__name__)


class NetatmoOAuth2:
"""Handle authentication with OAuth2."""

def __init__(
self,
client_id: str,
client_secret: str,
redirect_uri: str | None = None,
token: dict[str, str] | None = None,
token_updater: Callable[[str], None] | None = None,
scope: str | None = "read_station",
user_prefix: str | None = None,
base_url: str = DEFAULT_BASE_URL,
) -> None:
"""Initialize self."""

# Keyword Arguments:
# client_id {str} -- Application client ID delivered by Netatmo on dev.netatmo.com (default: {None})
# client_secret {str} -- Application client secret delivered by Netatmo on dev.netatmo.com (default: {None})
# redirect_uri {Optional[str]} -- Redirect URI where to the authorization server will redirect with an authorization code (default: {None})
# token {Optional[Dict[str, str]]} -- Authorization token (default: {None})
# token_updater {Optional[Callable[[str], None]]} -- Callback when the token is updated (default: {None})
# scope {Optional[str]} -- List of scopes (default: {"read_station"})
# read_station: to retrieve weather station data (Getstationsdata, Getmeasure)
# read_camera: to retrieve Welcome data (Gethomedata, Getcamerapicture)
# access_camera: to access the camera, the videos and the live stream
# write_camera: to set home/away status of persons (Setpersonsaway, Setpersonshome)
# read_thermostat: to retrieve thermostat data (Getmeasure, Getthermostatsdata)
# write_thermostat: to set up the thermostat (Syncschedule, Setthermpoint)
# read_presence: to retrieve Presence data (Gethomedata, Getcamerapicture)
# access_presence: to access the live stream, any video stored on the SD card and to retrieve Presence's lightflood status
# read_homecoach: to retrieve Home Coache data (Gethomecoachsdata)
# read_smokedetector: to retrieve the smoke detector status (Gethomedata)
# Several values can be used at the same time, ie: 'read_station read_camera'
# user_prefix {Optional[str]} -- API prefix for the Netatmo customer
# base_url {str} -- Base URL of the Netatmo API (default: {_DEFAULT_BASE_URL})

self.client_id = client_id
self.client_secret = client_secret
self.redirect_uri = redirect_uri
self.token_updater = token_updater
self.user_prefix = user_prefix
self.base_url = base_url

if token:
self.scope = " ".join(token["scope"])

else:
self.scope = scope or " ".join(ALL_SCOPES)

self.extra = {"client_id": self.client_id, "client_secret": self.client_secret}

self._oauth = OAuth2Session(
client_id=self.client_id,
token=token,
token_updater=self.token_updater,
redirect_uri=self.redirect_uri,
scope=self.scope,
)

def refresh_tokens(self) -> Any:
"""Refresh and return new tokens."""

token = self._oauth.refresh_token(
self.base_url + AUTH_REQ_ENDPOINT,
**self.extra,
)

if self.token_updater is not None:
self.token_updater(token)

return token

def post_api_request(
self,
endpoint: str,
params: dict[str, Any] | None = None,
timeout: int = 5,
) -> requests.Response:
"""Wrap post requests."""

return self.post_request(
url=self.base_url + endpoint,
params=params,
timeout=timeout,
)

def post_request(
self,
url: str,
params: dict[str, Any] | None = None,
timeout: int = 5,
) -> requests.Response:
"""Wrap post requests."""

resp = requests.Response()
req_args = {"data": params if params is not None else {}}

if "json" in req_args["data"]:
req_args["json"] = req_args["data"]["json"]
req_args.pop("data")

if "https://" not in url:
try:
resp = requests.post(url, data=params, timeout=timeout)
except requests.exceptions.ChunkedEncodingError:
LOG.debug("Encoding error when connecting to '%s'", url)
except requests.exceptions.ConnectTimeout:
LOG.debug("Connection to %s timed out", url)
except requests.exceptions.ConnectionError:
LOG.debug("Remote end closed connection without response (%s)", url)

else:

def query(
url: str,
params: dict[str, Any],
timeout: int,
retries: int,
) -> Any:
if retries == 0:
LOG.error("Too many retries")
return requests.Response()

try:
return self._oauth.post(url=url, timeout=timeout, **params)

except (
TokenExpiredError,
requests.exceptions.ReadTimeout,
requests.exceptions.ConnectionError,
):
self._oauth.token = self.refresh_tokens()
# Sleep for 1 sec to prevent authentication related
# timeouts after a token refresh.
sleep(1)
return query(url, params, timeout * 2, retries - 1)

resp = query(url, req_args, timeout, 3)

if resp.status_code is None:
LOG.debug("Resp is None - %s", resp)
return requests.Response()

if not resp.ok:
LOG.debug(
"The Netatmo API returned %s (%s)",
resp.content,
resp.status_code,
)
try:
raise ApiError(
f"{resp.status_code} - "
f"{ERRORS.get(resp.status_code, '')} - "
f"{resp.json()['error']['message']} "
f"({resp.json()['error']['code']}) "
f"when accessing '{url}'",
)

except JSONDecodeError as exc:
raise ApiError(
f"{resp.status_code} - "
f"{ERRORS.get(resp.status_code, '')} - "
f"when accessing '{url}'",
) from exc

if "application/json" in resp.headers.get(
"content-type",
[],
) or resp.content not in [b"", b"None"]:
return resp

return requests.Response()

def get_authorization_url(self, state: str | None = None) -> Any:
"""Return the authorization URL."""

return self._oauth.authorization_url(self.base_url + AUTH_URL_ENDPOINT, state)

def request_token(
self,
authorization_response: str | None = None,
code: str | None = None,
) -> Any:
"""Request token."""

return self._oauth.fetch_token(
self.base_url + AUTH_REQ_ENDPOINT,
authorization_response=authorization_response,
code=code,
client_secret=self.client_secret,
include_client_id=True,
user_prefix=self.user_prefix,
)

def addwebhook(self, webhook_url: str) -> None:
"""Register webhook."""

post_params = {"url": webhook_url}
resp = self.post_api_request(WEBHOOK_URL_ADD_ENDPOINT, post_params)
LOG.debug("addwebhook: %s", resp)

def dropwebhook(self) -> None:
"""Unregister webhook."""

post_params = {"app_types": "app_security"}
resp = self.post_api_request(WEBHOOK_URL_DROP_ENDPOINT, post_params)
LOG.debug("dropwebhook: %s", resp)


class ClientAuth(NetatmoOAuth2):
"""Request authentication and keep access token available through token method."""

# Renew it automatically if necessary
# Args:
# clientId (str): Application clientId delivered by Netatmo on dev.netatmo.com
# clientSecret (str): Application Secret key delivered by Netatmo on dev.netatmo.com
# username (str)
# password (str)
# scope (Optional[str]):
# read_station: to retrieve weather station data (Getstationsdata, Getmeasure)
# read_camera: to retrieve Welcome data (Gethomedata, Getcamerapicture)
# access_camera: to access the camera, the videos and the live stream
# write_camera: to set home/away status of persons (Setpersonsaway, Setpersonshome)
# read_thermostat: to retrieve thermostat data (Getmeasure, Getthermostatsdata)
# write_thermostat: to set up the thermostat (Syncschedule, Setthermpoint)
# read_presence: to retrieve Presence data (Gethomedata, Getcamerapicture)
# access_presence: to access the live stream, any video stored on the SD card and to retrieve Presence's lightflood status
# read_homecoach: to retrieve Home Coache data (Gethomecoachsdata)
# read_smokedetector: to retrieve the smoke detector status (Gethomedata)
# Several value can be used at the same time, ie: 'read_station read_camera'
# user_prefix (Optional[str]) -- API prefix for the Netatmo customer
# base_url (str) -- Base URL of the Netatmo API (default: {_DEFAULT_BASE_URL}).

def __init__(
self,
client_id: str,
client_secret: str,
username: str,
password: str,
scope: str = "read_station",
user_prefix: str | None = None,
base_url: str = DEFAULT_BASE_URL,
) -> None:
"""Initialize self."""

super().__init__(
client_id=client_id,
client_secret=client_secret,
scope=scope,
user_prefix=user_prefix,
base_url=base_url,
)

self._oauth = OAuth2Session(
client=LegacyApplicationClient(client_id=self.client_id),
)
self._oauth.fetch_token(
token_url=self.base_url + AUTH_REQ_ENDPOINT,
username=username,
password=password,
client_id=self.client_id,
client_secret=self.client_secret,
scope=self.scope,
user_prefix=self.user_prefix,
)


class AbstractAsyncAuth(ABC):
"""Abstract class to make authenticated requests."""

Expand Down
Loading

0 comments on commit a032919

Please sign in to comment.