"""LinkedIn API client via RapidAPI."""
import httpx
from typing import Optional, Any
from dataclasses import dataclass
@dataclass
class LinkedInClient:
"""Client for LinkedIn data via RapidAPI."""
rapidapi_key: Optional[str] = None
base_url: str = "https://fresh-linkedin-profile-data.p.rapidapi.com"
_client: Optional[httpx.AsyncClient] = None
async def start(self):
"""Initialize HTTP client."""
if not self._client and self.rapidapi_key:
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"x-rapidapi-host": "fresh-linkedin-profile-data.p.rapidapi.com",
"x-rapidapi-key": self.rapidapi_key,
},
timeout=30.0,
)
async def close(self):
"""Close HTTP client."""
if self._client:
await self._client.aclose()
self._client = None
async def find_profile(self, name: str, company: Optional[str] = None) -> dict[str, Any]:
"""
Find LinkedIn profile URL by name.
Args:
name: Person's full name
company: Company name for disambiguation
Returns:
LinkedIn profile URL if found
"""
if not self.rapidapi_key:
return {"error": "RapidAPI key not configured"}
await self.start()
try:
params = {"name": name}
if company:
params["company"] = company
resp = await self._client.get("/search-people", params=params)
resp.raise_for_status()
data = resp.json()
# Return first match
items = data.get("data", {}).get("items", [])
if items:
return {
"found": True,
"name": name,
"linkedin_url": items[0].get("profile_url"),
"headline": items[0].get("headline"),
"location": items[0].get("location"),
"all_matches": len(items),
}
return {"found": False, "name": name}
except Exception as e:
return {"error": str(e)}
async def get_profile(self, linkedin_url: str) -> dict[str, Any]:
"""
Get full LinkedIn profile data.
Args:
linkedin_url: LinkedIn profile URL
Returns:
Full profile including experience, education, skills
"""
if not self.rapidapi_key:
return {"error": "RapidAPI key not configured"}
await self.start()
try:
resp = await self._client.get(
"/get-linkedin-profile",
params={"linkedin_url": linkedin_url},
)
resp.raise_for_status()
data = resp.json().get("data", {})
return {
"name": data.get("full_name"),
"headline": data.get("headline"),
"location": data.get("location"),
"about": data.get("about"),
"followers": data.get("followers_count"),
"connections": data.get("connections_count"),
"profile_url": linkedin_url,
"profile_pic": data.get("profile_pic_url"),
"experience": [
{
"title": exp.get("title"),
"company": exp.get("company"),
"location": exp.get("location"),
"duration": exp.get("duration"),
"description": exp.get("description"),
}
for exp in data.get("experiences", [])
],
"education": [
{
"school": edu.get("school"),
"degree": edu.get("degree"),
"field": edu.get("field_of_study"),
"dates": edu.get("date_range"),
}
for edu in data.get("educations", [])
],
"skills": data.get("skills", []),
"certifications": data.get("certifications", []),
}
except Exception as e:
return {"error": str(e)}
async def get_company(self, linkedin_url: str = None, domain: str = None) -> dict[str, Any]:
"""
Get LinkedIn company data.
Args:
linkedin_url: Company LinkedIn URL
domain: Company domain (alternative)
Returns:
Company profile data
"""
if not self.rapidapi_key:
return {"error": "RapidAPI key not configured"}
await self.start()
try:
params = {}
if linkedin_url:
params["linkedin_url"] = linkedin_url
elif domain:
params["domain"] = domain
else:
return {"error": "Provide linkedin_url or domain"}
resp = await self._client.get("/get-company-details", params=params)
resp.raise_for_status()
data = resp.json().get("data", {})
return {
"name": data.get("name"),
"tagline": data.get("tagline"),
"description": data.get("description"),
"website": data.get("website"),
"industry": data.get("industries", []),
"company_size": data.get("company_size"),
"headquarters": data.get("headquarters"),
"founded": data.get("founded"),
"specialties": data.get("specialties", []),
"followers": data.get("follower_count"),
"employee_count": data.get("employee_count"),
"linkedin_url": data.get("linkedin_url"),
}
except Exception as e:
return {"error": str(e)}
