"""Minimal PropertyMe HTTP client.
This client is intentionally small while Alba Core is still backend-first. It
can refresh the cache later, but matching must continue to work from the local
privacy-shaped seed when API access is unavailable.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import httpx
[docs]
class PropertyMeClient:
"""Small HTTP client for PropertyMe endpoints used by the backend.
The Alba Core should keep working from cache when this client is unavailable.
"""
def __init__(self, access_token: str, base_url: str = "https://app.propertyme.com/api/v1") -> None:
"""Create a client using a caller-provided OAuth access token."""
self.access_token = access_token
self.base_url = base_url.rstrip("/")
[docs]
async def list_rentals(self) -> list[dict[str, Any]]:
"""Fetch rental lots from PropertyMe and validate the response shape."""
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(
f"{self.base_url}/lots/rentals",
headers={"Authorization": f"Bearer {self.access_token}"},
)
response.raise_for_status()
payload = response.json()
if isinstance(payload, list):
return payload
if isinstance(payload, dict) and isinstance(payload.get("items"), list):
return payload["items"]
raise ValueError("Unexpected PropertyMe rentals response shape")
[docs]
async def save_rentals_cache(self, path: str | Path) -> int:
"""Save a raw rentals response to disk and return the number of records.
This is a low-level helper. Before committing any cache, map or redact it
into the privacy-shaped format used by ``PropertyRecord``.
"""
rentals = await self.list_rentals()
cache_path = Path(path)
cache_path.write_text(
__import__("json").dumps({"source": "PropertyMe /v1/lots/rentals", "items": rentals}, indent=2),
encoding="utf-8",
)
return len(rentals)