Module supertokens_python.recipe.passwordless.smsdelivery.services.backward_compatibility

# 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.

import json
from os import environ
from typing import Any, Awaitable, Callable, Dict, Union

from httpx import AsyncClient, HTTPStatusError, Response
from supertokens_python.ingredients.smsdelivery.services.supertokens import \
    SUPERTOKENS_SMS_SERVICE_URL
from supertokens_python.ingredients.smsdelivery.types import \
    SMSDeliveryInterface
from supertokens_python.logger import log_debug_message
from supertokens_python.supertokens import AppInfo
from supertokens_python.utils import handle_httpx_client_exceptions

from ....types import PasswordlessLoginSMSTemplateVars


def default_create_and_send_custom_sms(app_info: AppInfo):
    async def func(input_: PasswordlessLoginSMSTemplateVars, _user_context: Dict[str, Any]):
        if ('SUPERTOKENS_ENV' in environ) and (environ['SUPERTOKENS_ENV'] == 'testing'):
            return
        sms_input_json = {
            'appName': app_info.app_name,
            'type': 'PASSWORDLESS_LOGIN',
            'phoneNumber': input_.phone_number,
            'codeLifetime': input_.code_life_time,
        }
        if input_.user_input_code:
            sms_input_json['userInputCode'] = input_.user_input_code
        if input_.url_with_link_code:
            sms_input_json['urlWithLinkCode'] = input_.url_with_link_code

        try:
            async with AsyncClient() as client:
                res = await client.post(  # type: ignore
                    SUPERTOKENS_SMS_SERVICE_URL,
                    json={
                        "smsInput": sms_input_json,
                    },
                    headers={'api-version': '0'}
                )
                res.raise_for_status()
                log_debug_message("Passwordless login SMS sent to %s", input_.phone_number)
                return
        except Exception as e:
            log_debug_message("Error sending passwordless login SMS")
            handle_httpx_client_exceptions(e)

            if isinstance(e, HTTPStatusError):  # type: ignore
                res: Response = e.response  # type: ignore
                if res.status_code != 429:  # type: ignore (429 == Too many requests)
                    data = res.json()
                    if "err" in data:
                        raise Exception(data["err"])
                    if data:
                        raise Exception(json.dumps(data))
                    if data is None:
                        raise e
                else:
                    pass  # Reach Point (1)
            else:
                log_debug_message("Error: %s", str(e))
                raise e

        # Point (1): Reached only when we get HTTPStatusError with e.response.status_code == 429
        print("Free daily SMS quota reached. If you want to use SuperTokens to send SMS, please sign up on supertokens.com to get your SMS API key, else you can also define your own method by overriding the service. For now, we are logging it below:")
        print("SMS content:\n", json.dumps(input_.__dict__, indent=2))

    return func


class BackwardCompatibilityService(SMSDeliveryInterface[PasswordlessLoginSMSTemplateVars]):
    def __init__(self,
                 app_info: AppInfo,
                 create_and_send_custom_sms: Union[Callable[[PasswordlessLoginSMSTemplateVars, Dict[str, Any]], Awaitable[None]], None] = None
                 ) -> None:
        self.create_and_send_custom_sms = default_create_and_send_custom_sms(app_info) if create_and_send_custom_sms is None else create_and_send_custom_sms

    async def send_sms(self, template_vars: PasswordlessLoginSMSTemplateVars, user_context: Dict[str, Any]) -> None:
        await self.create_and_send_custom_sms(template_vars, user_context)  # Note: intentionally not using try-except (unlike other recipes)