File size: 3,540 Bytes
3cdfd33 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
import asyncio
import chainlit as cl
from phi.tools import Toolkit
from phi.utils.log import logger
from src.libs.rpc_client import rpc_call
class UserConfirmationPinToolkit(Toolkit):
def __init__(self):
super().__init__(name="user_confirmation_pin_toolkit")
# Registering methods to make them accessible via the toolkit
self.register(self.has_user_confirmation_pin)
self.register(self.set_user_confirmation_pin)
self.register(self.update_user_confirmation_pin)
self.register(self.verify_user_confirmation_pin)
def has_user_confirmation_pin(self, user_id: str) -> str:
"""
Checks if a user has a confirmation PIN.
This method takes a user's ID and checks if the user has a confirmation
PIN via a remote procedure call (RPC).
Args:
user_id (str): The ID of the user to check for a confirmation PIN.
Returns:
str: The response from the RPC call indicating if the user has a PIN.
"""
logger.info("Checking if user has confirmation pin")
params = {
"userId": user_id,
}
response = asyncio.run(rpc_call(method_name="hasUserPin", params=params))
return f"{response}"
def set_user_confirmation_pin(self, user_id: str, pin: str) -> str:
"""
Sets a user's confirmation PIN.
This method takes a user's ID and a new PIN, and sets the user's PIN
via a remote procedure call (RPC).
Args:
user_id (str): The ID of the user whose PIN is to be set.
pin (str): The new confirmation PIN to be set.
Returns:
str: The response from the RPC call setting the PIN.
"""
logger.info("Setting user confirmation pin")
params = {
"userId": user_id,
"pin": pin
}
response = asyncio.run(rpc_call(method_name="setUserPin", params=params))
return f"{response}"
def update_user_confirmation_pin(self, user_id: str, pin: str) -> str:
"""
Updates a user's confirmation PIN.
This method takes a user's ID and a new PIN, and updates the user's PIN
via a remote procedure call (RPC).
Args:
user_id (str): The ID of the user whose PIN is to be updated.
pin (str): The new confirmation PIN to be set.
Returns:
str: The response from the RPC call updating the PIN.
"""
logger.info("Updating user confirmation pin")
params = {
"userId": user_id,
"pin": pin
}
response = asyncio.run(rpc_call(method_name="updateUserPin", params=params))
return f"{response}"
def verify_user_confirmation_pin(self, user_id: str, pin: str) -> str:
"""
Verifies a user's confirmation PIN.
This method takes a user's ID and a PIN, converts the email to lowercase,
and verifies the PIN via a remote procedure call (RPC).
Args:
user_id (str): The ID of the user whose PIN is to be verified.
pin (str): The confirmation PIN to be verified.
Returns:
str: The response from the RPC call verifying the PIN.
"""
logger.info("Verifying user confirmation pin")
params = {
"userId": user_id,
"pin": pin
}
response = asyncio.run(rpc_call(method_name="verifyUserPin", params=params))
return f"{response}"
|