Spaces:
Runtime error
Runtime error
File size: 3,645 Bytes
efdc12d ef1d021 13e5718 efdc12d 3d00f61 ef1d021 13e5718 efdc12d 13e5718 efdc12d 13e5718 efdc12d ef1d021 efdc12d ef1d021 efdc12d ef1d021 efdc12d ef1d021 efdc12d ef1d021 efdc12d ef1d021 efdc12d ef1d021 efdc12d |
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 |
from typing import Annotated
from aiocache import Cache, cached
from fastapi import Depends, HTTPException,status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer, OAuth2PasswordBearer
from firebase_admin import auth
from firebase_admin import credentials
import firebase_admin
from aiocache.serializers import JsonSerializer
from databases.supabase_db import create_user_with_id, get_user_with_id
bearer_scheme = HTTPBearer(auto_error=True)
# oauth2_scheme = OAuth2PasswordBearer(auto_error=True)
def initialize_firebase():
try:
# Check if the default app is already initialized
app = firebase_admin.get_app()
except ValueError:
# Initialize the app if it is not already initialized
cred = credentials.Certificate("databases/service-account.json")
app = firebase_admin.initialize_app(cred)
print("Firebase app initialized")
return app
async def verify_id_token(id_token: str) -> dict:
"""Verifies the Firebase ID token"""
try:
print("stats 1")
decoded_token = auth.verify_id_token(id_token)
return decoded_token
except auth.InvalidIdTokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid ID token",
headers={"WWW-Authenticate": "Bearer"},
)
except auth.ExpiredIdTokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Expired ID token",
headers={"WWW-Authenticate": "Bearer"},
)
async def get_firebase_user_from_token(
# token: str,
token: Annotated[HTTPAuthorizationCredentials, Depends(bearer_scheme)],
) -> dict | None:
"""Uses bearer token to identify firebase user id
Args:
token : the bearer token. Can be None as we set auto_error to False
Returns:
dict: the firebase user on success
Raises:
HTTPException 401 if user does not exist or token is invalid
"""
try:
if not token:
# raise and catch to return 401, only needed because fastapi returns 403
# by default instead of 401 so we set auto_error to False
raise ValueError("No token")
# final_user = await handle_auth_token(token.credentials)
print("token",token.credentials)
final_user = await handle_auth_token(token.credentials)
return final_user
# lots of possible exceptions, see firebase_admin.auth,
# but most of the time it is a credentials issue
except Exception as e:
# we also set the header
print(f"error {e}")
# see https://fastapi.tiangolo.com/tutorial/security/simple-oauth2/
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not logged in or Invalid credentials",
headers={"WWW-Authenticate": "Bearer"},
)
def fetch_or_create_user(user):
try:
user_list = get_user_with_id(user["user_id"])
if not user_list:
user_list = create_user_with_id(user["user_id"],user['email'])
final_user = user_list[0]
return final_user
except Exception as e:
raise HTTPException(status_code=500, detail="Failed to create user" + str(e))
@cached(ttl=36000, cache=Cache.MEMORY,serializer=JsonSerializer()) # caches the result for 10 seconds
async def handle_auth_token(credentials):
print("running func..")
user = await verify_id_token(credentials)
print("user:",user)
# check and create user in database
final_user = fetch_or_create_user(user)
return final_user
|