Get funding data from pitchbook

src/app/domain/companies/services.py

@@ -11,7 +11,7 @@
 from uuid_utils.compat import uuid4
 
 from app.lib.pdl import get_company_details
-from app.lib.pitchbook import get_company_investors
+from app.lib.pitchbook import get_company_funding_data
 from app.lib.schema import CamelizedBaseStruct, Location, Funding, OrgSize
 from app.db.models import Company
 from app.lib.utils import get_domain
@@ -106,14 +106,19 @@ async def create(
             last_round_name = company_details.get("funding_stages")[0]
             words = last_round_name.split("_")
             words = [word.capitalize() for word in words]
-            obj.last_funding = Funding(round_name=" ".join(words))
 
         # Get investor names and last funding round name
         try:
-            investors = await get_company_investors(obj.url)
-            obj.last_funding.investors = investors
+            funding_data = await get_company_funding_data(obj.url)
+            obj.last_funding = Funding(
+                round_name=funding_data["round_name"],
+                money_raised=funding_data["money_raised"],
+                announced_date=funding_data["announced_date"],
+                investors=funding_data["investors"],
+            )
         except Exception as e:
-            await logger.awarn("Failed to get company investors", url=obj.url, exc_info=e)
+            obj.last_funding = Funding()
+            await logger.awarn("Failed to get company funding data", url=obj.url, exc_info=e)
 
         obj.ios_app_url = await get_ios_app_url(obj.url)
         obj.android_app_url = await get_android_app_url(obj.name, obj.url)

src/app/lib/pitchbook.py

@@ -1,17 +1,47 @@
 import os
 import json
 import httpx
+import asyncio
 import structlog
-from typing import Any
+from typing import Any, Dict
+from datetime import date
 
 from app.lib.schema import Investor
 
 
 logger = structlog.get_logger()
 pb_api_key = os.environ["PB_API_KEY"]
 
+pitchbook_round_map = {
+    "SeedA": "Pre-Seed",
+    "Seed": "Seed",
+    "A": "Series A",
+    "B": "Series B",
+    "C": "Series C",
+    "D": "Series D",
+    "E": "Series E",
+    "F": "Series F",
+    "PEGTH": "Private Equity",
+    "IPO": "Public",
+}
 
-async def get_company_investors(url: str) -> list[str]:
+
+async def fetch_url(client: httpx.AsyncClient, url: str, headers: Dict[str, Any] = None, params: Dict[str, Any] = None):
+    """Fetch url."""
+    try:
+        if not headers:
+            headers = []
+        if not params:
+            params = []
+        response = await client.get(url, headers=headers)
+        if response.status_code != 200 or not response.text:
+            return {}
+        return response.json()
+    except httpx.RequestError as e:
+        await logger.awarn("Error while fetching data.", exc_info=e)
+
+
+async def get_company_funding_data(url: str) -> list[str]:
     """Get company investors."""
     if not url:
         raise Exception("url is required")
@@ -29,24 +59,55 @@ async def get_company_investors(url: str) -> list[str]:
         data = response.json()
         try:
             pb_company_id = data["items"][0]["companyId"]
-        except (KeyError, TypeError, IndexError):
-            await logger.awarn("Company not found.", response=data, url=url)
+        except (KeyError, TypeError, IndexError) as e:
+            await logger.awarn("Company not found.", response=data, url=url, exc_info=e)
             raise Exception("Company not found.")
 
-        # Fetch investor list
-        response = await client.get(
-            f"https://api.pitchbook.com/companies/{pb_company_id}/active-investors", headers=headers, params=params
-        )
-        data = response.json()
+        results = []
         try:
-            investors = [
-                item["investorName"]
-                for item in data
-                if "individual" not in item["investorTypes"][0]["type"]["description"]
+            urls = [
+                f"https://api.pitchbook.com/companies/{pb_company_id}/active-investors",
+                f"https://api.pitchbook.com/companies/{pb_company_id}/most-recent-financing",
             ]
-        except (KeyError, TypeError, IndexError):
-            await logger.awarn("Investors found.", response=data, url=url)
-            raise Exception("Investors not found.")
 
-        # TODO: Fetch last funding round name
-        return investors
+            tasks = [fetch_url(client, url, headers=headers) for url in urls]
+            results = await asyncio.gather(*tasks)
+
+            # Get funding round name (Awesome API design Pitchbook!)
+            print(results[1]["lastFinancingDealType3"])
+            print(results[1]["lastFinancingDealType2"])
+            print(results[1]["lastFinancingDealType"])
+            try:
+                round_name = pitchbook_round_map[results[1]["lastFinancingDealType3"]["code"]]
+            except (KeyError, TypeError):
+                try:
+                    round_name = pitchbook_round_map[results[1]["lastFinancingDealType2"]["code"]]
+                except (KeyError, TypeError):
+                    try:
+                        round_name = pitchbook_round_map[results[1]["lastFinancingDealType"]["code"]]
+                    except (KeyError, TypeError):
+                        round_name = "Series Unknown"
+
+            try:
+                announced_date = date.fromisoformat(results[1]["lastFinancingDate"])
+            except:
+                announced_date = None
+
+            try:
+                money_raised = results[1]["lastFinancingSize"]["amount"]
+            except:
+                money_raised = None
+
+            return {
+                "investors": [
+                    item["investorName"]
+                    for item in results[0]
+                    if "individual" not in item["investorTypes"][0]["type"]["description"]
+                ],
+                "round_name": round_name,
+                "announced_date": announced_date,
+                "money_raised": money_raised,
+            }
+        except (KeyError, TypeError, IndexError) as e:
+            await logger.awarn("Failed to get round and investors", response=results, url=url, exc_info=e)
+            raise Exception("Failed to get round and investors.")

src/app/lib/schema.py

@@ -67,7 +67,7 @@ class Funding(CamelizedBaseStruct):
     """Funding data."""
 
     round_name: FundingRound = FundingRound.SERIES_UNKNOWN
-    money_raised: int | None = None
+    money_raised: float | None = None
     announced_date: date | None = None
     investors: list[Investor] = []