feat(MyPY): WiP

sinch/core/adapters/asyncio_http_adapter.py

@@ -17,13 +17,18 @@ async def request(self, endpoint: HTTPEndpoint) -> SinchBaseModel:
             f" to URL: {request_data_with_auth.url}"
         )
 
+        if request_data_with_auth.auth:
+            auth_data = aiohttp.BasicAuth(request_data_with_auth.auth[0], request_data_with_auth.auth[1])
+        else:
+            auth_data = None
+
         async with aiohttp.ClientSession() as session:
             async with session.request(
                 method=request_data_with_auth.http_method,
                 headers=request_data_with_auth.headers,
                 url=request_data_with_auth.url,
                 data=request_data_with_auth.request_body,
-                auth=request_data_with_auth.auth,
+                auth=auth_data,
                 params=request_data_with_auth.query_params,
                 timeout=aiohttp.ClientTimeout(
                     total=self.sinch.configuration.connection_timeout

sinch/core/adapters/requests_http_transport.py

@@ -1,5 +1,6 @@
 import requests
 import json
+from typing import cast
 from sinch.core.ports.http_transport import HTTPTransport, HttpRequest
 from sinch.core.endpoint import HTTPEndpoint
 from sinch.core.models.http_response import HTTPResponse
@@ -14,7 +15,7 @@ def __init__(self, sinch: ClientBase):
 
     def request(self, endpoint: HTTPEndpoint) -> SinchBaseModel:
         request_data: HttpRequest = self.prepare_request(endpoint)
-        request_data_with_auth: HttpRequest = self.authenticate(endpoint, request_data)
+        request_data_with_auth = cast(HttpRequest, self.authenticate(endpoint, request_data))
 
         self.sinch.configuration.logger.debug(
             f"Sync HTTP {request_data_with_auth.http_method} call with headers:"
@@ -41,11 +42,11 @@ def request(self, endpoint: HTTPEndpoint) -> SinchBaseModel:
             f"and body: {response_body} from URL: {request_data_with_auth.url}"
         )
 
-        return self.handle_response(
+        return cast(SinchBaseModel, self.handle_response(
             endpoint=endpoint,
             http_response=HTTPResponse(
                 status_code=response.status_code,
                 body=response_body,
                 headers=dict(response.headers)
             )
-        )
+        ))

sinch/core/endpoint.py

@@ -1,7 +1,6 @@
 from abc import ABC, abstractmethod
 from sinch.core.models.http_response import HTTPResponse
-from sinch.core.models.http_request import HttpRequest
-from sinch.core.models.base_model import SinchBaseModel
+from sinch.core.models.base_model import SinchBaseModel, SinchRequestBaseModel
 from sinch.core.enums import HTTPAuthentication, HTTPMethod
 from typing import TYPE_CHECKING
 
@@ -13,8 +12,10 @@ class HTTPEndpoint(ABC):
     ENDPOINT_URL: str
     HTTP_METHOD: HTTPMethod
     HTTP_AUTHENTICATION: HTTPAuthentication
+    project_id: str
+    request_data: SinchRequestBaseModel
 
-    def __init__(self, project_id: str, request_data: 'HttpRequest'):
+    def __init__(self, project_id: str, request_data: SinchRequestBaseModel):
         pass
 
     def build_url(self, sinch: 'ClientBase') -> str:

sinch/core/models/http_request.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
@@ -9,4 +10,4 @@ class HttpRequest:
     http_method: str
     request_body: dict
     query_params: dict
-    auth: str
+    auth: Optional[tuple[str, str]]

sinch/core/models/pagination.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+
+from sinch.core.models.base_model import SinchRequestBaseModel, SinchBaseModel
+
+
+@dataclass
+class TokenPaginatedRequest(SinchRequestBaseModel):
+    page_token: str
+
+
+@dataclass
+class IntPaginatedRequest(SinchRequestBaseModel):
+    page: int
+    page_size: int
+
+
+@dataclass
+class TokenPaginatedResponse(SinchBaseModel):
+    next_page_token: str
+
+
+@dataclass
+class IntPaginatedResponse(SinchBaseModel):
+    page: int
+    page_size: int

sinch/core/pagination.py

@@ -1,7 +1,14 @@
 from abc import ABC, abstractmethod
 from sinch.core.endpoint import HTTPEndpoint
-from sinch.core.models.http_response import HTTPResponse
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING, Union, cast, Coroutine, Any
+from sinch.core.models.base_model import SinchBaseModel
+from sinch.core.models.pagination import (
+    TokenPaginatedRequest,
+    TokenPaginatedResponse,
+    IntPaginatedRequest,
+    IntPaginatedResponse
+)
+
 
 if TYPE_CHECKING:
     from sinch.core.clients.sinch_client_base import ClientBase
@@ -21,7 +28,12 @@ class Paginator(ABC):
         if paginated_response.has_next_page:
             paginated_response = paginated_response.next_page()
     """
-    def __init__(self, sinch: 'ClientBase', endpoint: HTTPEndpoint, result: HTTPResponse):
+    def __init__(
+        self,
+        sinch: 'ClientBase',
+        endpoint: HTTPEndpoint,
+        result: Union[TokenPaginatedResponse, IntPaginatedResponse]
+    ):
         self._sinch = sinch
         self.result = result
         self.endpoint = endpoint
@@ -36,7 +48,7 @@ def auto_paging_iter(self) -> Union['PageIterator', 'AsyncPageIterator']:
         pass
 
     @abstractmethod
-    def next_page(self) -> 'Paginator':
+    def next_page(self) -> Union['Paginator', Coroutine[Any, Any, 'Paginator']]:
         pass
 
     @abstractmethod
@@ -45,7 +57,11 @@ def _calculate_next_page(self) -> None:
 
     @classmethod
     @abstractmethod
-    def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'Paginator':
+    def _initialize(
+        cls,
+        sinch: 'ClientBase',
+        endpoint: HTTPEndpoint
+    ) -> Union['Paginator', Coroutine[Any, Any, 'Paginator']]:
         pass
 
 
@@ -58,7 +74,10 @@ def __iter__(self) -> 'PageIterator':
 
     def __next__(self) -> Paginator:
         if self.paginator.has_next_page:
-            return self.paginator.next_page()
+            return cast(
+                Paginator,
+                self.paginator.next_page()
+            )
         else:
             raise StopIteration
 
@@ -70,14 +89,18 @@ def __init__(self, paginator: Paginator):
     def __aiter__(self) -> 'AsyncPageIterator':
         return self
 
-    async def __anext__(self):
+    async def __anext__(self) -> Paginator:
         if self.paginator.has_next_page:
-            return await self.paginator.next_page()
+            return await cast(
+                Coroutine[Any, Any, Paginator],
+                self.paginator.next_page()
+            )
         else:
             raise StopAsyncIteration
 
 
 class IntBasedPaginator(Paginator):
+    result: IntPaginatedResponse
     __doc__ = Paginator.__doc__
 
     def _calculate_next_page(self) -> None:
@@ -86,69 +109,94 @@ def _calculate_next_page(self) -> None:
         else:
             self.has_next_page = False
 
-    def next_page(self):
-        self.endpoint.request_data.page += 1
-        self.result = self._sinch.configuration.transport.request(self.endpoint)
+    def next_page(self) -> Union['Paginator', Coroutine[Any, Any, 'Paginator']]:
+        cast(IntPaginatedRequest, self.endpoint.request_data).page += 1
+        self.result = cast(
+            IntPaginatedResponse,
+            self._sinch.configuration.transport.request(self.endpoint)
+        )
         self._calculate_next_page()
         return self
 
-    def auto_paging_iter(self) -> PageIterator:
+    def auto_paging_iter(self) -> Union['PageIterator', 'AsyncPageIterator']:
         return PageIterator(self)
 
     @classmethod
-    def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'IntBasedPaginator':
+    def _initialize(
+        cls,
+        sinch: 'ClientBase',
+        endpoint: HTTPEndpoint
+    ) -> Union['Paginator', Coroutine[Any, Any, 'Paginator']]:
+
         result = sinch.configuration.transport.request(endpoint)
-        return cls(sinch, endpoint, result)
+        return cls(sinch, endpoint, cast(IntPaginatedResponse, result))
 
 
 class AsyncIntBasedPaginator(IntBasedPaginator):
     __doc__ = IntBasedPaginator.__doc__
-
-    async def next_page(self) -> 'AsyncIntBasedPaginator':
-        self.endpoint.request_data.page += 1
-        self.result = await self._sinch.configuration.transport.request(self.endpoint)
+    result: IntPaginatedResponse
+
+    async def next_page(self) -> 'Paginator':
+        cast(IntPaginatedRequest, self.endpoint.request_data).page += 1
+        self.result = await cast(
+            Coroutine[Any, Any, IntPaginatedResponse],
+            self._sinch.configuration.transport.request(self.endpoint)
+        )
         self._calculate_next_page()
         return self
 
     def auto_paging_iter(self) -> AsyncPageIterator:
         return AsyncPageIterator(self)
 
     @classmethod
-    async def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'AsyncIntBasedPaginator':
-        result = await sinch.configuration.transport.request(endpoint)
+    async def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'Paginator':
+        result = await cast(
+            Coroutine[Any, Any, IntPaginatedResponse],
+            sinch.configuration.transport.request(endpoint)
+        )
         return cls(sinch, endpoint, result)
 
 
 class TokenBasedPaginator(Paginator):
     __doc__ = Paginator.__doc__
+    result: TokenPaginatedResponse
 
     def _calculate_next_page(self) -> None:
         if self.result.next_page_token:
             self.has_next_page = True
         else:
             self.has_next_page = False
 
-    def next_page(self) -> 'TokenBasedPaginator':
-        self.endpoint.request_data.page_token = self.result.next_page_token
-        self.result = self._sinch.configuration.transport.request(self.endpoint)
+    def next_page(self) -> Union['TokenBasedPaginator', Coroutine[Any, Any, 'AsyncTokenBasedPaginator']]:
+        cast(TokenPaginatedRequest, self.endpoint.request_data).page_token = self.result.next_page_token
+        self.result = cast(TokenPaginatedResponse, self._sinch.configuration.transport.request(self.endpoint))
         self._calculate_next_page()
         return self
 
-    def auto_paging_iter(self) -> PageIterator:
+    def auto_paging_iter(self) -> Union[PageIterator, AsyncPageIterator]:
         return PageIterator(self)
 
     @classmethod
-    def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'TokenBasedPaginator':
+    def _initialize(
+        cls,
+        sinch: 'ClientBase',
+        endpoint: HTTPEndpoint
+    ) -> Union['Paginator', Coroutine[Any, Any, 'Paginator']]:
+
         result = sinch.configuration.transport.request(endpoint)
-        return cls(sinch, endpoint, result)
+        return cls(sinch, endpoint, cast(TokenPaginatedResponse, result))
 
 
 class AsyncTokenBasedPaginator(TokenBasedPaginator):
     __doc__ = TokenBasedPaginator.__doc__
+    result: TokenPaginatedResponse
 
     async def next_page(self) -> 'AsyncTokenBasedPaginator':
-        self.endpoint.request_data.page_token = self.result.next_page_token
-        self.result = await self._sinch.configuration.transport.request(self.endpoint)
+        cast(TokenPaginatedRequest, self.endpoint.request_data).page_token = self.result.next_page_token
+        self.result = await cast(
+            Coroutine[Any, Any, TokenPaginatedResponse],
+            self._sinch.configuration.transport.request(self.endpoint)
+        )
         self._calculate_next_page()
         return self
 
@@ -157,5 +205,8 @@ def auto_paging_iter(self) -> AsyncPageIterator:
 
     @classmethod
     async def _initialize(cls, sinch: 'ClientBase', endpoint: HTTPEndpoint) -> 'AsyncTokenBasedPaginator':
-        result = await sinch.configuration.transport.request(endpoint)
+        result = await cast(
+            Coroutine[Any, Any, TokenPaginatedResponse],
+            sinch.configuration.transport.request(endpoint)
+        )
         return cls(sinch, endpoint, result)

sinch/core/ports/http_transport.py

@@ -1,6 +1,5 @@
-import aiohttp
 from abc import ABC, abstractmethod
-from typing import Optional, TYPE_CHECKING, Union, Any, Coroutine
+from typing import TYPE_CHECKING, Union, Any, Coroutine, cast
 from sinch.core.endpoint import HTTPEndpoint
 from sinch.core.models.http_request import HttpRequest
 from sinch.core.models.http_response import HTTPResponse
@@ -20,7 +19,12 @@ def __init__(self, sinch: 'ClientBase'):
     def request(self, endpoint: HTTPEndpoint) -> Union[SinchBaseModel, Coroutine[Any, Any, SinchBaseModel]]:
         pass
 
-    def authenticate(self, endpoint: HTTPEndpoint, request_data: HttpRequest) -> HttpRequest:
+    def authenticate(
+        self,
+        endpoint: HTTPEndpoint,
+        request_data: HttpRequest
+    ) -> Union[HttpRequest, Coroutine[Any, Any, HttpRequest]]:
+
         if endpoint.HTTP_AUTHENTICATION == HTTPAuthentication.BASIC.value:
             request_data.auth = (self.sinch.configuration.key_id, self.sinch.configuration.key_secret)
         else:
@@ -46,22 +50,26 @@ def prepare_request(self, endpoint: HTTPEndpoint) -> HttpRequest:
             http_method=endpoint.HTTP_METHOD.value,
             request_body=endpoint.request_body(),
             query_params=url_query_params,
-            auth=()
+            auth=None
         )
 
-    def handle_response(self, endpoint: HTTPEndpoint, http_response: HTTPResponse) -> SinchBaseModel:
+    def handle_response(
+        self,
+        endpoint: HTTPEndpoint,
+        http_response: HTTPResponse
+    ) -> Union[SinchBaseModel, Coroutine[Any, Any, SinchBaseModel]]:
         if http_response.status_code == 401:
             self.sinch.configuration.token_manager.handle_invalid_token(http_response)
             if self.sinch.configuration.token_manager.token_state == TokenState.EXPIRED:
-                return self.request(endpoint=endpoint)
+                return self.request(endpoint=endpoint)  # type: ignore
 
         return endpoint.handle_response(http_response)
 
 
 class AsyncHTTPTransport(HTTPTransport):
     async def authenticate(self, endpoint: HTTPEndpoint, request_data: HttpRequest) -> HttpRequest:
         if endpoint.HTTP_AUTHENTICATION == HTTPAuthentication.BASIC:
-            request_data.auth = aiohttp.BasicAuth(self.sinch.configuration.key_id, self.sinch.configuration.key_secret)
+            request_data.auth = (self.sinch.configuration.key_id, self.sinch.configuration.key_secret)
         else:
             request_data.auth = None
 
@@ -78,6 +86,6 @@ async def handle_response(self, endpoint: HTTPEndpoint, http_response: HTTPRespo
         if http_response.status_code == 401:
             self.sinch.configuration.token_manager.handle_invalid_token(http_response)
             if self.sinch.configuration.token_manager.token_state == TokenState.EXPIRED:
-                return await self.request(endpoint=endpoint)
+                return await cast(Coroutine[Any, Any, SinchBaseModel], self.request(endpoint=endpoint))
 
         return endpoint.handle_response(http_response)