Module supertokens_python.recipe.dashboard.api

async def api_key_protector(
    api_implementation: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    api_function: Callable[
        [APIInterface, str, APIOptions, Dict[str, Any]], Awaitable[APIResponse]
    ],
    user_context: Dict[str, Any],
) -> Optional[BaseResponse]:
    should_allow_access = False

    try:
        should_allow_access = (
            await api_options.recipe_implementation.should_allow_access(
                api_options.request, api_options.config, user_context
            )
        )
    except DashboardOperationNotAllowedError as _:
        return send_non_200_response_with_message(
            "You are not permitted to perform this operation",
            403,
            api_options.response,
        )

    if should_allow_access is False:
        return send_non_200_response_with_message(
            "Unauthorised access", 401, api_options.response
        )

    response = await api_function(
        api_implementation, tenant_id, api_options, user_context
    )
    return send_200_response(response.to_json(), api_options.response)

async def handle_analytics_post(
    _: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> AnalyticsResponse:
    if not Supertokens.get_instance().telemetry:
        return AnalyticsResponse()
    body = await api_options.request.json()
    if body is None:
        raise_bad_input_exception("Please send body")
    email = body.get("email")
    dashboard_version = body.get("dashboardVersion")

    if email is None:
        raise_bad_input_exception("Missing required property 'email'")
    if dashboard_version is None:
        raise_bad_input_exception("Missing required property 'dashboardVersion'")

    telemetry_id = None

    try:
        response = await Querier.get_instance().send_get_request(
            NormalisedURLPath("/telemetry"),
            None,
            _user_context,
        )
        if "exists" in response and response["exists"] and "telemetryId" in response:
            telemetry_id = response["telemetryId"]

        number_of_users = await Supertokens.get_instance().get_user_count(
            include_recipe_ids=None
        )

    except Exception as __:
        # If either telemetry id API or user count fetch fails, no event should be sent
        return AnalyticsResponse()

    apiDomain, websiteDomain, appName = (
        api_options.app_info.api_domain,
        api_options.app_info.get_origin(api_options.request, {}),
        api_options.app_info.app_name,
    )

    data = {
        "websiteDomain": websiteDomain.get_as_string_dangerous(),
        "apiDomain": apiDomain.get_as_string_dangerous(),
        "appName": appName,
        "sdk": "python",
        "sdkVersion": SDKVersion,
        "numberOfUsers": number_of_users,
        "email": email,
        "dashboardVersion": dashboard_version,
    }

    if telemetry_id is not None:
        data["telemetryId"] = telemetry_id

    try:
        async with AsyncClient(timeout=30.0) as client:
            await client.post(  # type: ignore
                url=TELEMETRY_SUPERTOKENS_API_URL,
                json=data,
                headers={"api-version": TELEMETRY_SUPERTOKENS_API_VERSION},
            )
    except Exception as __:
        # If telemetry event fails, no error should be thrown
        pass

    return AnalyticsResponse()

async def handle_dashboard_api(
    api_implementation: APIInterface,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Optional[BaseResponse]:
    if api_implementation.dashboard_get is None:
        return None

    html_str = await api_implementation.dashboard_get(api_options, user_context)

    api_options.response.set_html_content(html_str)
    return api_options.response

async def handle_email_verify_token_post(
    _api_interface: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Union[
    UserEmailVerifyTokenPostAPIOkResponse,
    UserEmailVerifyTokenPostAPIEmailAlreadyVerifiedErrorResponse,
]:
    request_body: Dict[str, Any] = await api_options.request.json()  # type: ignore
    recipe_user_id = request_body.get("recipeUserId")

    if recipe_user_id is None or not isinstance(recipe_user_id, str):
        raise_bad_input_exception(
            "Required parameter 'recipeUserId' is missing or has an invalid type"
        )

    user = await get_user(recipe_user_id, user_context)

    if user is None:
        raise_bad_input_exception("User not found")

    res = await send_email_verification_email(
        tenant_id=tenant_id,
        user_id=user.id,
        recipe_user_id=RecipeUserId(recipe_user_id),
        email=None,
        user_context=user_context,
    )

    if isinstance(res, SendEmailVerificationEmailAlreadyVerifiedError):
        return UserEmailVerifyTokenPostAPIEmailAlreadyVerifiedErrorResponse()

    return UserEmailVerifyTokenPostAPIOkResponse()

async def handle_emailpassword_signin_api(
    _: APIInterface, api_options: APIOptions, _user_context: Dict[str, Any]
):
    body = await api_options.request.json()
    if body is None:
        raise_bad_input_exception("Please send body")
    email = body.get("email")
    password = body.get("password")

    if email is None or not isinstance(email, str):
        raise_bad_input_exception("Missing required parameter 'email'")
    if password is None or not isinstance(password, str):
        raise_bad_input_exception("Missing required parameter 'password'")
    response = await Querier.get_instance().send_post_request(
        NormalisedURLPath("/recipe/dashboard/signin"),
        {"email": email, "password": password},
        user_context=_user_context,
    )

    if "status" in response and response["status"] == "OK":
        return send_200_response(
            {"status": "OK", "sessionId": response["sessionId"]}, api_options.response
        )
    if "status" in response and response["status"] == "INVALID_CREDENTIALS_ERROR":
        return send_200_response(
            {"status": "INVALID_CREDENTIALS_ERROR"},
            api_options.response,
        )
    if "status" in response and response["status"] == "USER_SUSPENDED_ERROR":
        return send_200_response(
            {"status": "USER_SUSPENDED_ERROR", "message": response["message"]},
            api_options.response,
        )

async def handle_emailpassword_signout_api(
    _: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> SignOutOK:
    if api_options.config.auth_mode == "api-key":
        return SignOutOK()
    session_id_form_auth_header = api_options.request.get_header("authorization")
    if not session_id_form_auth_header:
        return raise_bad_input_exception(
            "Neither 'API Key' nor 'Authorization' header was found"
        )
    session_id_form_auth_header = session_id_form_auth_header.split()[1]
    await Querier.get_instance().send_delete_request(
        NormalisedURLPath("/recipe/dashboard/session"),
        {"sessionId": session_id_form_auth_header},
        user_context=_user_context,
    )
    return SignOutOK()

async def handle_get_tags(
    _: APIInterface, _tenant_id: str, __: APIOptions, _user_context: Dict[str, Any]
) -> SearchTagsOK:
    response = await Querier.get_instance().send_get_request(
        NormalisedURLPath("/user/search/tags"), None, _user_context
    )
    return SearchTagsOK(tags=response["tags"])

async def handle_metadata_get(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Union[UserMetadataGetAPIOkResponse, FeatureNotEnabledError]:
    user_id = api_options.request.get_query_param("userId")

    if user_id is None:
        raise_bad_input_exception("Missing required parameter 'userId'")

    try:
        UserMetadataRecipe.get_instance()
    except Exception:
        return FeatureNotEnabledError()

    metadata_response = await get_user_metadata(user_id, user_context=user_context)
    return UserMetadataGetAPIOkResponse(metadata_response.metadata)

async def handle_metadata_put(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> UserMetadataPutAPIResponse:
    request_body: Dict[str, Any] = await api_options.request.json()  # type: ignore
    user_id = request_body.get("userId")
    data = request_body.get("data")

    # This is to throw an error early in case the recipe has not been initialised
    UserMetadataRecipe.get_instance()

    if user_id is None or isinstance(user_id, str) is False:
        raise_bad_input_exception(
            "Required parameter 'userId' is missing or has an invalid type"
        )

    if data is None or isinstance(data, str) is False:
        raise_bad_input_exception(
            "Required parameter 'data' is missing or has an invalid type"
        )

    parsed_data: Dict[str, Any] = {}
    try:
        parsed_data = json.loads(data)
        if not isinstance(parsed_data, dict):  # type: ignore
            raise Exception()

    except Exception:
        raise_bad_input_exception("'data' must be a valid JSON body")

    # This API is meant to set the user metadata of a user. We delete the existing data
    # before updating it because we want to make sure that shallow merging does not result
    # in the data being incorrect
    #
    # For example if the old data is {test: "test", test2: "test2"} and the user wants to delete
    # test2 from the data simply calling updateUserMetadata with {test: "test"} would not remove
    # test2 because of shallow merging.
    #
    # Removing first ensures that the final data is exactly what the user wanted it to be

    await clear_user_metadata(user_id, user_context)
    await update_user_metadata(user_id, parsed_data, user_context)

    return UserMetadataPutAPIResponse()

async def handle_sessions_get(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> UserSessionsGetAPIResponse:
    user_id = api_options.request.get_query_param("userId")

    if user_id is None:
        raise_bad_input_exception("Missing required parameter 'userId'")

    # Passing tenant id as None sets fetch_across_all_tenants to True
    # which is what we want here.
    session_handles = await get_all_session_handles_for_user(user_id)
    sessions: List[Optional[SessionInfo]] = [None for _ in session_handles]

    async def call_(i: int, session_handle: str):
        try:
            session_response = await get_session_information(
                session_handle, user_context
            )
            if session_response is not None:
                sessions[i] = SessionInfo(session_response)
        except Exception:
            sessions[i] = None

    session_info_promises = [
        call_(i, handle) for i, handle in enumerate(session_handles)
    ]

    await asyncio.gather(*session_info_promises)

    return UserSessionsGetAPIResponse([s for s in sessions if s is not None])

async def handle_user_delete(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> UserDeleteAPIResponse:
    user_id = api_options.request.get_query_param("userId")
    remove_all_linked_accounts_query_value = api_options.request.get_query_param(
        "removeAllLinkedAccounts"
    )

    if remove_all_linked_accounts_query_value is not None:
        remove_all_linked_accounts_query_value = (
            remove_all_linked_accounts_query_value.strip().lower()
        )

    remove_all_linked_accounts = (
        True
        if remove_all_linked_accounts_query_value is None
        else remove_all_linked_accounts_query_value == "true"
    )

    if user_id is None or user_id == "":
        raise_bad_input_exception("Missing required parameter 'userId'")

    await delete_user(user_id, remove_all_linked_accounts)

    return UserDeleteAPIResponse()

async def handle_user_email_verify_get(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Union[UserEmailVerifyGetAPIResponse, FeatureNotEnabledError]:
    req = api_options.request
    recipe_user_id = req.get_query_param("recipeUserId")

    if recipe_user_id is None:
        raise_bad_input_exception("Missing required parameter 'recipeUserId'")

    try:
        EmailVerificationRecipe.get_instance_or_throw()
    except Exception:
        return FeatureNotEnabledError()

    is_verified = await is_email_verified(
        RecipeUserId(recipe_user_id), user_context=user_context
    )
    return UserEmailVerifyGetAPIResponse(is_verified)

async def handle_user_email_verify_put(
    _api_interface: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> UserEmailVerifyPutAPIResponse:
    request_body: Dict[str, Any] = await api_options.request.json()  # type: ignore
    recipe_user_id = request_body.get("recipeUserId")
    verified = request_body.get("verified")

    if recipe_user_id is None or not isinstance(recipe_user_id, str):
        raise_bad_input_exception(
            "Required parameter 'recipeUserId' is missing or has an invalid type"
        )

    if verified is None or not isinstance(verified, bool):
        raise_bad_input_exception(
            "Required parameter 'verified' is missing or has an invalid type"
        )

    if verified:
        token_response = await create_email_verification_token(
            tenant_id=tenant_id,
            recipe_user_id=RecipeUserId(recipe_user_id),
            email=None,
            user_context=user_context,
        )

        if isinstance(
            token_response, CreateEmailVerificationTokenEmailAlreadyVerifiedError
        ):
            return UserEmailVerifyPutAPIResponse()

        verify_response = await verify_email_using_token(
            tenant_id=tenant_id, token=token_response.token, user_context=user_context
        )

        if isinstance(verify_response, VerifyEmailUsingTokenInvalidTokenError):
            # This should never happen because we consume the token immediately after creating it
            raise Exception("Should not come here")

    else:
        await unverify_email(RecipeUserId(recipe_user_id), user_context=user_context)

    return UserEmailVerifyPutAPIResponse()

async def handle_user_get(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> Union[UserGetAPINoUserFoundError, UserGetAPIOkResponse,]:
    user_id = api_options.request.get_query_param("userId")

    if user_id is None:
        raise_bad_input_exception("Missing required parameter 'userId'")

    user_response = await get_user(user_id, _user_context)
    if user_response is None:
        return UserGetAPINoUserFoundError()

    user_with_metadata: UserWithMetadata = UserWithMetadata().from_user(user_response)

    try:
        UserMetadataRecipe.get_instance()
    except Exception:
        user_with_metadata.first_name = "FEATURE_NOT_ENABLED"
        user_with_metadata.last_name = "FEATURE_NOT_ENABLED"

        return UserGetAPIOkResponse(user_with_metadata)

    user_metadata = await get_user_metadata(user_id, user_context=_user_context)
    first_name = user_metadata.metadata.get("first_name", "")
    last_name = user_metadata.metadata.get("last_name", "")

    user_with_metadata.first_name = first_name
    user_with_metadata.last_name = last_name

    return UserGetAPIOkResponse(user_with_metadata)

async def handle_user_password_put(
    _api_interface: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Union[UserPasswordPutAPIResponse, UserPasswordPutAPIInvalidPasswordErrorResponse]:
    request_body: Dict[str, Any] = await api_options.request.json()  # type: ignore
    recipe_user_id = request_body.get("recipeUserId")
    new_password = request_body.get("newPassword")

    if recipe_user_id is None or not isinstance(recipe_user_id, str):
        raise_bad_input_exception("Missing required parameter 'recipeUserId'")

    if new_password is None or not isinstance(new_password, str):
        raise_bad_input_exception("Missing required parameter 'newPassword'")

    email_password_recipe = EmailPasswordRecipe.get_instance()
    update_response = (
        await email_password_recipe.recipe_implementation.update_email_or_password(
            recipe_user_id=RecipeUserId(recipe_user_id),
            email=None,
            password=new_password,
            apply_password_policy=True,
            tenant_id_for_password_policy=tenant_id,
            user_context=user_context,
        )
    )

    if isinstance(update_response, PasswordPolicyViolationError):
        return UserPasswordPutAPIInvalidPasswordErrorResponse(
            error=update_response.failure_reason
        )

    if isinstance(update_response, UnknownUserIdError):
        raise Exception("Should never come here")

    return UserPasswordPutAPIResponse()

async def handle_user_put(
    _api_interface: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> Union[
    OkResponse,
    InvalidEmailErrorResponse,
    EmailAlreadyExistsErrorResponse,
    InvalidPhoneErrorResponse,
    PhoneAlreadyExistsErrorResponse,
    EmailChangeNotAllowedErrorResponse,
    PhoneNumberChangeNotAllowedErrorResponse,
]:
    request_body = await api_options.request.json()
    if request_body is None:
        raise_bad_input_exception("Request body is missing")
    recipe_user_id = request_body.get("recipeUserId")
    recipe_id = request_body.get("recipeId")
    first_name = request_body.get("firstName")
    last_name = request_body.get("lastName")
    email = request_body.get("email")
    phone = request_body.get("phone")

    if not isinstance(recipe_user_id, str):
        raise_bad_input_exception(
            "Required parameter 'recipeUserId' is missing or has an invalid type"
        )

    if not isinstance(recipe_id, str):
        raise_bad_input_exception(
            "Required parameter 'recipeId' is missing or has an invalid type"
        )

    if not isinstance(first_name, str):
        raise_bad_input_exception(
            "Required parameter 'firstName' is missing or has an invalid type"
        )

    if not isinstance(last_name, str):
        raise_bad_input_exception(
            "Required parameter 'lastName' is missing or has an invalid type"
        )

    if not isinstance(email, str):
        raise_bad_input_exception(
            "Required parameter 'email' is missing or has an invalid type"
        )

    if not isinstance(phone, str):
        raise_bad_input_exception(
            "Required parameter 'phone' is missing or has an invalid type"
        )

    user_response = await get_user_for_recipe_id(
        RecipeUserId(recipe_user_id), recipe_id, user_context
    )

    if user_response.user is None or user_response.recipe is None:
        raise Exception("Should never come here")

    if first_name.strip() or last_name.strip():
        is_recipe_initialized = False
        try:
            UserMetadataRecipe.get_instance()
            is_recipe_initialized = True
        except Exception:
            pass

        if is_recipe_initialized:
            metadata_update: Dict[str, Any] = {}

            if first_name.strip():
                metadata_update["first_name"] = first_name.strip()

            if last_name.strip():
                metadata_update["last_name"] = last_name.strip()

            await update_user_metadata(
                user_response.user.user.id, metadata_update, user_context
            )

    if email.strip():
        email_update_response = await update_email_for_recipe_id(
            user_response.recipe,
            RecipeUserId(recipe_user_id),
            email.strip(),
            tenant_id,
            user_context,
        )

        if isinstance(email_update_response, EmailChangeNotAllowedErrorResponse):
            return EmailChangeNotAllowedErrorResponse(email_update_response.error)

        if not isinstance(email_update_response, OkResponse):
            return email_update_response

    if phone.strip():
        phone_update_response = await update_phone_for_recipe_id(
            RecipeUserId(recipe_user_id),
            phone.strip(),
            tenant_id,
            user_context,
        )

        if isinstance(phone_update_response, PhoneNumberChangeNotAllowedErrorResponse):
            return PhoneNumberChangeNotAllowedErrorResponse(phone_update_response.error)

        if not isinstance(phone_update_response, OkResponse):
            return phone_update_response

    return OkResponse()

async def handle_user_sessions_post(
    _api_interface: APIInterface,
    _tenant_id: str,
    api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> UserSessionsPostAPIResponse:
    request_body = await api_options.request.json()  # type: ignore
    session_handles: Optional[List[str]] = request_body.get("sessionHandles")  # type: ignore

    if not isinstance(session_handles, list):
        raise_bad_input_exception(
            "Required parameter 'sessionHandles' is missing or has an invalid type"
        )

    await revoke_multiple_sessions(session_handles, _user_context)
    return UserSessionsPostAPIResponse()

async def handle_users_count_get_api(
    _: APIInterface,
    tenant_id: str,
    _api_options: APIOptions,
    _user_context: Dict[str, Any],
) -> UserCountGetAPIResponse:
    count = await Supertokens.get_instance().get_user_count(
        None,
        tenant_id,
    )
    return UserCountGetAPIResponse(count=count)

async def handle_users_get_api(
    api_implementation: APIInterface,
    tenant_id: str,
    api_options: APIOptions,
    user_context: Dict[str, Any],
) -> DashboardUsersGetResponse:
    _ = api_implementation

    limit = api_options.request.get_query_param("limit")
    if limit is None:
        raise_bad_input_exception("Missing required parameter 'limit'")

    time_joined_order = api_options.request.get_query_param("timeJoinedOrder", "DESC")
    if time_joined_order not in ["ASC", "DESC"]:
        raise_bad_input_exception("Invalid value received for 'timeJoinedOrder'")

    pagination_token = api_options.request.get_query_param("paginationToken")
    query = get_search_params_from_url(api_options.request.get_original_url())

    users_response = await (
        get_users_newest_first
        if time_joined_order == "DESC"
        else get_users_oldest_first
    )(
        tenant_id,
        limit=int(limit),
        pagination_token=pagination_token,
        query=query,
        user_context=user_context,
    )

    try:
        UserMetadataRecipe.get_instance()
    except GeneralError:
        users_with_metadata: List[UserWithMetadata] = [
            UserWithMetadata().from_user(user) for user in users_response.users
        ]
        return DashboardUsersGetResponse(
            users_with_metadata, users_response.next_pagination_token
        )

    users_with_metadata: List[UserWithMetadata] = [
        UserWithMetadata().from_user(user) for user in users_response.users
    ]
    metadata_fetch_awaitables: List[Awaitable[Any]] = []

    async def get_user_metadata_and_update_user(user_idx: int) -> None:
        user = users_response.users[user_idx]
        user_metadata = await get_user_metadata(user.id)
        first_name = user_metadata.metadata.get("first_name")
        last_name = user_metadata.metadata.get("last_name")

        users_with_metadata[user_idx].first_name = first_name
        users_with_metadata[user_idx].last_name = last_name

    for i, _ in enumerate(users_response.users):
        metadata_fetch_awaitables.append(get_user_metadata_and_update_user(i))

    promise_arr_start_position = 0
    batch_size = 5

    while promise_arr_start_position < len(metadata_fetch_awaitables):
        # We want to query only 5 in parallel at a time
        promises_to_call = [
            metadata_fetch_awaitables[i]
            for i in range(
                promise_arr_start_position,
                min(
                    promise_arr_start_position + batch_size,
                    len(metadata_fetch_awaitables),
                ),
            )
        ]
        await asyncio.gather(*promises_to_call)
        promise_arr_start_position += batch_size

    return DashboardUsersGetResponse(
        users_with_metadata,
        users_response.next_pagination_token,
    )

async def handle_validate_key_api(
    _api_implementation: APIInterface,
    api_options: APIOptions,
    user_context: Dict[str, Any],
):

    is_valid_key = await validate_api_key(
        api_options.request, api_options.config, user_context
    )

    if is_valid_key:
        return send_200_response({"status": "OK"}, api_options.response)
    return send_non_200_response_with_message("Unauthorised", 401, api_options.response)