Lazy-import boto3 in NexityAuthStrategy (#1990)

## Summary
- Adds `AuthContext.update_from_token` for shared OAuth token handling
across auth strategies
- Lazy-imports `boto3` and `warrant-lite` in NexityAuthStrategy so
they're only loaded when actually needed
- Users who don't use Nexity no longer pay the boto3 import cost

## Test plan
- [x] Nexity auth tests pass
- [x] Other auth strategies unaffected
- [x] Verify boto3 is not imported at module load time

Author: iMicknl

pyoverkiz/auth/base.py

@@ -5,7 +5,7 @@
 import datetime
 from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Protocol
+from typing import Any, Protocol
 
 
 @dataclass(slots=True)
@@ -25,6 +25,18 @@ def is_expired(self, *, skew_seconds: int = 5) -> bool:
             datetime.UTC
         ) >= self.expires_at - datetime.timedelta(seconds=skew_seconds)
 
+    def update_from_token(self, token: dict[str, Any]) -> None:
+        """Update context from an OAuth token response."""
+        self.access_token = str(token["access_token"])
+        self.refresh_token = (
+            str(token["refresh_token"]) if "refresh_token" in token else None
+        )
+        expires_in = token.get("expires_in")
+        if expires_in is not None:
+            self.expires_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(
+                seconds=int(expires_in) - 5
+            )
+
 
 class AuthStrategy(Protocol):
     """Protocol for authentication strategies."""

pyoverkiz/auth/strategies.py

@@ -5,18 +5,15 @@
 import asyncio
 import base64
 import binascii
-import datetime
 import json
 import ssl
 from collections.abc import Mapping
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
+
+if TYPE_CHECKING:
+    from botocore.client import BaseClient
 
-import boto3
 from aiohttp import ClientSession, FormData
-from botocore.client import BaseClient
-from botocore.config import Config
-from botocore.exceptions import ClientError
-from warrant_lite import WarrantLite
 
 from pyoverkiz.auth.base import AuthContext, AuthStrategy
 from pyoverkiz.auth.credentials import (
@@ -98,7 +95,7 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
-        """Initialize SessionLoginStrategy with given parameters."""
+        """Create a session-login strategy bound to the given credentials."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 
@@ -142,7 +139,7 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
-        """Initialize SomfyAuthStrategy with given parameters."""
+        """Create a Somfy OAuth2 strategy with a fresh auth context."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
         self.context = AuthContext()
@@ -201,13 +198,7 @@ async def _request_access_token(
             if not access_token:
                 raise SomfyServiceError("No Somfy access token provided.")
 
-            self.context.access_token = cast(str, access_token)
-            self.context.refresh_token = token.get("refresh_token")
-            expires_in = token.get("expires_in")
-            if expires_in:
-                self.context.expires_at = datetime.datetime.now(
-                    datetime.UTC
-                ) + datetime.timedelta(seconds=cast(int, expires_in) - 5)
+            self.context.update_from_token(token)
 
 
 class CozytouchAuthStrategy(SessionLoginStrategy):
@@ -257,6 +248,11 @@ class NexityAuthStrategy(SessionLoginStrategy):
 
     async def login(self) -> None:
         """Perform login using Nexity username and password."""
+        import boto3
+        from botocore.config import Config
+        from botocore.exceptions import ClientError
+        from warrant_lite import WarrantLite
+
         loop = asyncio.get_running_loop()
 
         def _client() -> BaseClient:
@@ -307,7 +303,7 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
-        """Initialize LocalTokenAuthStrategy with given parameters."""
+        """Create a local-token strategy bound to the given credentials."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 
@@ -332,7 +328,7 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
-        """Initialize RexelAuthStrategy with given parameters."""
+        """Create a Rexel OAuth2 strategy with a fresh auth context."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
         self.context = AuthContext()
@@ -395,13 +391,7 @@ async def _exchange_token(self, payload: Mapping[str, str]) -> None:
                 raise InvalidTokenError("No Rexel access token provided.")
 
             self._ensure_consent(access_token)
-            self.context.access_token = cast(str, access_token)
-            self.context.refresh_token = token.get("refresh_token")
-            expires_in = token.get("expires_in")
-            if expires_in:
-                self.context.expires_at = datetime.datetime.now(
-                    datetime.UTC
-                ) + datetime.timedelta(seconds=cast(int, expires_in) - 5)
+            self.context.update_from_token(token)
 
     @staticmethod
     def _ensure_consent(access_token: str) -> None:
@@ -423,7 +413,7 @@ def __init__(
         ssl_context: ssl.SSLContext | bool,
         api_type: APIType,
     ) -> None:
-        """Initialize BearerTokenAuthStrategy with given parameters."""
+        """Create a bearer-token strategy bound to the given credentials."""
         super().__init__(session, server, ssl_context, api_type)
         self.credentials = credentials
 

tests/test_auth.py

@@ -8,6 +8,7 @@
 import base64
 import datetime
 import json
+import sys
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -491,6 +492,26 @@ def test_auth_headers_with_token(self):
 class TestNexityAuthStrategy:
     """Tests for Nexity auth error mapping behavior."""
 
+    def test_boto3_not_imported_at_module_load(self):
+        """Verify boto3 and warrant_lite are lazy-imported, not at module load."""
+        saved = {}
+        for mod in ("boto3", "botocore", "warrant_lite"):
+            saved[mod] = sys.modules.pop(mod, None)
+
+        try:
+            import importlib
+
+            import pyoverkiz.auth.strategies
+
+            importlib.reload(pyoverkiz.auth.strategies)
+
+            assert "boto3" not in sys.modules
+            assert "warrant_lite" not in sys.modules
+        finally:
+            for mod, value in saved.items():
+                if value is not None:
+                    sys.modules[mod] = value
+
     @pytest.mark.asyncio
     async def test_login_maps_invalid_credentials_client_error(self):
         """Map Cognito bad-credential errors to NexityBadCredentialsError."""
@@ -512,10 +533,8 @@ async def test_login_maps_invalid_credentials_client_error(self):
         warrant_instance.authenticate_user.side_effect = bad_credentials_error
 
         with (
-            patch("pyoverkiz.auth.strategies.boto3.client", return_value=MagicMock()),
-            patch(
-                "pyoverkiz.auth.strategies.WarrantLite", return_value=warrant_instance
-            ),
+            patch("boto3.client", return_value=MagicMock()),
+            patch("warrant_lite.WarrantLite", return_value=warrant_instance),
             pytest.raises(NexityBadCredentialsError),
         ):
             strategy = NexityAuthStrategy(
@@ -544,10 +563,8 @@ async def test_login_propagates_non_auth_client_error(self):
         warrant_instance.authenticate_user.side_effect = service_error
 
         with (
-            patch("pyoverkiz.auth.strategies.boto3.client", return_value=MagicMock()),
-            patch(
-                "pyoverkiz.auth.strategies.WarrantLite", return_value=warrant_instance
-            ),
+            patch("boto3.client", return_value=MagicMock()),
+            patch("warrant_lite.WarrantLite", return_value=warrant_instance),
             pytest.raises(ClientError, match="InternalErrorException"),
         ):
             strategy = NexityAuthStrategy(