diff --git a/tests/test_passwordless.py b/tests/test_passwordless.py index a1aa2009..75b8e2b1 100644 --- a/tests/test_passwordless.py +++ b/tests/test_passwordless.py @@ -1,16 +1,16 @@ -import json -from requests import Response - import pytest -import workos from workos.passwordless import Passwordless +from workos.utils.http_client import SyncHTTPClient -class TestPasswordless(object): +class TestPasswordless: @pytest.fixture(autouse=True) def setup(self, set_api_key_and_client_id): - self.passwordless = Passwordless() + self.http_client = SyncHTTPClient( + base_url="https://api.workos.test", version="test" + ) + self.passwordless = Passwordless(http_client=self.http_client) @pytest.fixture def mock_passwordless_session(self): @@ -23,24 +23,24 @@ def mock_passwordless_session(self): } def test_create_session_succeeds( - self, mock_passwordless_session, mock_request_method + self, mock_passwordless_session, mock_http_client_with_response ): - mock_request_method("post", mock_passwordless_session, 201) + mock_http_client_with_response(self.http_client, mock_passwordless_session, 201) session_options = { "email": "demo@workos-okta.com", "type": "MagicLink", "expires_in": 300, } - passwordless_session = self.passwordless.create_session(session_options) + passwordless_session = self.passwordless.create_session(**session_options) - assert passwordless_session == mock_passwordless_session + assert passwordless_session.dict() == mock_passwordless_session - def test_get_send_session_succeeds(self, mock_request_method): + def test_get_send_session_succeeds(self, mock_http_client_with_response): response = { "success": True, } - mock_request_method("post", response, 200) + mock_http_client_with_response(self.http_client, response, 200) response = self.passwordless.send_session( "passwordless_session_01EHDAK2BFGWCSZXP9HGZ3VK8C" diff --git a/workos/client.py b/workos/client.py index 9c4e7405..7c8e222d 100644 --- a/workos/client.py +++ b/workos/client.py @@ -55,7 +55,7 @@ def organizations(self): @property def passwordless(self): if not getattr(self, "_passwordless", None): - self._passwordless = Passwordless() + self._passwordless = Passwordless(self._http_client) return self._passwordless @property diff --git a/workos/passwordless.py b/workos/passwordless.py index 26079b42..21a1188a 100644 --- a/workos/passwordless.py +++ b/workos/passwordless.py @@ -1,13 +1,21 @@ -from typing import Literal, Protocol +from typing import Literal, Optional, Protocol import workos -from workos.utils.request import RequestHelper, REQUEST_METHOD_POST +from workos.utils.http_client import SyncHTTPClient +from workos.utils.request import REQUEST_METHOD_POST from workos.utils.validation import PASSWORDLESS_MODULE, validate_settings -from workos.resources.passwordless import WorkOSPasswordlessSession +from workos.resources.passwordless import PasswordlessSession, PasswordlessSessionType class PasswordlessModule(Protocol): - def create_session(self, session_options: dict) -> dict: ... + def create_session( + self, + email: str, + type: PasswordlessSessionType, + redirect_uri: Optional[str] = None, + state: Optional[str] = None, + expires_in: Optional[int] = None, + ) -> PasswordlessSession: ... def send_session(self, session_id: str) -> Literal[True]: ... @@ -15,47 +23,57 @@ def send_session(self, session_id: str) -> Literal[True]: ... class Passwordless(PasswordlessModule): """Offers methods through the WorkOS Passwordless service.""" - @validate_settings(PASSWORDLESS_MODULE) - def __init__(self): - pass - - @property - def request_helper(self): - if not getattr(self, "_request_helper", None): - self._request_helper = RequestHelper() - return self._request_helper + _http_client: SyncHTTPClient - def create_session(self, session_options): + @validate_settings(PASSWORDLESS_MODULE) + def __init__(self, http_client: SyncHTTPClient): + self._http_client = http_client + + def create_session( + self, + email: str, + type: PasswordlessSessionType, + redirect_uri: Optional[str] = None, + state: Optional[str] = None, + expires_in: Optional[int] = None, + ) -> PasswordlessSession: """Create a Passwordless Session. Args: - session_options (dict) - An session options object - session_options[email] (str): The email of the user to authenticate. - session_options[redirect_uri] (str): Optional parameter to - specify the redirect endpoint which will handle the callback - from WorkOS. Defaults to the default Redirect URI in the - WorkOS dashboard. - session_options[state] (str): Optional parameter that the redirect - URI received from WorkOS will contain. The state parameter - can be used to encode arbitrary information to help - restore application state between redirects. - session_options[type] (str): The type of Passwordless Session to - create. Currently, the only supported value is 'MagicLink'. - session_options[expires_in] (int): The number of seconds the Passwordless Session should live before expiring. - This value must be between 900 (15 minutes) and 86400 (24 hours), inclusive. + email (str): The email of the user to authenticate. + redirect_uri (str): Optional parameter to + specify the redirect endpoint which will handle the callback + from WorkOS. Defaults to the default Redirect URI in the + WorkOS dashboard. + state (str): Optional parameter that the redirect + URI received from WorkOS will contain. The state parameter + can be used to encode arbitrary information to help + restore application state between redirects. + type (str): The type of Passwordless Session to + create. Currently, the only supported value is 'MagicLink'. + expires_in (int): The number of seconds the Passwordless Session should live before expiring. + This value must be between 900 (15 minutes) and 86400 (24 hours), inclusive. Returns: - dict: Passwordless Session + PasswordlessSession """ - response = self.request_helper.request( + params = { + "email": email, + "type": type, + "expires_in": expires_in, + "redirect_uri": redirect_uri, + "state": state, + } + + response = self._http_client.request( "passwordless/sessions", method=REQUEST_METHOD_POST, - params=session_options, + params=params, token=workos.api_key, ) - return WorkOSPasswordlessSession.construct_from_response(response).to_dict() + return PasswordlessSession.model_validate(response) def send_session(self, session_id: str) -> Literal[True]: """Send a Passwordless Session via email. @@ -67,7 +85,7 @@ def send_session(self, session_id: str) -> Literal[True]: Returns: boolean: Returns True """ - self.request_helper.request( + self._http_client.request( "passwordless/sessions/{session_id}/send".format(session_id=session_id), method=REQUEST_METHOD_POST, token=workos.api_key, diff --git a/workos/resources/passwordless.py b/workos/resources/passwordless.py index a0e89757..8afd48ec 100644 --- a/workos/resources/passwordless.py +++ b/workos/resources/passwordless.py @@ -1,30 +1,14 @@ -from workos.resources.base import WorkOSBaseResource +from typing import Literal +from workos.resources.workos_model import WorkOSModel +PasswordlessSessionType = Literal["MagicLink"] -class WorkOSPasswordlessSession(WorkOSBaseResource): - """Representation of a Passwordless Session Response as returned by WorkOS through the Magic Link feature. - Attributes: - OBJECT_FIELDS (list): List of fields a WorkOSPasswordlessSession is comprised of. - """ +class PasswordlessSession(WorkOSModel): + """Representation of a WorkOS Passwordless Session Response.""" - OBJECT_FIELDS = [ - "object", - "id", - "email", - "expires_at", - "link", - ] - - @classmethod - def construct_from_response(cls, response): - create_session_response = super( - WorkOSPasswordlessSession, cls - ).construct_from_response(response) - - return create_session_response - - def to_dict(self): - passwordless_session_response = super(WorkOSPasswordlessSession, self).to_dict() - - return passwordless_session_response + object: Literal["passwordless_session"] + id: str + email: str + expires_at: str + link: str