Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/integrations/creds_manager.py (Python)

Code before:
        """
        # Use a low-priority (!time_sensitive) locking queue on top of the general lock
        # to allow priority access for refreshing/updating the tokens.
        with self._locked(user_id, credentials_id, "!time_sensitive"):
            lock = self._acquire_lock(user_id, credentials_id)
        credentials = self.get(user_id, credentials_id, lock=False)
        if not credentials:
            raise ValueError(
                f"Credentials #{credentials_id} for user #{user_id} not found"
            )
        return credentials, lock

    def update(self, user_id: str, updated: Credentials) -> None:
        with self._locked(user_id, updated.id):
            self.store.update_creds(user_id, updated)

    def delete(self, user_id: str, credentials_id: str) -> None:
        with self._locked(user_id, credentials_id):
            self.store.delete_creds_by_id(user_id, credentials_id)

    # -- Locking utilities -- #

    def _acquire_lock(self, user_id: str, credentials_id: str, *args: str) -> RedisLock:
        key = (
            self.store.supabase.supabase_url,
            f"user:{user_id}",
            f"credentials:{credentials_id}",
            *args,
        )
        return self._locks.acquire(key)

    @contextmanager
    def _locked(self, user_id: str, credentials_id: str, *args: str):
        lock = self._acquire_lock(user_id, credentials_id, *args)
        try:
            yield
        finally:
            lock.release()

    def release_all_locks(self):
        """Call this on process termination to ensure all locks are released"""
        self._locks.release_all_locks()
        self.store.locks.release_all_locks()


def _get_provider_oauth_handler(provider_name: str) -> BaseOAuthHandler:
    if provider_name not in HANDLERS_BY_NAME:
        raise KeyError(f"Unknown provider '{provider_name}'")

    client_id = getattr(settings.secrets, f"{provider_name}_client_id")
    client_secret = getattr(settings.secrets, f"{provider_name}_client_secret")

Review comment (security):
Unless `self.store.db_manager` magically stringifies to a string representing the DB connection, this doesn't work. Can be fixed by either of:
- Replace `self.store.db_manager` with a variable uniquely identifying the DB that the credentials are stored in
- Omit `self.store.db_manager` if we can absolutely assume that `IntegrationCredentialsStore` will never be used with multiple different DBs within the same system

Code after:
        See the class docstring for more info.
        """
        # Use a low-priority (!time_sensitive) locking queue on top of the general lock
        # to allow priority access for refreshing/updating the tokens.
        with self._locked(user_id, credentials_id, "!time_sensitive"):
            lock = self._acquire_lock(user_id, credentials_id)
        credentials = self.get(user_id, credentials_id, lock=False)
        if not credentials:
            raise ValueError(
                f"Credentials #{credentials_id} for user #{user_id} not found"
            )
        return credentials, lock

    def update(self, user_id: str, updated: Credentials) -> None:
        with self._locked(user_id, updated.id):
            self.store.update_creds(user_id, updated)

    def delete(self, user_id: str, credentials_id: str) -> None:
        with self._locked(user_id, credentials_id):
            self.store.delete_creds_by_id(user_id, credentials_id)

    # -- Locking utilities -- #

    def _acquire_lock(self, user_id: str, credentials_id: str, *args: str) -> RedisLock:
        key = (
            self.store.db_manager,
            f"user:{user_id}",
            f"credentials:{credentials_id}",
            *args,
        )
        return self._locks.acquire(key)

    @contextmanager
    def _locked(self, user_id: str, credentials_id: str, *args: str):
        lock = self._acquire_lock(user_id, credentials_id, *args)
        try:
            yield
        finally:
            lock.release()

    def release_all_locks(self):
        """Call this on process termination to ensure all locks are released"""
        self._locks.release_all_locks()
        self.store.locks.release_all_locks()


def _get_provider_oauth_handler(provider_name: str) -> BaseOAuthHandler:
    if provider_name not in HANDLERS_BY_NAME:
        raise KeyError(f"Unknown provider '{provider_name}'")

    client_id = getattr(settings.secrets, f"{provider_name}_client_id")