Async LinkedIn scraper (#5)

src/vsp/app/bindings.py

@@ -1,6 +1,8 @@
 from vsp.app.prompts.prompt_loader import PromptLoader
+from vsp.app.scrapers.linkedin_downloader import LinkedinDownloader
 from vsp.llm.openai.openai import AsyncOpenAIService
 from vsp.llm.openai.openai_model import OpenAIModel
 
 prompt_loader = PromptLoader()
 open_ai_service = AsyncOpenAIService(OpenAIModel.GPT_4_MINI)
+linkedin_downloader = LinkedinDownloader()

src/vsp/app/scrapers/linkedin_downloader.py

@@ -0,0 +1,128 @@
+"""
+linkedin_downloader.py
+
+This module provides functionality for asynchronously downloading LinkedIn profile data
+using the RapidAPI LinkedIn API. It handles authentication, rate limiting, and retrying
+of requests in case of failures.
+
+Classes:
+    LinkedInFetchFailedError: Custom exception for LinkedIn fetch failures.
+    LinkedinDownloader: Main class for downloading LinkedIn profile data.
+
+Usage:
+    downloader = LinkedinDownloader()
+    profile = await downloader.fetch_linkedin_data("https://www.linkedin.com/in/username/")
+"""
+
+import asyncio
+from typing import Final
+
+import aiohttp
+from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_random_exponential
+
+from vsp.app.model.linkedin.linkedin_models import LinkedinProfile
+from vsp.shared import aws_clients, config, logger_factory
+
+logger = logger_factory.get_logger(__name__)
+
+
+class LinkedInFetchFailedError(Exception):
+    """Custom exception raised when fetching LinkedIn profile data fails."""
+
+
+class LinkedinDownloader:
+    """
+    A class for asynchronously downloading LinkedIn profile data.
+
+    This class handles authentication, rate limiting, and retrying of requests
+    when interacting with the RapidAPI LinkedIn API.
+
+    Attributes:
+        _URL (Final[str]): The RapidAPI endpoint for LinkedIn data.
+        _X_RAPIDAPI_HOST (Final[str]): The RapidAPI host for LinkedIn API.
+        _api_key (str): The RapidAPI key for authentication.
+        _semaphore (asyncio.Semaphore): Semaphore for limiting concurrent requests.
+    """
+
+    _URL: Final[str] = "https://linkedin-api8.p.rapidapi.com/"
+    _X_RAPIDAPI_HOST: Final[str] = "linkedin-api8.p.rapidapi.com"
+
+    def __init__(self, max_concurrency: int = 10):
+        """
+        Initialize the LinkedinDownloader.
+
+        Args:
+            max_concurrency (int): Maximum number of concurrent API calls. Defaults to 10.
+        """
+        self._api_key = self._fetch_api_key()
+        self._semaphore = asyncio.Semaphore(max_concurrency)
+
+    @staticmethod
+    def _fetch_api_key() -> str:
+        """
+        Fetch the LinkedIn RapidAPI key from AWS Parameter Store.
+
+        Returns:
+            str: The LinkedIn RapidAPI key.
+
+        Raises:
+            ValueError: If the API key is not found in the Parameter Store.
+            RuntimeError: If there's an error accessing the Parameter Store.
+        """
+        try:
+            return aws_clients.fetch_from_parameter_store(config.get_linkedin_key_path(), is_secret=True)
+        except aws_clients.ParameterNotFoundError as e:
+            logger.error("LinkedIn RapidAPI key not found in Parameter Store", error=str(e))
+            raise ValueError("LinkedIn RapidAPI key not found") from e
+        except aws_clients.ParameterStoreAccessError as e:
+            logger.error("Error accessing Parameter Store", error=str(e))
+            raise RuntimeError("Unable to access LinkedIn RapidAPI key") from e
+
+    def _compose_request(self, linkedin_url: str) -> tuple[dict[str, str], dict[str, str]]:
+        """
+        Compose the request headers and query parameters for the API call.
+
+        Args:
+            linkedin_url (str): The LinkedIn profile URL.
+
+        Returns:
+            tuple[dict[str, str], dict[str, str]]: A tuple containing the headers and query parameters.
+        """
+        username = linkedin_url.split("/")[-2] if linkedin_url.endswith("/") else linkedin_url.split("/")[-1]
+        querystring = {"username": username}
+        headers = {"X-RapidAPI-Key": self._api_key, "X-RapidAPI-Host": self._X_RAPIDAPI_HOST}
+        return (headers, querystring)
+
+    @retry(
+        wait=wait_random_exponential(min=1, max=60),
+        stop=stop_after_attempt(3),
+        retry=retry_if_exception_type(LinkedInFetchFailedError),
+        reraise=True,
+    )
+    async def fetch_linkedin_data(self, linkedin_url: str) -> LinkedinProfile:
+        """
+        Fetch LinkedIn profile data for a given URL.
+
+        This method is decorated with a retry mechanism that will attempt to retry
+        the call up to 3 times with exponential backoff if a LinkedInFetchFailedError is raised.
+
+        Args:
+            linkedin_url (str): The LinkedIn profile URL to fetch data for.
+
+        Returns:
+            LinkedinProfile: The fetched LinkedIn profile data.
+
+        Raises:
+            LinkedInFetchFailedError: If the API call fails after all retry attempts.
+        """
+        async with self._semaphore:
+            headers, querystring = self._compose_request(linkedin_url)
+            logger.info("Fetching LinkedIn profile", url=linkedin_url)
+            async with aiohttp.ClientSession() as session:
+                async with session.get(self._URL, headers=headers, params=querystring) as response:
+                    if response.status == 200:
+                        data = await response.json()
+                        return LinkedinProfile.profile_from_json(data)
+                    else:
+                        logger.error("Failed to fetch LinkedIn profile", url=linkedin_url, status=response.status)
+                        raise LinkedInFetchFailedError(f"Failed to fetch LinkedIn profile for {linkedin_url}")

src/vsp/llm/openai/openai.py

@@ -77,7 +77,7 @@ class AsyncOpenAIService(LLMService):
         try:
             return aws_clients.fetch_from_parameter_store(config.get_openai_api_key_path(), is_secret=True)
         except aws_clients.ParameterNotFoundError as e:
-            logger.error("API key not found in Parameter Store", error=str(e))
+            logger.error("OpenAI API key not found in Parameter Store", error=str(e))
             raise ValueError("OpenAI API key not found") from e
         except aws_clients.ParameterStoreAccessError as e:
             logger.error("Error accessing Parameter Store", error=str(e))

src/vsp/shared/config.py

@@ -91,3 +91,13 @@ def get_openai_api_key_path() -> str:
     """
     config = _get_config()
     return str(config["openai"]["openai_api_key_parameter_store_path"])
+
+
+@cache
+def get_linkedin_key_path() -> str:
+    """
+    Reads the RapidAPI key path from the TOML configuration file for LinkedIn.
+    Key is in AWS parameter store
+    """
+    config = _get_config()
+    return str(config["linkedin"]["linkedin_api_key_parameter_store_path"])

src/vsp/shared/config.toml

@@ -6,4 +6,7 @@ shared_services_aws_account = "851725506657"
 bedrock_aws_account = "339713101814"
 
 [openai]
-openai_api_key_parameter_store_path = "/secrets/openai/api_key"
+openai_api_key_parameter_store_path = "/secrets/openai/api_key"
+
+[linkedin]
+linkedin_api_key_parameter_store_path = "/secrets/rapidapi/linkedin"

tests/vsp/app/scrapers/test_integration_linkedin_downloader.py

@@ -0,0 +1,41 @@
+import asyncio
+
+import pytest
+
+from vsp.app.scrapers.linkedin_downloader import LinkedinDownloader
+from vsp.shared import logger_factory
+
+logger = logger_factory.get_logger(__name__)
+
+
+@pytest.mark.asyncio
+async def test_fetch_linkedin_data_integration():
+    """
+    Integration test for LinkedinDownloader.
+
+    This test makes an actual network request to fetch the LinkedIn profile of 'navkast'.
+    It requires a valid RapidAPI key to be set in the AWS Parameter Store.
+
+    Note: This test should be run sparingly to avoid unnecessary API calls and potential rate limiting.
+    """
+    downloader = LinkedinDownloader()
+    linkedin_url = "https://www.linkedin.com/in/navkast/"
+
+    try:
+        profile = await downloader.fetch_linkedin_data(linkedin_url)
+
+        # Log the fetched profile data
+        logger.info("Fetched LinkedIn profile", first_name=profile.first_name, last_name=profile.last_name)
+
+        # Assertions to verify the fetched data
+        assert profile.first_name == "Naveen"
+        assert profile.last_name == "K."
+        assert "University of Pennsylvania" in [edu.school_name for edu in profile.educations]
+        logger.info("Integration test passed successfully")
+    except Exception as e:
+        logger.error("Integration test failed", error=str(e))
+        raise
+
+
+if __name__ == "__main__":
+    asyncio.run(test_fetch_linkedin_data_integration())