from anydi import singleton from fansifter_common.adapters.encrypter import EncryptedStr, Encrypter from fansifter_common.adapters.twilio import TwilioClient from ows_text_campaigns.config import Settings from ows_text_campaigns.twilio.exceptions import TwilioAccountNotFoundError from ows_text_campaigns.twilio.models import TwilioAccount from ows_text_campaigns.twilio.repositories import TwilioAccountRepository @singleton class TwilioAccountService: def __init__( self, encrypter: Encrypter, repository: TwilioAccountRepository, twilio_client: TwilioClient, settings: Settings, ) -> None: self.encrypter = encrypter self.repository = repository self.twilio_client = twilio_client self.settings = settings def save_account( self, *, account_sid: str, auth_token: str, refresh_api_key: bool = False, ) -> TwilioAccount: account = self.repository.get(account_sid) if account is None: api_key_id, api_secret = self._create_api_key(account_sid) account = TwilioAccount( account_sid=account_sid, auth_token=self.encrypter.encrypt(auth_token), api_key_id=api_key_id, api_secret=api_secret, ) else: account.auth_token = self.encrypter.encrypt(auth_token) if not account.api_key_id or not account.api_secret or refresh_api_key: api_key_id, api_secret = self._create_api_key(account_sid) account.api_key_id = api_key_id account.api_secret = api_secret self.repository.save(account) return account def _create_api_key(self, account_sid: str) -> tuple[str, EncryptedStr]: api_key = self.twilio_client.create_api_key( account_sid=account_sid, friendly_name=f"{self.settings.environment}-api-key", ) api_key_id = api_key.sid api_secret = self.encrypter.encrypt(api_key.secret) return api_key_id, api_secret def get_api_key(self, account_sid: str) -> tuple[str, str]: twilio_account = self.repository.get(account_sid) if twilio_account is None: raise TwilioAccountNotFoundError api_key_id = twilio_account.api_key_id api_secret = self.encrypter.decrypt(twilio_account.api_secret) return api_key_id, api_secret def get_auth_token(self, account_sid: str) -> str: twilio_account = self.repository.get(account_sid) if twilio_account is None: raise TwilioAccountNotFoundError return self.encrypter.decrypt(twilio_account.auth_token)