address review: move Instagram token exchange logic to fetch_token method

Author: YueTwo

backend/aci/server/oauth2_manager.py

@@ -114,6 +114,48 @@ async def create_authorization_url(
 
         return str(authorization_url)
 
+    async def exchange_short_lived_token(self, short_lived_token: str) -> dict[str, Any]:
+        """
+        Exchange short-lived access token for long-lived access token.
+        This is specific to Instagram's API requirements.
+
+        Args:
+            short_lived_token: The short-lived access token from the initial OAuth flow
+
+        Returns:
+            Token response dictionary with long-lived access token
+        """
+        if self.app_name != "INSTAGRAM":
+            raise OAuth2Error("Token exchange is only supported for Instagram")
+
+        exchange_token_url = self.custom_data.get(
+            "exchange_token_url", "https://graph.instagram.com/access_token"
+        )
+
+        try:
+            response = await self.oauth2_client.get(
+                exchange_token_url,
+                params={
+                    "grant_type": "ig_exchange_token",
+                    "client_secret": self.client_secret,
+                    "access_token": short_lived_token,
+                },
+                timeout=30.0,
+            )
+            response.raise_for_status()
+
+            token_data = cast(dict[str, Any], response.json())
+            logger.info(
+                f"Successfully exchanged short-lived token for long-lived token, app_name={self.app_name}"
+            )
+            return token_data
+
+        except Exception as e:
+            logger.error(
+                f"Failed to exchange short-lived token, app_name={self.app_name}, error={e}"
+            )
+            raise OAuth2Error("Failed to exchange short-lived token for long-lived token") from e
+
     # TODO: some app may not support "code_verifier"?
     async def fetch_token(
         self,
@@ -143,6 +185,24 @@ async def fetch_token(
                     scope=self.scope,
                 ),
             )
+            # handle Instagram's special case - exchange short-lived token for long-lived token
+            if self.app_name == "INSTAGRAM":
+                if "access_token" in token:
+                    short_lived_token = token["access_token"]
+                    logger.info(
+                        f"Exchanging short-lived token for long-lived token, app_name={self.app_name}"
+                    )
+                    long_lived_token_response = await self.exchange_short_lived_token(
+                        short_lived_token
+                    )
+                    # Update data with long-lived token response: add expires_in and token_type, update access_token
+                    token.update(long_lived_token_response)
+                else:
+                    logger.error(
+                        f"Missing access_token in Instagram OAuth response, app={self.app_name}"
+                    )
+                    raise OAuth2Error("Missing access_token in Instagram OAuth response")
+            # return the token response with long-lived access token
             return token
         except Exception as e:
             logger.error(f"Failed to fetch access token, app_name={self.app_name}, error={e}")
@@ -164,48 +224,6 @@ async def refresh_token(
             logger.error(f"Failed to refresh access token, app_name={self.app_name}, error={e}")
             raise OAuth2Error("Failed to refresh access token") from e
 
-    async def exchange_short_lived_token(self, short_lived_token: str) -> dict[str, Any]:
-        """
-        Exchange short-lived access token for long-lived access token.
-        This is specific to Instagram's API requirements.
-
-        Args:
-            short_lived_token: The short-lived access token from the initial OAuth flow
-
-        Returns:
-            Token response dictionary with long-lived access token
-        """
-        if self.app_name != "INSTAGRAM":
-            raise OAuth2Error("Token exchange is only supported for Instagram")
-
-        exchange_token_url = self.custom_data.get(
-            "exchange_token_url", "https://graph.instagram.com/access_token"
-        )
-
-        try:
-            response = await self.oauth2_client.get(
-                exchange_token_url,
-                params={
-                    "grant_type": "ig_exchange_token",
-                    "client_secret": self.client_secret,
-                    "access_token": short_lived_token,
-                },
-                timeout=30.0,
-            )
-            response.raise_for_status()
-
-            token_data = cast(dict[str, Any], response.json())
-            logger.info(
-                f"Successfully exchanged short-lived token for long-lived token, app_name={self.app_name}"
-            )
-            return token_data
-
-        except Exception as e:
-            logger.error(
-                f"Failed to exchange short-lived token, app_name={self.app_name}, error={e}"
-            )
-            raise OAuth2Error("Failed to exchange short-lived token for long-lived token") from e
-
     async def parse_fetch_token_response(self, token: dict) -> OAuth2SchemeCredentials:
         """
         Parse OAuth2SchemeCredentials from token response with app-specific handling.
@@ -226,22 +244,6 @@ async def parse_fetch_token_response(self, token: dict) -> OAuth2SchemeCredentia
                 logger.error(f"Missing authed_user in Slack OAuth response, app={self.app_name}")
                 raise OAuth2Error("Missing access_token in Slack OAuth response")
 
-        # handle Instagram's special case - exchange short-lived token for long-lived token
-        if self.app_name == "INSTAGRAM":
-            if "access_token" in data:
-                short_lived_token = data["access_token"]
-                logger.info(
-                    f"Exchanging short-lived token for long-lived token, app_name={self.app_name}"
-                )
-                long_lived_token_response = await self.exchange_short_lived_token(short_lived_token)
-                # Update data with long-lived token response: add expires_in and token_type, update access_token
-                data.update(long_lived_token_response)
-            else:
-                logger.error(
-                    f"Missing access_token in Instagram OAuth response, app={self.app_name}"
-                )
-                raise OAuth2Error("Missing access_token in Instagram OAuth response")
-
         if "access_token" not in data:
             logger.error(f"Missing access_token in OAuth response, app={self.app_name}")
             raise OAuth2Error("Missing access_token in OAuth response")