generalizes OAuth2 implementation. Provides example for Zoom Video co…

…mmunications

dlt/sources/helpers/rest_client/auth.py

@@ -1,4 +1,5 @@
 import math
+from abc import abstractmethod
 from base64 import b64encode
 from typing import (
     TYPE_CHECKING,
@@ -151,7 +152,7 @@ class OAuth2ImplicitFlow(OAuth2AuthBase):
     def __init__(
         self,
         access_token_url: str,
-        access_token_request_data,
+        access_token_request_data: Dict[str, Any],
         client_id: TSecretStrValue,
         client_secret: TSecretStrValue,
         access_token: TSecretStrValue = None,
@@ -174,21 +175,34 @@ def __call__(self, request: PreparedRequest) -> PreparedRequest:
     def is_token_expired(self) -> bool:
         return pendulum.now() >= self.token_expiry
 
+    @abstractmethod
+    def build_access_token_request(self) -> Dict[str, Any]:
+        pass
+
     def obtain_token(self) -> None:
+        response = requests.post(**self.build_access_token_request())
+        response.raise_for_status()
+        self.access_token = response.json()["access_token"]
+        expires_in = response.json().get("expires_in", self.default_token_expiration)
+        self.token_expiry = pendulum.now().add(seconds=expires_in)
+
+
+class OAuth2Zoom(OAuth2ImplicitFlow):
+    def build_access_token_request(self) -> Dict[str, Any]:
+        """
+        This b64-encoded access token request is specific to Zoom.
+        Many other APIs implement OAuth2 differently
+        """
         authentication: str = b64encode(f"{self.client_id}:{self.client_secret}".encode()).decode()
 
-        response = requests.post(
-            url=self.access_token_url,
-            headers={
+        return {
+            "url": self.access_token_url,
+            "headers": {
                 "Authorization": f"Basic {authentication}",
                 "Content-Type": "application/x-www-form-urlencoded",
             },
-            data=self.access_token_request_data,
-        )
-        response.raise_for_status()
-        self.access_token = response.json()["access_token"]
-        expires_in = response.json().get("expires_in", self.default_token_expiration)
-        self.token_expiry = pendulum.now().add(seconds=expires_in)
+            "data": self.access_token_request_data,
+        }
 
 
 @configspec