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