Add Google Places ETL pipelines

Author: TristanGramsch

etl/pipelines/google_places/__init__.py

@@ -0,0 +1 @@
+"""Google Places ETL pipelines."""

etl/pipelines/google_places/common.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from dtos import Activity, Address, Availability, Contact, ItemCategory, Service
+
+
+@dataclass(frozen=True)
+class GooglePlacesQuery:
+    text_query: str
+    data_source: str
+    item_category: ItemCategory
+    activity: Activity
+
+
+def extract_postcode(formatted_address: str | None) -> str | None:
+    if not formatted_address:
+        return None
+    parts = formatted_address.split()
+    for part in parts:
+        if len(part) == 5 and part.isdigit():
+            return part
+    return None
+
+
+def normalize_google_place(
+    raw: dict[str, Any],
+    *,
+    data_source: str,
+    data_source_id: str,
+    item_category: ItemCategory,
+    activity: Activity,
+) -> dict[str, Any]:
+    display_name = raw.get("displayName") or {}
+    formatted_address = raw.get("formattedAddress")
+    location = raw.get("location") or {}
+    return {
+        "data_source_id": data_source_id,
+        "data_source": data_source,
+        "name": display_name.get("text") or data_source_id,
+        "lat": location.get("latitude", 0.0),
+        "lon": location.get("longitude", 0.0),
+        "address": Address(
+            street=formatted_address,
+            postcode=extract_postcode(formatted_address),
+        ),
+        "contact": Contact(),
+        "services": [Service(activity=activity, item_category=item_category)],
+        "availability": Availability(),
+    }

etl/pipelines/google_places/normalizer.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dtos import Activity, ItemCategory, NormalizedLocation, RawLocation
+from base.normalizer import BaseNormalizer
+
+from pipelines.google_places.common import normalize_google_place
+
+
+class GooglePlacesRepairNormalizer(BaseNormalizer):
+    def normalize(self, raw_locations: list[RawLocation]) -> list[NormalizedLocation]:
+        return [
+            NormalizedLocation(**normalize_google_place(
+                raw.payload,
+                data_source=raw.data_source,
+                data_source_id=raw.data_source_id,
+                item_category=ItemCategory.SHOES,
+                activity=Activity.REPAIR_PAID,
+            ))
+            for raw in raw_locations
+        ]
+
+
+class GooglePlacesDonationNormalizer(BaseNormalizer):
+    def normalize(self, raw_locations: list[RawLocation]) -> list[NormalizedLocation]:
+        return [
+            NormalizedLocation(**normalize_google_place(
+                raw.payload,
+                data_source=raw.data_source,
+                data_source_id=raw.data_source_id,
+                item_category=ItemCategory.CLOTHING,
+                activity=Activity.DONATION_DROP,
+            ))
+            for raw in raw_locations
+        ]

etl/pipelines/google_places/querier.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+import json
+import os
+from typing import Any
+
+from base.querier import BaseQuerier
+from dtos import RawLocation
+
+
+GOOGLE_PLACES_ENDPOINT = "https://places.googleapis.com/v1/places:searchText"
+
+
+class GooglePlacesQuerier(BaseQuerier):
+    def __init__(self, *, api_key: str | None = None, text_query: str, data_source: str):
+        self.api_key = api_key or os.environ.get("GOOGLE_API_KEY")
+        self.text_query = text_query
+        self.data_source = data_source
+
+    def fetch(self) -> list[RawLocation]:
+        if not self.api_key:
+            raise RuntimeError("GOOGLE_API_KEY is required")
+
+        payload = self._request({"textQuery": self.text_query})
+        places = payload.get("places", [])
+        fetched_at = datetime.now(timezone.utc)
+        raw_locations: list[RawLocation] = []
+        for index, place in enumerate(places):
+            raw_locations.append(
+                RawLocation(
+                    data_source=self.data_source,
+                    data_source_id=place.get("id") or f"{self.data_source}-{index}",
+                    fetched_at=fetched_at,
+                    payload=place,
+                )
+            )
+        return raw_locations
+
+    def _request(self, body: dict[str, Any]) -> dict[str, Any]:
+        request = Request(
+            GOOGLE_PLACES_ENDPOINT,
+            data=json.dumps(body).encode("utf-8"),
+            headers={
+                "Content-Type": "application/json",
+                "X-Goog-Api-Key": self.api_key,
+                "X-Goog-FieldMask": "places.id,places.displayName,places.formattedAddress,places.location,places.types",
+            },
+            method="POST",
+        )
+        try:
+            with urlopen(request, timeout=30) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except (HTTPError, URLError) as exc:
+            raise RuntimeError(f"Google Places request failed: {exc}") from exc
+
+
+class GooglePlacesRepairQuerier(GooglePlacesQuerier):
+    def __init__(self, api_key: str | None = None):
+        super().__init__(api_key=api_key, text_query="shoe repair Boston MA", data_source="google_places_repair")
+
+
+class GooglePlacesDonationQuerier(GooglePlacesQuerier):
+    def __init__(self, api_key: str | None = None):
+        super().__init__(api_key=api_key, text_query="donation centers Boston MA", data_source="google_places_donations")

etl/pipelines/google_places/test_pipeline.py

@@ -0,0 +1,54 @@
+from datetime import datetime, timezone
+from pathlib import Path
+import json
+
+from dtos import Activity, ItemCategory, RawLocation
+from pipelines.google_places.normalizer import (
+    GooglePlacesDonationNormalizer,
+    GooglePlacesRepairNormalizer,
+)
+
+
+SAMPLES = Path(__file__).resolve().parents[3] / "data-explorations" / "google-places" / "samples"
+
+
+def _load_places(name: str):
+    return json.loads((SAMPLES / name).read_text())["places"]
+
+
+def _raw_locations(data_source: str, prefix: str, filename: str):
+    fetched_at = datetime.now(timezone.utc)
+    return [
+        RawLocation(
+            data_source=data_source,
+            data_source_id=f"{prefix}-{index}",
+            fetched_at=fetched_at,
+            payload=place,
+        )
+        for index, place in enumerate(_load_places(filename))
+    ]
+
+
+def test_repair_normalizer_maps_google_places_sample():
+    normalized = GooglePlacesRepairNormalizer().normalize(
+        _raw_locations("google_places_repair", "repair", "shoe-repair-00.json")
+    )
+
+    assert normalized
+    assert normalized[0].data_source == "google_places_repair"
+    assert normalized[0].services[0].activity == Activity.REPAIR_PAID
+    assert normalized[0].services[0].item_category == ItemCategory.SHOES
+    assert normalized[0].name == "David's Instant Shoe Repair"
+    assert normalized[0].address.street == "281 Franklin St, Boston, MA 02110, USA"
+
+
+def test_donation_normalizer_maps_google_places_sample():
+    normalized = GooglePlacesDonationNormalizer().normalize(
+        _raw_locations("google_places_donations", "donation", "donations-00.json")
+    )
+
+    assert normalized
+    assert normalized[0].data_source == "google_places_donations"
+    assert normalized[0].services[0].activity == Activity.DONATION_DROP
+    assert normalized[0].services[0].item_category == ItemCategory.CLOTHING
+    assert normalized[0].name == "Morgan Memorial Goodwill Industries"