api/sources/agent/data_sources/user.py

import os

from dotenv import load_dotenv
from twilio.jwt.access_token import AccessToken
from twilio.jwt.access_token.grants import VideoGrant

from core.constants import Tables
from core.interface.supabase_client import supabase_config, async_supabase_client
from core.schemas.user_location import UserLocation

load_dotenv()


class AgentDataSource:
    @staticmethod
    async def update_location(uuid: str, lat: float, long: float):
        supabase = await async_supabase_client(supabase_config)
        location: UserLocation = UserLocation(uuid=uuid, location_lat=lat, location_long=long)
        response = await supabase.table(Tables.USER_LOCATIONS) \
            .select('*') \
            .eq('uuid', uuid) \
            .execute()
        is_location_already_added = bool(response.data)
        if is_location_already_added:
            response = await supabase.table(Tables.USER_LOCATIONS) \
                .update(location.to_json()) \
                .eq('uuid', uuid) \
                .execute()
        else:
            response = await supabase.table(Tables.USER_LOCATIONS) \
                .insert(location.to_json()) \
                .execute()
        location_id = response.data[0]['location_id']
        await supabase.table('users') \
            .update({'location_id': location_id}) \
            .eq('uuid', uuid) \
            .execute()

    @staticmethod
    async def generate_token(agent_uuid: str):
        account_sid = os.getenv('TWILIO_SID')
        api_key_sid = os.getenv('TWILIO_API_KEY')
        api_key_secret = os.getenv('TWILIO_SECRET')

        token = AccessToken(account_sid, api_key_sid, api_key_secret)

        identity = agent_uuid
        token.identity = identity

        grant = VideoGrant()
        grant.room = agent_uuid
        token.add_grant(grant)

        jwt_token = token.to_jwt()
        return jwt_token, agent_uuid

    @staticmethod
    async def accept_user_call(agent_uuid: str, token: str):
        supabase = await async_supabase_client(supabase_config)
        response = await supabase.table(Tables.VIDEO_CALLS) \
            .select('*') \
            .eq('uuid', agent_uuid) \
            .execute()
        return response.data[0]['id']

    @staticmethod
    async def delete_call(call_id: int):
        supabase = await async_supabase_client(supabase_config)
        response = await supabase.table(Tables.VIDEO_CALLS) \
            .delete() \
            .eq('id', call_id) \
            .execute()
        return response.data