Add docstrings to authentication strategies and improve API request method comments

Author: iMicknl

pyoverkiz/auth/__init__.py

@@ -1,3 +1,5 @@
+"""Authentication module for pyoverkiz."""
+
 from __future__ import annotations
 
 from pyoverkiz.auth.base import AuthContext, AuthStrategy

pyoverkiz/auth/base.py

@@ -1,3 +1,5 @@
+"""Base classes for authentication strategies."""
+
 from __future__ import annotations
 
 import datetime
@@ -8,11 +10,14 @@
 
 @dataclass(slots=True)
 class AuthContext:
+    """Authentication context holding tokens and expiration."""
+
     access_token: str | None = None
     refresh_token: str | None = None
     expires_at: datetime.datetime | None = None
 
     def is_expired(self, *, skew_seconds: int = 5) -> bool:
+        """Check if the access token is expired, considering a skew time."""
         if not self.expires_at:
             return False
 
@@ -22,10 +27,16 @@ def is_expired(self, *, skew_seconds: int = 5) -> bool:
 
 
 class AuthStrategy(Protocol):
-    async def login(self) -> None: ...
+    """Protocol for authentication strategies."""
+
+    async def login(self) -> None:
+        """Perform login to obtain tokens."""
 
-    async def refresh_if_needed(self) -> bool: ...
+    async def refresh_if_needed(self) -> bool:
+        """Refresh tokens if they are expired. Return True if refreshed."""
 
-    def auth_headers(self, path: str | None = None) -> Mapping[str, str]: ...
+    def auth_headers(self, path: str | None = None) -> Mapping[str, str]:
+        """Generate authentication headers for requests."""
 
-    async def close(self) -> None: ...
+    async def close(self) -> None:
+        """Clean up any resources held by the strategy."""

pyoverkiz/auth/credentials.py

@@ -1,3 +1,5 @@
+"""Credentials for authentication strategies."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass
@@ -9,20 +11,27 @@ class Credentials:
 
 @dataclass(slots=True)
 class UsernamePasswordCredentials(Credentials):
+    """Credentials using username and password."""
+
     username: str
     password: str
 
 
 @dataclass(slots=True)
 class TokenCredentials(Credentials):
+    """Credentials using an (API) token."""
+
     token: str
 
 
 @dataclass(slots=True)
-class LocalTokenCredentials(TokenCredentials): ...
+class LocalTokenCredentials(TokenCredentials):
+    """Credentials using a local API token."""
 
 
 @dataclass(slots=True)
 class RexelOAuthCodeCredentials(Credentials):
+    """ "Credentials using Rexel OAuth2 authorization code."""
+
     code: str
     redirect_uri: str

pyoverkiz/auth/factory.py

@@ -1,3 +1,5 @@
+"""Factory to build authentication strategies based on server and credentials."""
+
 from __future__ import annotations
 
 import ssl

pyoverkiz/auth/strategies.py

@@ -1,3 +1,5 @@
+"""Authentication strategies for Overkiz API."""
+
 from __future__ import annotations
 
 import asyncio
@@ -11,6 +13,7 @@
 
 import boto3
 from aiohttp import ClientSession, FormData
+from botocore.client import BaseClient
 from botocore.config import Config
 from warrant_lite import WarrantLite
 
@@ -51,6 +54,8 @@
 
 
 class BaseAuthStrategy(AuthStrategy):
+    """Base class for authentication strategies."""
+
     def __init__(
         self,
         session: ClientSession,
@@ -82,6 +87,8 @@ async def close(self) -> None:
 
 
 class SessionLoginStrategy(BaseAuthStrategy):
+    """Authentication strategy using session-based login."""
+
     def __init__(
         self,
         credentials: UsernamePasswordCredentials,
@@ -90,17 +97,20 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize SessionLoginStrategy with given parameters."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 
     async def login(self) -> None:
+        """Perform login using username and password."""
         payload = {
             "userId": self.credentials.username,
             "userPassword": self.credentials.password,
         }
         await self._post_login(payload)
 
     async def _post_login(self, data: Mapping[str, Any]) -> None:
+        """Post login data to the server and handle response."""
         async with self.session.post(
             f"{self.server.endpoint}login",
             data=data,
@@ -117,6 +127,8 @@ async def _post_login(self, data: Mapping[str, Any]) -> None:
 
 
 class SomfyAuthStrategy(BaseAuthStrategy):
+    """Authentication strategy using Somfy OAuth2."""
+
     def __init__(
         self,
         credentials: UsernamePasswordCredentials,
@@ -125,11 +137,13 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize SomfyAuthStrategy with given parameters."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
         self.context = AuthContext()
 
     async def login(self) -> None:
+        """Perform login using Somfy OAuth2."""
         await self._request_access_token(
             grant_type="password",
             extra_fields={
@@ -139,6 +153,7 @@ async def login(self) -> None:
         )
 
     async def refresh_if_needed(self) -> bool:
+        """Refresh Somfy OAuth2 tokens if needed."""
         if not self.context.is_expired() or not self.context.refresh_token:
             return False
 
@@ -149,6 +164,7 @@ async def refresh_if_needed(self) -> bool:
         return True
 
     def auth_headers(self, path: str | None = None) -> Mapping[str, str]:
+        """Return authentication headers for a request path."""
         if self.context.access_token:
             return {"Authorization": f"Bearer {self.context.access_token}"}
 
@@ -190,6 +206,8 @@ async def _request_access_token(
 
 
 class CozytouchAuthStrategy(SessionLoginStrategy):
+    """Authentication strategy using Cozytouch session-based login."""
+
     def __init__(
         self,
         credentials: UsernamePasswordCredentials,
@@ -198,9 +216,11 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize CozytouchAuthStrategy with given parameters."""
         super().__init__(credentials, session, server, ssl_context, api_type)
 
     async def login(self) -> None:
+        """Perform login using Cozytouch username and password."""
         form = FormData(
             {
                 "grant_type": "password",
@@ -239,6 +259,8 @@ async def login(self) -> None:
 
 
 class NexityAuthStrategy(SessionLoginStrategy):
+    """Authentication strategy using Nexity session-based login."""
+
     def __init__(
         self,
         credentials: UsernamePasswordCredentials,
@@ -247,12 +269,14 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize NexityAuthStrategy with given parameters."""
         super().__init__(credentials, session, server, ssl_context, api_type)
 
     async def login(self) -> None:
+        """Perform login using Nexity username and password."""
         loop = asyncio.get_event_loop()
 
-        def _client() -> boto3.session.Session.client:
+        def _client() -> BaseClient:
             return boto3.client(
                 "cognito-idp", config=Config(region_name=NEXITY_COGNITO_REGION)
             )
@@ -287,6 +311,8 @@ def _client() -> boto3.session.Session.client:
 
 
 class LocalTokenAuthStrategy(BaseAuthStrategy):
+    """Authentication strategy using a local API token."""
+
     def __init__(
         self,
         credentials: LocalTokenCredentials,
@@ -295,18 +321,23 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize LocalTokenAuthStrategy with given parameters."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 
     async def login(self) -> None:
+        """Validate that a token is provided for local API access."""
         if not self.credentials.token:
             raise InvalidTokenException("Local API requires a token.")
 
     def auth_headers(self, path: str | None = None) -> Mapping[str, str]:
+        """Return authentication headers for a request path."""
         return {"Authorization": f"Bearer {self.credentials.token}"}
 
 
 class RexelAuthStrategy(BaseAuthStrategy):
+    """Authentication strategy using Rexel OAuth2."""
+
     def __init__(
         self,
         credentials: RexelOAuthCodeCredentials,
@@ -315,11 +346,13 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize RexelAuthStrategy with given parameters."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
         self.context = AuthContext()
 
     async def login(self) -> None:
+        """Perform login using Rexel OAuth2 authorization code."""
         await self._exchange_token(
             {
                 "grant_type": "authorization_code",
@@ -331,6 +364,7 @@ async def login(self) -> None:
         )
 
     async def refresh_if_needed(self) -> bool:
+        """ "Refresh Rexel OAuth2 tokens if needed."""
         if not self.context.is_expired() or not self.context.refresh_token:
             return False
 
@@ -345,11 +379,13 @@ async def refresh_if_needed(self) -> bool:
         return True
 
     def auth_headers(self, path: str | None = None) -> Mapping[str, str]:
+        """Return authentication headers for a request path."""
         if self.context.access_token:
             return {"Authorization": f"Bearer {self.context.access_token}"}
         return {}
 
     async def _exchange_token(self, payload: Mapping[str, str]) -> None:
+        """Exchange authorization code or refresh token for access token."""
         form = FormData(payload)
         async with self.session.post(
             REXEL_OAUTH_TOKEN_URL,
@@ -373,6 +409,7 @@ async def _exchange_token(self, payload: Mapping[str, str]) -> None:
 
     @staticmethod
     def _ensure_consent(access_token: str) -> None:
+        """Ensure that the Rexel token has the required consent."""
         payload = _decode_jwt_payload(access_token)
         consent = payload.get("consent")
         if consent != REXEL_REQUIRED_CONSENT:
@@ -382,6 +419,8 @@ def _ensure_consent(access_token: str) -> None:
 
 
 class BearerTokenAuthStrategy(BaseAuthStrategy):
+    """Authentication strategy using a static bearer token."""
+
     def __init__(
         self,
         credentials: TokenCredentials,
@@ -390,16 +429,19 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
+        """Initialize BearerTokenAuthStrategy with given parameters."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 
     def auth_headers(self, path: str | None = None) -> Mapping[str, str]:
+        """Return authentication headers for a request path."""
         if self.credentials.token:
             return {"Authorization": f"Bearer {self.credentials.token}"}
         return {}
 
 
 def _decode_jwt_payload(token: str) -> dict[str, Any]:
+    """Decode the payload of a JWT token."""
     parts = token.split(".")
     if len(parts) < 2:
         raise InvalidTokenException("Malformed JWT received.")

pyoverkiz/client.py

@@ -215,7 +215,7 @@ def _resolve_server_key(self) -> Server:
                 return Server(key)
 
         if self.api_type == APIType.LOCAL:
-            return Server.SOMFY_DEVELOPER_MODE
+            return Server(Server.SOMFY_DEVELOPER_MODE)
 
         raise OverkizException(
             "Unknown server configuration; provide server_key explicitly."