Module `supertokens_python.recipe.thirdpartypasswordless.recipeimplementation.passwordless_recipe_implementation`

Expand source code

# Copyright (c) 2021, VRAI Labs and/or its affiliates. All rights reserved.
#
# This software is licensed under the Apache License, Version 2.0 (the
# "License") as published by the Apache Software Foundation.
#
# You may not use this file except in compliance with the License. You may
# obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import annotations

from typing import Any, Dict, List, Union

from supertokens_python.recipe.passwordless.interfaces import (
    ConsumeCodeExpiredUserInputCodeError,
    ConsumeCodeIncorrectUserInputCodeError,
    ConsumeCodeOkResult,
    ConsumeCodeRestartFlowError,
    CreateCodeOkResult,
    CreateNewCodeForDeviceOkResult,
    CreateNewCodeForDeviceRestartFlowError,
    CreateNewCodeForDeviceUserInputCodeAlreadyUsedError,
    DeleteUserInfoOkResult,
    DeleteUserInfoUnknownUserIdError,
    DeviceType,
    RecipeInterface,
    RevokeAllCodesOkResult,
    RevokeCodeOkResult,
    UpdateUserEmailAlreadyExistsError,
    UpdateUserOkResult,
    UpdateUserPhoneNumberAlreadyExistsError,
    UpdateUserUnknownUserIdError,
)

from ...passwordless.types import User
from ..interfaces import ConsumeCodeOkResult as ThirdPartyConsumeCodeOkResult
from ..interfaces import RecipeInterface as ThirdPartyPasswordlessInterface


class RecipeImplementation(RecipeInterface):
    def __init__(self, recipe_implementation: ThirdPartyPasswordlessInterface):
        super().__init__()
        self.recipe_implementation = recipe_implementation

    async def create_code(
        self,
        email: Union[None, str],
        phone_number: Union[None, str],
        user_input_code: Union[None, str],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> CreateCodeOkResult:
        return await self.recipe_implementation.create_code(
            email, phone_number, user_input_code, tenant_id, user_context
        )

    async def create_new_code_for_device(
        self,
        device_id: str,
        user_input_code: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> Union[
        CreateNewCodeForDeviceOkResult,
        CreateNewCodeForDeviceRestartFlowError,
        CreateNewCodeForDeviceUserInputCodeAlreadyUsedError,
    ]:
        return await self.recipe_implementation.create_new_code_for_device(
            device_id, user_input_code, tenant_id, user_context
        )

    async def consume_code(
        self,
        pre_auth_session_id: str,
        user_input_code: Union[str, None],
        device_id: Union[str, None],
        link_code: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> Union[
        ConsumeCodeOkResult,
        ConsumeCodeIncorrectUserInputCodeError,
        ConsumeCodeExpiredUserInputCodeError,
        ConsumeCodeRestartFlowError,
    ]:
        result = await self.recipe_implementation.consume_code(
            pre_auth_session_id,
            user_input_code,
            device_id,
            link_code,
            tenant_id,
            user_context,
        )
        if isinstance(result, ThirdPartyConsumeCodeOkResult):
            return ConsumeCodeOkResult(
                result.created_new_user,
                User(
                    result.user.user_id,
                    result.user.email,
                    result.user.phone_number,
                    result.user.time_joined,
                    result.user.tenant_ids,
                ),
            )
        return result

    async def get_user_by_id(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        otherTypeUser = await self.recipe_implementation.get_user_by_id(
            user_id, user_context
        )
        if otherTypeUser is not None:
            if otherTypeUser.third_party_info is None:
                return User(
                    otherTypeUser.user_id,
                    otherTypeUser.email,
                    otherTypeUser.phone_number,
                    otherTypeUser.time_joined,
                    otherTypeUser.tenant_ids,
                )

        return None

    async def get_user_by_email(
        self, email: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        users = await self.recipe_implementation.get_users_by_email(
            email, tenant_id, user_context
        )
        for user in users:
            if user.third_party_info is not None:
                return User(
                    user.user_id,
                    user.email,
                    user.phone_number,
                    user.time_joined,
                    user.tenant_ids,
                )

        return None

    async def get_user_by_phone_number(
        self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        otherTypeUser = await self.recipe_implementation.get_user_by_phone_number(
            phone_number, tenant_id, user_context
        )
        if otherTypeUser is not None:
            if otherTypeUser.third_party_info is None:
                return User(
                    otherTypeUser.user_id,
                    otherTypeUser.email,
                    otherTypeUser.phone_number,
                    otherTypeUser.time_joined,
                    otherTypeUser.tenant_ids,
                )

        return None

    async def update_user(
        self,
        user_id: str,
        email: Union[str, None],
        phone_number: Union[str, None],
        user_context: Dict[str, Any],
    ) -> Union[
        UpdateUserOkResult,
        UpdateUserUnknownUserIdError,
        UpdateUserEmailAlreadyExistsError,
        UpdateUserPhoneNumberAlreadyExistsError,
    ]:
        return await self.recipe_implementation.update_passwordless_user(
            user_id, email, phone_number, user_context
        )

    async def delete_email_for_user(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
        return await self.recipe_implementation.delete_email_for_passwordless_user(
            user_id, user_context
        )

    async def delete_phone_number_for_user(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
        return await self.recipe_implementation.delete_phone_number_for_user(
            user_id, user_context
        )

    async def revoke_all_codes(
        self,
        email: Union[str, None],
        phone_number: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> RevokeAllCodesOkResult:
        return await self.recipe_implementation.revoke_all_codes(
            email, phone_number, tenant_id, user_context
        )

    async def revoke_code(
        self, code_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> RevokeCodeOkResult:
        return await self.recipe_implementation.revoke_code(
            code_id, tenant_id, user_context
        )

    async def list_codes_by_email(
        self, email: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> List[DeviceType]:
        return await self.recipe_implementation.list_codes_by_email(
            email, tenant_id, user_context
        )

    async def list_codes_by_phone_number(
        self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> List[DeviceType]:
        return await self.recipe_implementation.list_codes_by_phone_number(
            phone_number, tenant_id, user_context
        )

    async def list_codes_by_device_id(
        self, device_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[DeviceType, None]:
        return await self.recipe_implementation.list_codes_by_device_id(
            device_id, tenant_id, user_context
        )

    async def list_codes_by_pre_auth_session_id(
        self, pre_auth_session_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[DeviceType, None]:
        return await self.recipe_implementation.list_codes_by_pre_auth_session_id(
            pre_auth_session_id, tenant_id, user_context
        )

Classes

class RecipeImplementation (recipe_implementation: ThirdPartyPasswordlessInterface)

Helper class that provides a standard way to create an ABC using inheritance.

Expand source code

class RecipeImplementation(RecipeInterface):
    def __init__(self, recipe_implementation: ThirdPartyPasswordlessInterface):
        super().__init__()
        self.recipe_implementation = recipe_implementation

    async def create_code(
        self,
        email: Union[None, str],
        phone_number: Union[None, str],
        user_input_code: Union[None, str],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> CreateCodeOkResult:
        return await self.recipe_implementation.create_code(
            email, phone_number, user_input_code, tenant_id, user_context
        )

    async def create_new_code_for_device(
        self,
        device_id: str,
        user_input_code: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> Union[
        CreateNewCodeForDeviceOkResult,
        CreateNewCodeForDeviceRestartFlowError,
        CreateNewCodeForDeviceUserInputCodeAlreadyUsedError,
    ]:
        return await self.recipe_implementation.create_new_code_for_device(
            device_id, user_input_code, tenant_id, user_context
        )

    async def consume_code(
        self,
        pre_auth_session_id: str,
        user_input_code: Union[str, None],
        device_id: Union[str, None],
        link_code: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> Union[
        ConsumeCodeOkResult,
        ConsumeCodeIncorrectUserInputCodeError,
        ConsumeCodeExpiredUserInputCodeError,
        ConsumeCodeRestartFlowError,
    ]:
        result = await self.recipe_implementation.consume_code(
            pre_auth_session_id,
            user_input_code,
            device_id,
            link_code,
            tenant_id,
            user_context,
        )
        if isinstance(result, ThirdPartyConsumeCodeOkResult):
            return ConsumeCodeOkResult(
                result.created_new_user,
                User(
                    result.user.user_id,
                    result.user.email,
                    result.user.phone_number,
                    result.user.time_joined,
                    result.user.tenant_ids,
                ),
            )
        return result

    async def get_user_by_id(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        otherTypeUser = await self.recipe_implementation.get_user_by_id(
            user_id, user_context
        )
        if otherTypeUser is not None:
            if otherTypeUser.third_party_info is None:
                return User(
                    otherTypeUser.user_id,
                    otherTypeUser.email,
                    otherTypeUser.phone_number,
                    otherTypeUser.time_joined,
                    otherTypeUser.tenant_ids,
                )

        return None

    async def get_user_by_email(
        self, email: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        users = await self.recipe_implementation.get_users_by_email(
            email, tenant_id, user_context
        )
        for user in users:
            if user.third_party_info is not None:
                return User(
                    user.user_id,
                    user.email,
                    user.phone_number,
                    user.time_joined,
                    user.tenant_ids,
                )

        return None

    async def get_user_by_phone_number(
        self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[User, None]:
        otherTypeUser = await self.recipe_implementation.get_user_by_phone_number(
            phone_number, tenant_id, user_context
        )
        if otherTypeUser is not None:
            if otherTypeUser.third_party_info is None:
                return User(
                    otherTypeUser.user_id,
                    otherTypeUser.email,
                    otherTypeUser.phone_number,
                    otherTypeUser.time_joined,
                    otherTypeUser.tenant_ids,
                )

        return None

    async def update_user(
        self,
        user_id: str,
        email: Union[str, None],
        phone_number: Union[str, None],
        user_context: Dict[str, Any],
    ) -> Union[
        UpdateUserOkResult,
        UpdateUserUnknownUserIdError,
        UpdateUserEmailAlreadyExistsError,
        UpdateUserPhoneNumberAlreadyExistsError,
    ]:
        return await self.recipe_implementation.update_passwordless_user(
            user_id, email, phone_number, user_context
        )

    async def delete_email_for_user(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
        return await self.recipe_implementation.delete_email_for_passwordless_user(
            user_id, user_context
        )

    async def delete_phone_number_for_user(
        self, user_id: str, user_context: Dict[str, Any]
    ) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
        return await self.recipe_implementation.delete_phone_number_for_user(
            user_id, user_context
        )

    async def revoke_all_codes(
        self,
        email: Union[str, None],
        phone_number: Union[str, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ) -> RevokeAllCodesOkResult:
        return await self.recipe_implementation.revoke_all_codes(
            email, phone_number, tenant_id, user_context
        )

    async def revoke_code(
        self, code_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> RevokeCodeOkResult:
        return await self.recipe_implementation.revoke_code(
            code_id, tenant_id, user_context
        )

    async def list_codes_by_email(
        self, email: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> List[DeviceType]:
        return await self.recipe_implementation.list_codes_by_email(
            email, tenant_id, user_context
        )

    async def list_codes_by_phone_number(
        self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> List[DeviceType]:
        return await self.recipe_implementation.list_codes_by_phone_number(
            phone_number, tenant_id, user_context
        )

    async def list_codes_by_device_id(
        self, device_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[DeviceType, None]:
        return await self.recipe_implementation.list_codes_by_device_id(
            device_id, tenant_id, user_context
        )

    async def list_codes_by_pre_auth_session_id(
        self, pre_auth_session_id: str, tenant_id: str, user_context: Dict[str, Any]
    ) -> Union[DeviceType, None]:
        return await self.recipe_implementation.list_codes_by_pre_auth_session_id(
            pre_auth_session_id, tenant_id, user_context
        )

Ancestors

RecipeInterface
abc.ABC

Methods

async def consume_code(self, pre_auth_session_id: str, user_input_code: Union[str, None], device_id: Union[str, None], link_code: Union[str, None], tenant_id: str, user_context: Dict[str, Any]) ‑> Union[ConsumeCodeOkResult, ConsumeCodeIncorrectUserInputCodeError, ConsumeCodeExpiredUserInputCodeError, ConsumeCodeRestartFlowError]

Expand source code

async def consume_code(
    self,
    pre_auth_session_id: str,
    user_input_code: Union[str, None],
    device_id: Union[str, None],
    link_code: Union[str, None],
    tenant_id: str,
    user_context: Dict[str, Any],
) -> Union[
    ConsumeCodeOkResult,
    ConsumeCodeIncorrectUserInputCodeError,
    ConsumeCodeExpiredUserInputCodeError,
    ConsumeCodeRestartFlowError,
]:
    result = await self.recipe_implementation.consume_code(
        pre_auth_session_id,
        user_input_code,
        device_id,
        link_code,
        tenant_id,
        user_context,
    )
    if isinstance(result, ThirdPartyConsumeCodeOkResult):
        return ConsumeCodeOkResult(
            result.created_new_user,
            User(
                result.user.user_id,
                result.user.email,
                result.user.phone_number,
                result.user.time_joined,
                result.user.tenant_ids,
            ),
        )
    return result

async def create_code(self, email: Union[None, str], phone_number: Union[None, str], user_input_code: Union[None, str], tenant_id: str, user_context: Dict[str, Any]) ‑> CreateCodeOkResult

Expand source code

async def create_code(
    self,
    email: Union[None, str],
    phone_number: Union[None, str],
    user_input_code: Union[None, str],
    tenant_id: str,
    user_context: Dict[str, Any],
) -> CreateCodeOkResult:
    return await self.recipe_implementation.create_code(
        email, phone_number, user_input_code, tenant_id, user_context
    )

async def create_new_code_for_device(self, device_id: str, user_input_code: Union[str, None], tenant_id: str, user_context: Dict[str, Any]) ‑> Union[CreateNewCodeForDeviceOkResult, CreateNewCodeForDeviceRestartFlowError, CreateNewCodeForDeviceUserInputCodeAlreadyUsedError]

Expand source code

async def create_new_code_for_device(
    self,
    device_id: str,
    user_input_code: Union[str, None],
    tenant_id: str,
    user_context: Dict[str, Any],
) -> Union[
    CreateNewCodeForDeviceOkResult,
    CreateNewCodeForDeviceRestartFlowError,
    CreateNewCodeForDeviceUserInputCodeAlreadyUsedError,
]:
    return await self.recipe_implementation.create_new_code_for_device(
        device_id, user_input_code, tenant_id, user_context
    )

async def delete_email_for_user(self, user_id: str, user_context: Dict[str, Any]) ‑> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]

Expand source code

async def delete_email_for_user(
    self, user_id: str, user_context: Dict[str, Any]
) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
    return await self.recipe_implementation.delete_email_for_passwordless_user(
        user_id, user_context
    )

async def delete_phone_number_for_user(self, user_id: str, user_context: Dict[str, Any]) ‑> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]

Expand source code

async def delete_phone_number_for_user(
    self, user_id: str, user_context: Dict[str, Any]
) -> Union[DeleteUserInfoOkResult, DeleteUserInfoUnknownUserIdError]:
    return await self.recipe_implementation.delete_phone_number_for_user(
        user_id, user_context
    )

async def get_user_by_email(self, email: str, tenant_id: str, user_context: Dict[str, Any]) ‑> Optional[User]

Expand source code

async def get_user_by_email(
    self, email: str, tenant_id: str, user_context: Dict[str, Any]
) -> Union[User, None]:
    users = await self.recipe_implementation.get_users_by_email(
        email, tenant_id, user_context
    )
    for user in users:
        if user.third_party_info is not None:
            return User(
                user.user_id,
                user.email,
                user.phone_number,
                user.time_joined,
                user.tenant_ids,
            )

    return None

async def get_user_by_id(self, user_id: str, user_context: Dict[str, Any]) ‑> Optional[User]

Expand source code

async def get_user_by_id(
    self, user_id: str, user_context: Dict[str, Any]
) -> Union[User, None]:
    otherTypeUser = await self.recipe_implementation.get_user_by_id(
        user_id, user_context
    )
    if otherTypeUser is not None:
        if otherTypeUser.third_party_info is None:
            return User(
                otherTypeUser.user_id,
                otherTypeUser.email,
                otherTypeUser.phone_number,
                otherTypeUser.time_joined,
                otherTypeUser.tenant_ids,
            )

    return None

async def get_user_by_phone_number(self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]) ‑> Optional[User]

Expand source code

async def get_user_by_phone_number(
    self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
) -> Union[User, None]:
    otherTypeUser = await self.recipe_implementation.get_user_by_phone_number(
        phone_number, tenant_id, user_context
    )
    if otherTypeUser is not None:
        if otherTypeUser.third_party_info is None:
            return User(
                otherTypeUser.user_id,
                otherTypeUser.email,
                otherTypeUser.phone_number,
                otherTypeUser.time_joined,
                otherTypeUser.tenant_ids,
            )

    return None

async def list_codes_by_device_id(self, device_id: str, tenant_id: str, user_context: Dict[str, Any]) ‑> Optional[DeviceType]

Expand source code

async def list_codes_by_device_id(
    self, device_id: str, tenant_id: str, user_context: Dict[str, Any]
) -> Union[DeviceType, None]:
    return await self.recipe_implementation.list_codes_by_device_id(
        device_id, tenant_id, user_context
    )

async def list_codes_by_email(self, email: str, tenant_id: str, user_context: Dict[str, Any]) ‑> List[DeviceType]

Expand source code

async def list_codes_by_email(
    self, email: str, tenant_id: str, user_context: Dict[str, Any]
) -> List[DeviceType]:
    return await self.recipe_implementation.list_codes_by_email(
        email, tenant_id, user_context
    )

async def list_codes_by_phone_number(self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]) ‑> List[DeviceType]

Expand source code

async def list_codes_by_phone_number(
    self, phone_number: str, tenant_id: str, user_context: Dict[str, Any]
) -> List[DeviceType]:
    return await self.recipe_implementation.list_codes_by_phone_number(
        phone_number, tenant_id, user_context
    )

async def list_codes_by_pre_auth_session_id(self, pre_auth_session_id: str, tenant_id: str, user_context: Dict[str, Any]) ‑> Optional[DeviceType]

Expand source code

async def list_codes_by_pre_auth_session_id(
    self, pre_auth_session_id: str, tenant_id: str, user_context: Dict[str, Any]
) -> Union[DeviceType, None]:
    return await self.recipe_implementation.list_codes_by_pre_auth_session_id(
        pre_auth_session_id, tenant_id, user_context
    )

async def revoke_all_codes(self, email: Union[str, None], phone_number: Union[str, None], tenant_id: str, user_context: Dict[str, Any]) ‑> RevokeAllCodesOkResult

Expand source code

async def revoke_all_codes(
    self,
    email: Union[str, None],
    phone_number: Union[str, None],
    tenant_id: str,
    user_context: Dict[str, Any],
) -> RevokeAllCodesOkResult:
    return await self.recipe_implementation.revoke_all_codes(
        email, phone_number, tenant_id, user_context
    )

async def revoke_code(self, code_id: str, tenant_id: str, user_context: Dict[str, Any]) ‑> RevokeCodeOkResult

Expand source code

async def revoke_code(
    self, code_id: str, tenant_id: str, user_context: Dict[str, Any]
) -> RevokeCodeOkResult:
    return await self.recipe_implementation.revoke_code(
        code_id, tenant_id, user_context
    )

async def update_user(self, user_id: str, email: Union[str, None], phone_number: Union[str, None], user_context: Dict[str, Any]) ‑> Union[UpdateUserOkResult, UpdateUserUnknownUserIdError, UpdateUserEmailAlreadyExistsError, UpdateUserPhoneNumberAlreadyExistsError]

Expand source code

async def update_user(
    self,
    user_id: str,
    email: Union[str, None],
    phone_number: Union[str, None],
    user_context: Dict[str, Any],
) -> Union[
    UpdateUserOkResult,
    UpdateUserUnknownUserIdError,
    UpdateUserEmailAlreadyExistsError,
    UpdateUserPhoneNumberAlreadyExistsError,
]:
    return await self.recipe_implementation.update_passwordless_user(
        user_id, email, phone_number, user_context
    )