feat: add create_user and list_users

supabase/auth-js#166

gotrue/_async/api.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
+from ..exceptions import APIError
 from ..helpers import check_response, encode_uri_component
 from ..http_clients import AsyncClient
 from ..types import (
@@ -38,6 +39,60 @@ async def __aexit__(self, exc_t, exc_v, exc_tb) -> None:
     async def close(self) -> None:
         await self.http_client.aclose()
 
+    async def create_user(self, *, attributes: UserAttributes) -> User:
+        """Creates a new user.
+
+        This function should only be called on a server.
+        Never expose your `service_role` key in the browser.
+
+        Parameters
+        ----------
+        attributes: UserAttributes
+            The data you want to create the user with.
+
+        Returns
+        -------
+        response : User
+            The created user
+
+        Raises
+        ------
+        error : APIError
+            If an error occurs
+        """
+        headers = self.headers
+        data = attributes.dict()
+        url = f"{self.url}/admin/users"
+        response = await self.http_client.post(url, json=data, headers=headers)
+        return User.parse_response(response)
+
+    async def list_users(self) -> List[User]:
+        """Get a list of users.
+
+        This function should only be called on a server.
+        Never expose your `service_role` key in the browser.
+
+        Returns
+        -------
+        response : List[User]
+            A list of users
+
+        Raises
+        ------
+        error : APIError
+            If an error occurs
+        """
+        headers = self.headers
+        url = f"{self.url}/admin/users"
+        response = await self.http_client.get(url, headers=headers)
+        check_response(response)
+        users = response.json().get("users")
+        if users is None:
+            raise APIError(f"No users found in response", 400)
+        if not isinstance(users, list):
+            raise APIError(f"Expected a list of users", 400)
+        return [User.parse_obj(response) for response in users]
+
     async def sign_up_with_email(
         self,
         *,

gotrue/_sync/api.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
+from ..exceptions import APIError
 from ..helpers import check_response, encode_uri_component
 from ..http_clients import SyncClient
 from ..types import (
@@ -38,6 +39,60 @@ def __exit__(self, exc_t, exc_v, exc_tb) -> None:
     def close(self) -> None:
         self.http_client.aclose()
 
+    def create_user(self, *, attributes: UserAttributes) -> User:
+        """Creates a new user.
+
+        This function should only be called on a server.
+        Never expose your `service_role` key in the browser.
+
+        Parameters
+        ----------
+        attributes: UserAttributes
+            The data you want to create the user with.
+
+        Returns
+        -------
+        response : User
+            The created user
+
+        Raises
+        ------
+        error : APIError
+            If an error occurs
+        """
+        headers = self.headers
+        data = attributes.dict()
+        url = f"{self.url}/admin/users"
+        response = self.http_client.post(url, json=data, headers=headers)
+        return User.parse_response(response)
+
+    def list_users(self) -> List[User]:
+        """Get a list of users.
+
+        This function should only be called on a server.
+        Never expose your `service_role` key in the browser.
+
+        Returns
+        -------
+        response : List[User]
+            A list of users
+
+        Raises
+        ------
+        error : APIError
+            If an error occurs
+        """
+        headers = self.headers
+        url = f"{self.url}/admin/users"
+        response = self.http_client.get(url, headers=headers)
+        check_response(response)
+        users = response.json().get("users")
+        if users is None:
+            raise APIError(f"No users found in response", 400)
+        if not isinstance(users, list):
+            raise APIError(f"Expected a list of users", 400)
+        return [User.parse_obj(response) for response in users]
+
     def sign_up_with_email(
         self,
         *,

infra/docker-compose.yml

@@ -61,6 +61,11 @@ services:
       DATABASE_URL: 'postgres://postgres:postgres@db:5432/postgres?sslmode=disable'
       GOTRUE_EXTERNAL_PHONE_ENABLED: 'true'
       GOTRUE_SMS_AUTOCONFIRM: 'true'
+      GOTRUE_SMTP_HOST: mail
+      GOTRUE_SMTP_PORT: 2500
+      GOTRUE_SMTP_USER: GOTRUE_SMTP_USER
+      GOTRUE_SMTP_PASS: GOTRUE_SMTP_PASS
+      GOTRUE_SMTP_ADMIN_EMAIL: admin@email.com
     depends_on:
       - db
     restart: on-failure
@@ -76,7 +81,6 @@ services:
       GOTRUE_API_HOST: 0.0.0.0
       PORT: 9997
       GOTRUE_DISABLE_SIGNUP: 'true'
-      GOTRUE_ENABLE_SIGNUP: 'false'
       API_EXTERNAL_URL: http://localhost:9997
       GOTRUE_SITE_URL: http://localhost:9997
       GOTRUE_MAILER_AUTOCONFIRM: 'false'
@@ -85,6 +89,11 @@ services:
       DATABASE_URL: 'postgres://postgres:postgres@db:5432/postgres?sslmode=disable'
       GOTRUE_EXTERNAL_PHONE_ENABLED: 'false'
       GOTRUE_EXTERNAL_EMAIL_ENABLED: 'false'
+      GOTRUE_SMTP_HOST: mail
+      GOTRUE_SMTP_PORT: 2500
+      GOTRUE_SMTP_USER: GOTRUE_SMTP_USER
+      GOTRUE_SMTP_PASS: GOTRUE_SMTP_PASS
+      GOTRUE_SMTP_ADMIN_EMAIL: admin@email.com
     depends_on:
       - db
     restart: on-failure

tests/_async/test_client_with_sign_ups_disabled.py

@@ -3,10 +3,23 @@
 import pytest
 from faker import Faker
 
-from gotrue import AsyncGoTrueClient
+from gotrue import AsyncGoTrueAPI, AsyncGoTrueClient
+from gotrue.constants import COOKIE_OPTIONS
 from gotrue.exceptions import APIError
+from gotrue.types import CookieOptions, LinkType, User, UserAttributes
 
 GOTRUE_URL = "http://localhost:9997"
+AUTH_ADMIN_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwicm9sZSI6InN1cGFiYXNlX2FkbWluIiwiaWF0IjoxNTE2MjM5MDIyfQ.0sOtTSTfPv5oPZxsjvBO249FI4S4p0ymHoIZ6H6z9Y8"
+
+
+@pytest.fixture(name="auth_admin")
+async def create_auth_admin() -> AsyncIterable[AsyncGoTrueAPI]:
+    async with AsyncGoTrueAPI(
+        url=GOTRUE_URL,
+        headers={"Authorization": f"Bearer {AUTH_ADMIN_TOKEN}"},
+        cookie_options=CookieOptions.parse_obj(COOKIE_OPTIONS),
+    ) as api:
+        yield api
 
 
 @pytest.fixture(name="client")
@@ -35,3 +48,73 @@ async def test_sign_up(client: AsyncGoTrueClient):
         assert e.msg == expected_error_message
     except Exception as e:
         assert False, str(e)
+
+
+invited_user = fake.email().lower()
+
+
+@pytest.mark.asyncio
+async def test_generate_link_should_be_able_to_generate_multiple_links(
+    auth_admin: AsyncGoTrueAPI,
+):
+    try:
+        response = await auth_admin.generate_link(
+            type=LinkType.invite,
+            email=invited_user,
+            redirect_to="http://localhost:9997",
+        )
+        assert isinstance(response, User)
+        assert response.email == invited_user
+        assert response.action_link
+        assert "http://localhost:9997/?token=" in response.action_link
+        assert response.app_metadata
+        assert response.app_metadata.get("provider") == "email"
+        providers = response.app_metadata.get("providers")
+        assert providers
+        assert isinstance(providers, list)
+        assert len(providers) == 1
+        assert providers[0] == "email"
+        assert response.role == ""
+        assert response.user_metadata == {}
+        assert response.identities == []
+        user = response
+        response = await auth_admin.generate_link(
+            type=LinkType.invite,
+            email=invited_user,
+        )
+        assert isinstance(response, User)
+        assert response.email == invited_user
+        assert response.action_link
+        assert "http://localhost:9997/?token=" in response.action_link
+        assert response.app_metadata
+        assert response.app_metadata.get("provider") == "email"
+        providers = response.app_metadata.get("providers")
+        assert providers
+        assert isinstance(providers, list)
+        assert len(providers) == 1
+        assert providers[0] == "email"
+        assert response.role == ""
+        assert response.user_metadata == {}
+        assert response.identities == []
+        user_again = response
+        assert user.id == user_again.id
+    except Exception as e:
+        assert False, str(e)
+
+
+email2 = fake.email().lower()
+
+
+@pytest.mark.asyncio
+async def test_create_user(auth_admin: AsyncGoTrueAPI):
+    try:
+        attributes = UserAttributes(email=email2)
+        response = await auth_admin.create_user(attributes=attributes)
+        assert isinstance(response, User)
+        assert response.email == email2
+        response = await auth_admin.list_users()
+        user = next((u for u in response if u.email == email2), None)
+        assert user
+        assert user.email == email2
+    except Exception as e:
+        assert False, str(e)

tests/_sync/test_client_with_sign_ups_disabled.py

@@ -3,10 +3,23 @@
 import pytest
 from faker import Faker
 
-from gotrue import SyncGoTrueClient
+from gotrue import SyncGoTrueAPI, SyncGoTrueClient
+from gotrue.constants import COOKIE_OPTIONS
 from gotrue.exceptions import APIError
+from gotrue.types import CookieOptions, LinkType, User, UserAttributes
 
 GOTRUE_URL = "http://localhost:9997"
+AUTH_ADMIN_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwicm9sZSI6InN1cGFiYXNlX2FkbWluIiwiaWF0IjoxNTE2MjM5MDIyfQ.0sOtTSTfPv5oPZxsjvBO249FI4S4p0ymHoIZ6H6z9Y8"
+
+
+@pytest.fixture(name="auth_admin")
+def create_auth_admin() -> Iterable[SyncGoTrueAPI]:
+    with SyncGoTrueAPI(
+        url=GOTRUE_URL,
+        headers={"Authorization": f"Bearer {AUTH_ADMIN_TOKEN}"},
+        cookie_options=CookieOptions.parse_obj(COOKIE_OPTIONS),
+    ) as api:
+        yield api
 
 
 @pytest.fixture(name="client")
@@ -35,3 +48,73 @@ def test_sign_up(client: SyncGoTrueClient):
         assert e.msg == expected_error_message
     except Exception as e:
         assert False, str(e)
+
+
+invited_user = fake.email().lower()
+
+
+@pytest.mark.asyncio
+def test_generate_link_should_be_able_to_generate_multiple_links(
+    auth_admin: SyncGoTrueAPI,
+):
+    try:
+        response = auth_admin.generate_link(
+            type=LinkType.invite,
+            email=invited_user,
+            redirect_to="http://localhost:9997",
+        )
+        assert isinstance(response, User)
+        assert response.email == invited_user
+        assert response.action_link
+        assert "http://localhost:9997/?token=" in response.action_link
+        assert response.app_metadata
+        assert response.app_metadata.get("provider") == "email"
+        providers = response.app_metadata.get("providers")
+        assert providers
+        assert isinstance(providers, list)
+        assert len(providers) == 1
+        assert providers[0] == "email"
+        assert response.role == ""
+        assert response.user_metadata == {}
+        assert response.identities == []
+        user = response
+        response = auth_admin.generate_link(
+            type=LinkType.invite,
+            email=invited_user,
+        )
+        assert isinstance(response, User)
+        assert response.email == invited_user
+        assert response.action_link
+        assert "http://localhost:9997/?token=" in response.action_link
+        assert response.app_metadata
+        assert response.app_metadata.get("provider") == "email"
+        providers = response.app_metadata.get("providers")
+        assert providers
+        assert isinstance(providers, list)
+        assert len(providers) == 1
+        assert providers[0] == "email"
+        assert response.role == ""
+        assert response.user_metadata == {}
+        assert response.identities == []
+        user_again = response
+        assert user.id == user_again.id
+    except Exception as e:
+        assert False, str(e)
+
+
+email2 = fake.email().lower()
+
+
+@pytest.mark.asyncio
+def test_create_user(auth_admin: SyncGoTrueAPI):
+    try:
+        attributes = UserAttributes(email=email2)
+        response = auth_admin.create_user(attributes=attributes)
+        assert isinstance(response, User)
+        assert response.email == email2
+        response = auth_admin.list_users()
+        user = next((u for u in response if u.email == email2), None)
+        assert user
+        assert user.email == email2
+    except Exception as e:
+        assert False, str(e)