Create auth.py

auth.py

@@ -0,0 +1,71 @@
+import os
+from typing import Optional, Dict
+from authlib.integrations.requests_client import OAuth2Session
+
+class OAuthManager:
+    """
+    Manages OAuth2 flows for third-party services: GitHub, Google Drive, Slack.
+    """
+    def __init__(self):
+        self.providers: Dict[str, Dict] = {
+            'github': {
+                'client_id': os.getenv('GITHUB_CLIENT_ID'),
+                'client_secret': os.getenv('GITHUB_CLIENT_SECRET'),
+                'authorize_url': 'https://github.com/login/oauth/authorize',
+                'token_url': 'https://github.com/login/oauth/access_token',
+                'scope': 'repo read:org'
+            },
+            'google': {
+                'client_id': os.getenv('GOOGLE_CLIENT_ID'),
+                'client_secret': os.getenv('GOOGLE_CLIENT_SECRET'),
+                'authorize_url': 'https://accounts.google.com/o/oauth2/auth',
+                'token_url': 'https://oauth2.googleapis.com/token',
+                'scope': 'openid email profile https://www.googleapis.com/auth/drive.readonly'
+            },
+            'slack': {
+                'client_id': os.getenv('SLACK_CLIENT_ID'),
+                'client_secret': os.getenv('SLACK_CLIENT_SECRET'),
+                'authorize_url': 'https://slack.com/oauth/v2/authorize',
+                'token_url': 'https://slack.com/api/oauth.v2.access',
+                'scope': 'channels:read chat:write'
+            }
+        }
+
+    def _create_session(self, provider: str, redirect_uri: str) -> OAuth2Session:
+        cfg = self.providers.get(provider)
+        if not cfg or not cfg['client_id'] or not cfg['client_secret']:
+            raise RuntimeError(f"OAuth credentials for '{provider}' are not configured.")
+        return OAuth2Session(
+            cfg['client_id'],
+            cfg['client_secret'],
+            scope=cfg['scope'],
+            redirect_uri=redirect_uri
+        )
+
+    def get_authorization_url(self, provider: str, redirect_uri: str, state: Optional[str] = None) -> (str, str):
+        """
+        Generate the OAuth2 authorization URL and state.
+
+        Returns:
+            (authorization_url, state)
+        """
+        session = self._create_session(provider, redirect_uri)
+        url, state = session.create_authorization_url(self.providers[provider]['authorize_url'], state=state)
+        return url, state
+
+    def fetch_token(self, provider: str, redirect_uri: str, authorization_response: str) -> Dict:
+        """
+        Exchange the authorization response for an access token.
+
+        Returns:
+            Token dict containing access_token, refresh_token, expires_in, etc.
+        """
+        session = self._create_session(provider, redirect_uri)
+        token = session.fetch_token(
+            self.providers[provider]['token_url'],
+            authorization_response=authorization_response
+        )
+        return token
+
+# Instantiate a global manager
+oauth_manager = OAuthManager()