File size: 6,006 Bytes
ad87194 |
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 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 |
from fastapi import Depends
from core import logging as logger
from core.models.apis_models import *
from fastapi.routing import APIRouter
from core.api.jwt_bearer import access_check_bearer, supabase_client
from core.pipeline.user_management_pipeline import SupabaseUserManagementPipeline
from core.utils.error_handling import raise_http_exception, create_success_response, create_error_response
user_management_api_router = APIRouter(tags=["User Management"])
user_management = SupabaseUserManagementPipeline()
@user_management_api_router.post("/user_signup")
async def user_signup(request: UserSignupRequest):
logger.info(f">>>user_signup API Triggered <<<")
response = user_management.user_signup_(username=request.username, password=request.password,email=request.email)
logger.info(f">>>user_signup API Success<<<")
return response
@user_management_api_router.post("/user_signin")
async def user_signin(request: UserSigninRequest):
logger.info(f">>>user_signin API Triggered <<<")
response = user_management.user_signin_(username=request.username, password=request.password,email=request.email)
if response != None:
logger.info(f">>>user_signin API Success.<<<")
return response
else:
logger.info(f">>> Username or password is incorrect please try again.<<<")
response = create_error_response(400, "Username or password is incorrect please try again.")
return response
@user_management_api_router.post("/get_user_data")
async def get_user_data(request: GetUserDataRequest):
logger.info(f">>>get_user_data API Triggered <<<")
response = user_management.get_user_data_(access_token=request.access_token)
return response
@user_management_api_router.post("/login_with_access_token")
async def login_with_access_token(request: LoginWithAccessTokenRequest):
logger.info(f">>>login_with_access_token API Triggered <<<")
response = user_management.login_with_access_token_(access_token=request.access_token,
refresh_token=request.refresh_token)
logger.info(f">>>login_with_access_token API Success<<<")
return response
@user_management_api_router.post("/set_session_data")
async def set_session_data(request: SetSessionDataRequest):
logger.info(f">>> set_session_data API Triggered <<<")
response = user_management.set_session_data_(access_token=request.access_token, refresh_token=request.refresh_token,
user_id=request.user_id)
return response
@user_management_api_router.post("/sign_out")
async def sign_out():
logger.info(f">>> sign_out API Triggered <<<")
response = user_management.sign_out_()
logger.info(f">>>sign_out API Success<<<")
return response
@user_management_api_router.post("/oauth_signin")
async def oauth_signin():
logger.info(f">>> oauth_signin API Triggered <<<")
response = user_management.oauth_signin_()
logger.info(f">>>oauth_signin API Success<<<")
return response
@user_management_api_router.post("/check_session")
async def check_session():
logger.info(f">>>check_session API Triggered <<<")
response = user_management.check_session_()
return response
@user_management_api_router.get("/get_public_chatbot")
async def get_public_chatbots():
logger.info(f">>>get_public_chatbot API Triggered<<<")
try:
response = supabase_client.table("ChatAI_ChatbotInfo").select("*").eq("isPrivate", False).execute().data
logger.info(f">>>Public chatbots fetched successfully.<<<")
return response
except Exception as e:
logger.error(f">>>Error in get_public_chatbot: {e}<<<")
raise_http_exception(500, "Internal Server Error")
@user_management_api_router.post("/public_private_check")
async def public_or_private(request: PublicPrivateCheckRequest):
vectorstore, mode = request.vectorstore, request.mode
logger.info(f">>>public_private_check API Triggered for {vectorstore}.<<<")
try:
username, chatbot_name = vectorstore.split("$")[1], vectorstore.split("$")[2]
if len(mode) == 0:
value = (
supabase_client.table("ChatAI_ChatbotInfo")
.select("isPrivate")
.eq("user_id", username)
.eq("chatbotname", chatbot_name)
.execute()
)
value = value.data[0]["isPrivate"]
response = create_success_response(200, {"output": value})
else:
response = (
supabase_client.table("ChatAI_ChatbotInfo")
.update({"isPrivate": mode})
.eq("user_id", username)
.eq("chatbotname", chatbot_name)
.execute()
)
response = create_success_response(200, {"output": response})
logger.info(f">>>Public/Private check successful for {vectorstore}.<<<")
return response
except Exception as e:
logger.error(f">>>Error in public_private_check: {e} for {vectorstore}.<<<")
raise_http_exception(500, "Internal Server Error")
@user_management_api_router.post("/refresh_session", dependencies=[Depends(access_check_bearer)])
async def refresh_session(request: RefreshSessionRequest):
logger.info(f">>>refresh_session API Triggered <<<")
response = user_management.refresh_session__(refresh_token=request.refresh_token)
logger.info(f">>>refresh token fetched successfully.<<<")
return response
@user_management_api_router.post("/username_creation_oauth", dependencies=[Depends(access_check_bearer)])
async def username_creation_oauth(request: UsernameCreationOauthRequest):
logger.info(f">>> username_creation_oauth API Triggered <<<")
response = user_management.username_creation_oauth_(username=request.username,email=request.email, user_id=request.user_id)
logger.info(f">>>username creation successful.<<<")
return response
|