Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 141 additions & 0 deletions src/auth/src/supabase_auth/_async/gotrue_admin_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,20 @@
AuthMFAAdminDeleteFactorResponse,
AuthMFAAdminListFactorsParams,
AuthMFAAdminListFactorsResponse,
CreateOAuthClientParams,
GenerateLinkParams,
GenerateLinkResponse,
InviteUserByEmailOptions,
OAuthClient,
OAuthClientListResponse,
OAuthClientResponse,
PageParams,
SignOutScope,
User,
UserResponse,
)
from .gotrue_admin_mfa_api import AsyncGoTrueAdminMFAAPI
from .gotrue_admin_oauth_api import AsyncGoTrueAdminOAuthAPI
from .gotrue_base_api import AsyncGoTrueBaseAPI


Expand All @@ -48,6 +54,12 @@ def __init__(
self.mfa = AsyncGoTrueAdminMFAAPI()
self.mfa.list_factors = self._list_factors
self.mfa.delete_factor = self._delete_factor
self.oauth = AsyncGoTrueAdminOAuthAPI()
self.oauth.list_clients = self._list_oauth_clients
self.oauth.create_client = self._create_oauth_client
self.oauth.get_client = self._get_oauth_client
self.oauth.delete_client = self._delete_oauth_client
self.oauth.regenerate_client_secret = self._regenerate_oauth_client_secret

async def sign_out(self, jwt: str, scope: SignOutScope = "global") -> None:
"""
Expand Down Expand Up @@ -200,3 +212,132 @@ async def _delete_factor(
def _validate_uuid(self, id: str) -> None:
if not is_valid_uuid(id):
raise ValueError(f"Invalid id, '{id}' is not a valid uuid")

async def _list_oauth_clients(
self,
params: PageParams = None,
) -> OAuthClientListResponse:
"""
Lists all OAuth clients with optional pagination.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
query = {}
if params:
if params.get("page") is not None:
query["page"] = str(params["page"])
if params.get("per_page") is not None:
query["per_page"] = str(params["per_page"])

response = await self._request(
"GET",
"admin/oauth/clients",
query=query,
no_resolve_json=True,
)

data = response.json()
result = OAuthClientListResponse(
clients=[model_validate(OAuthClient, client) for client in data],
aud=data.get("aud") if isinstance(data, dict) else None,
)

# Parse pagination headers
total = response.headers.get("x-total-count")
if total:
result.total = int(total)

links = response.headers.get("link")
if links:
for link in links.split(","):
parts = link.split(";")
if len(parts) >= 2:
page_match = parts[0].split("page=")
if len(page_match) >= 2:
page_num = int(page_match[1].split("&")[0].rstrip(">"))
rel = parts[1].split("=")[1].strip('"')
if rel == "next":
result.next_page = page_num
elif rel == "last":
result.last_page = page_num

return result

async def _create_oauth_client(
self,
params: CreateOAuthClientParams,
) -> OAuthClientResponse:
"""
Creates a new OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
return await self._request(
"POST",
"admin/oauth/clients",
body=params,
xform=lambda data: OAuthClientResponse(
client=model_validate(OAuthClient, data)
),
)

async def _get_oauth_client(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Gets details of a specific OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
return await self._request(
"GET",
f"admin/oauth/clients/{client_id}",
xform=lambda data: OAuthClientResponse(
client=model_validate(OAuthClient, data)
),
)

async def _delete_oauth_client(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Deletes an OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
return await self._request(
"DELETE",
f"admin/oauth/clients/{client_id}",
xform=lambda data: OAuthClientResponse(
client=model_validate(OAuthClient, data)
),
)

async def _regenerate_oauth_client_secret(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Regenerates the secret for an OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
return await self._request(
"POST",
f"admin/oauth/clients/{client_id}/regenerate_secret",
xform=lambda data: OAuthClientResponse(
client=model_validate(OAuthClient, data)
),
)
78 changes: 78 additions & 0 deletions src/auth/src/supabase_auth/_async/gotrue_admin_oauth_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
from ..types import (
CreateOAuthClientParams,
OAuthClientListResponse,
OAuthClientResponse,
PageParams,
)


class AsyncGoTrueAdminOAuthAPI:
"""
Contains all OAuth client administration methods.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.
"""

async def list_clients(
self,
params: PageParams = None,
) -> OAuthClientListResponse:
"""
Lists all OAuth clients with optional pagination.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
raise NotImplementedError() # pragma: no cover

async def create_client(
self,
params: CreateOAuthClientParams,
) -> OAuthClientResponse:
"""
Creates a new OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
raise NotImplementedError() # pragma: no cover

async def get_client(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Gets details of a specific OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
raise NotImplementedError() # pragma: no cover

async def delete_client(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Deletes an OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
raise NotImplementedError() # pragma: no cover

async def regenerate_client_secret(
self,
client_id: str,
) -> OAuthClientResponse:
"""
Regenerates the secret for an OAuth client.
Only relevant when the OAuth 2.1 server is enabled in Supabase Auth.

This function should only be called on a server.
Never expose your `service_role` key in the browser.
"""
raise NotImplementedError() # pragma: no cover
Loading
Loading