playlist_porter/backend/src/adapters/auth/supabase/client.py at 8dac36d270db343459f5ad2cbefd21d7bf17f86e · mariuspruvot/playlist_porter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from typing import Any

from src.core.config import AuthConfig, SupabaseConfig
from src.infrastructure.logging import get_logger
from supabase import Client, ClientOptions, create_client

logger = get_logger("adapters.auth.supabase.client")


class SupabaseClient:
    """Supabase client wrapper for OAuth authentication operations."""

    def __init__(self, supabase_config: SupabaseConfig, auth_config: AuthConfig):
        self.supabase_config = supabase_config
        self.auth_config = auth_config
        self._client: Client | None = None

    @property
    def client(self) -> Client:
        """Get or create Supabase client."""
        if self._client is None:
            self._client = create_client(
                self.supabase_config.url,
                self.supabase_config.anon_key,
                options=ClientOptions(flow_type="implicit"),
            )
        return self._client

    async def get_oauth_url(
        self, provider: str, redirect_url: str | None = None, scopes: str | None = None
    ) -> str:
        """Get OAuth authorization URL for the specified provider."""
        try:
            # Build OAuth credentials for Supabase
            credentials: dict[str, Any] = {"provider": provider}
            if redirect_url or scopes:
                options: dict[str, Any] = {}
                if redirect_url:
                    options["redirect_to"] = redirect_url
                if scopes:
                    options["scopes"] = scopes
                credentials["options"] = options

            response = self.client.auth.sign_in_with_oauth(credentials)  # type: ignore

            if hasattr(response, "url") and response.url:
                logger.info(f"OAuth URL generated for provider: {provider}")
                return response.url
            else:
                raise Exception("No OAuth URL returned from Supabase")

        except Exception as e:
            logger.error(f"OAuth URL generation failed for {provider}: {str(e)}")
            raise

    async def exchange_oauth_code(
        self, code: str, code_verifier: str | None = None
    ) -> dict:
        """Exchange OAuth authorization code for session."""
        try:
            # Build code exchange parameters
            code_params: dict[str, Any] = {"auth_code": code}
            if code_verifier:
                code_params["code_verifier"] = code_verifier

            response = self.client.auth.exchange_code_for_session(code_params)  # type: ignore
            logger.info("OAuth code exchanged successfully")
            return response.model_dump()
        except Exception as e:
            logger.error(f"OAuth code exchange failed: {str(e)}")
            raise

    async def get_session(self) -> dict | None:
        """Get current session."""
        try:
            session = self.client.auth.get_session()
            if session:
                return session.model_dump()
            return None
        except Exception as e:
            logger.warning(f"Failed to get session: {str(e)}")
            return None

    async def get_user_from_token(self, token: str) -> dict | None:
        """Get user information from session token."""
        try:
            # Set the session with the provided token
            self.client.auth.set_session(token, "")

            # Get the user
            user_response = self.client.auth.get_user()

            if user_response and user_response.user:
                logger.debug(f"Token verified for user: {user_response.user.email}")
                return user_response.user.model_dump()

            return None
        except Exception as e:
            logger.warning(f"Token verification failed: {str(e)}")
            return None

    async def refresh_session(self) -> dict:
        """Refresh current session using stored refresh token."""
        try:
            response = self.client.auth.refresh_session()
            logger.info("Session refreshed successfully")
            return response.model_dump()
        except Exception as e:
            logger.error(f"Session refresh failed: {str(e)}")
            raise

    async def sign_out(self) -> bool:
        """Sign out user and invalidate session."""
        try:
            self.client.auth.sign_out()
            logger.info("User signed out successfully")
            return True
        except Exception as e:
            logger.warning(f"Sign out failed: {str(e)}")
            # Return True anyway - sign out should be permissive
            return True