feat: add static maps service

.gitignore

@@ -137,4 +137,5 @@ deploy_hf.py
 ddl
 API_DESCRIPTION.md
 ENTERPRISE-API-ROADMAP*.md
-API-Implementation-Plan
+API-Implementation-Plan
+google_oauth_test_creds.postman_environment.json

app/api/server.py

@@ -107,6 +107,8 @@ async def lifespan(app: FastAPI):
     await pool_manager.close_all()
     from app.api.v1.google_oauth import close_oauth_service
     await close_oauth_service()
+    from app.api.v1.google_maps import close_maps_service
+    await close_maps_service()
     from app.services.supabase import get_supabase_client
     client = get_supabase_client()
     if client:

app/api/v1/google_maps.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import base64
+import json
 import time
-from typing import Any, Optional
+from typing import Any, Literal, Optional
 
-from fastapi import APIRouter, Depends, Header, Query, Response
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response
 
 from app.core.logger import get_logger
 from app.models.schemas import (
@@ -27,6 +29,8 @@ from app.models.schemas import (
     GoogleQueryAutocompleteRequest,
     GoogleQueryAutocompleteResponse,
     GoogleReverseGeocodeRequest,
+    GoogleStaticMapRequest,
+    GoogleStaticMapResponse,
 )
 from app.services.google_maps_service import GoogleMapsService, PRICE_LEVEL_MAP
 
@@ -34,8 +38,15 @@ router = APIRouter(prefix="/google", tags=["Google Maps"])
 _logger = get_logger(__name__)
 
 
+_maps_service = GoogleMapsService()
+
+
 def get_maps_service() -> GoogleMapsService:
-    return GoogleMapsService()
+    return _maps_service
+
+
+async def close_maps_service() -> None:
+    await _maps_service.close()
 
 
 # ---------------------------------------------------------------------------
@@ -632,35 +643,132 @@ async def place_photo(
     x_goog_api_key: str = Header(..., alias="X-Goog-Api-Key", description="Your Google API key"),
     service: GoogleMapsService = Depends(get_maps_service),
 ):
-    from app.config import get_settings
-    import json
-    import httpx
-
-    cfg = get_settings()
-    url = f"{cfg.google_maps_places_base_url}/{photo_reference}/media"
-    headers = {"X-Goog-Api-Key": x_goog_api_key}
-    params = {"maxWidthPx": max_width_px, "maxHeightPx": max_height_px}
-
-    async with httpx.AsyncClient(timeout=cfg.google_maps_timeout, follow_redirects=True) as client:
-        resp = await client.get(url, headers=headers, params=params)
-        if resp.status_code != 200:
-            # Try to extract a clean error message from the response
-            try:
-                error_body = resp.json()
-                err = error_body.get("error", {})
-                msg = err.get("message", "")
-                if "API key not valid" in msg or err.get("status") == "API_KEY_INVALID":
-                    clean_error = "Invalid Google API key. Please provide a valid API key via the X-Goog-Api-Key header."
-                elif msg:
-                    clean_error = msg.rstrip(".") + "."
-                else:
-                    clean_error = f"Failed to fetch photo (HTTP {resp.status_code})."
-            except Exception:
-                clean_error = f"Failed to fetch photo (HTTP {resp.status_code})."
-
-            return Response(
-                content=json.dumps({"success": False, "error": clean_error}),
-                media_type="application/json",
-                status_code=200,
-            )
-        return Response(content=resp.content, media_type=resp.headers.get("content-type", "image/jpeg"))
+    result = await service.place_photo(
+        photo_reference=photo_reference,
+        max_width_px=max_width_px,
+        max_height_px=max_height_px,
+        api_key=x_goog_api_key,
+    )
+    error = result.get("error")
+    if error:
+        return Response(
+            content=json.dumps({"success": False, "error": error}),
+            media_type="application/json",
+            status_code=200,
+        )
+    return Response(
+        content=result.get("content", b""),
+        media_type=result.get("content_type", "image/jpeg"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Routes — Static Map (Google Maps Static API)
+# ---------------------------------------------------------------------------
+
+def _parse_static_map_size(size: str) -> tuple[int, int]:
+    """Split a validated 'WIDTHxHEIGHT' size string into (width, height)."""
+    width, _, height = size.partition("x")
+    return int(width), int(height)
+
+
+@router.post("/staticmap", response_model=GoogleStaticMapResponse,
+             summary="Generate a static map image (Google Maps Static API)")
+async def static_map(
+    body: GoogleStaticMapRequest,
+    x_goog_api_key: str = Header(..., alias="X-Goog-Api-Key", description="Your Google API key"),
+    service: GoogleMapsService = Depends(get_maps_service),
+):
+    start = time.perf_counter()
+    params = body.model_dump(exclude_none=True)
+    result = await service.static_map(params, api_key=x_goog_api_key)
+    elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+    width, height = _parse_static_map_size(body.size)
+
+    error = result.get("error")
+    if error:
+        return GoogleStaticMapResponse(
+            success=False, time_ms=elapsed_ms, width=width, height=height,
+            scale=body.scale, format=body.format, error=error,
+        )
+
+    return GoogleStaticMapResponse(
+        success=True,
+        time_ms=elapsed_ms,
+        url=result.get("url"),
+        content_type=result.get("content_type", "image/png"),
+        image_base64=base64.b64encode(result.get("content", b"")).decode("ascii"),
+        width=width,
+        height=height,
+        scale=body.scale,
+        format=body.format,
+        warning=result.get("warning"),
+    )
+
+
+@router.get("/staticmap",
+            summary="Generate a static map image (GET, returns raw image bytes)")
+async def static_map_get(
+    center: Optional[str] = Query(None, max_length=1000, description="Center of the map: 'lat,lng' or address"),
+    zoom: Optional[int] = Query(None, ge=0, le=21, description="Map zoom level (0-21)"),
+    size: str = Query("640x640", pattern=r"^\d{1,3}x\d{1,3}$", description="Image size as 'WIDTHxHEIGHT'"),
+    scale: int = Query(1, ge=1, le=2, description="Pixel density scale (1 or 2)"),
+    format: Literal["png", "png8", "png32", "gif", "jpg", "jpg-baseline"] = Query("png", description="Image format"),
+    maptype: Literal["roadmap", "satellite", "hybrid", "terrain"] = Query("roadmap", description="Map type"),
+    language: Optional[str] = Query(None, max_length=10, description="Language code for labels"),
+    region: Optional[str] = Query(None, max_length=2, description="Region biasing (ccTLD)"),
+    map_id: Optional[str] = Query(None, max_length=200, description="Map ID for cloud-based styling"),
+    markers: Optional[list[str]] = Query(None, description="Raw 'markers' parameter values (pipe-delimited, repeated)"),
+    path: Optional[list[str]] = Query(None, description="Raw 'path' parameter values (pipe-delimited, repeated)"),
+    visible: Optional[str] = Query(None, description="Pipe-delimited locations to keep visible"),
+    styles: Optional[list[str]] = Query(None, description="Raw 'style' parameter values (pipe-delimited, repeated)"),
+    x_goog_api_key: str = Header(..., alias="X-Goog-Api-Key", description="Your Google API key"),
+    service: GoogleMapsService = Depends(get_maps_service),
+):
+    start = time.perf_counter()
+
+    width, height = _parse_static_map_size(size)
+    if not (1 <= width <= 640 and 1 <= height <= 640):
+        raise HTTPException(status_code=422, detail="size dimensions must each be between 1 and 640 pixels")
+    has_elements = bool(markers or path or visible)
+    if not has_elements and (not center or zoom is None):
+        raise HTTPException(
+            status_code=422,
+            detail="center and zoom are required unless markers, path, or visible are provided",
+        )
+
+    params: dict[str, Any] = {
+        "center": center,
+        "zoom": zoom,
+        "size": size,
+        "scale": scale,
+        "format": format,
+        "maptype": maptype,
+        "language": language,
+        "region": region,
+        "map_id": map_id,
+        "markers": markers or [],
+        "paths": path or [],
+        "visible": visible,
+        "styles": styles or [],
+    }
+    result = await service.static_map(params, api_key=x_goog_api_key)
+    elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+
+    error = result.get("error")
+    if error:
+        return Response(
+            content=json.dumps({"success": False, "time_ms": elapsed_ms, "error": error}),
+            media_type="application/json",
+            status_code=200,
+        )
+
+    headers = {}
+    warning = result.get("warning")
+    if warning:
+        headers["X-Staticmap-API-Warning"] = warning
+    return Response(
+        content=result.get("content", b""),
+        media_type=result.get("content_type", "image/png"),
+        headers=headers,
+    )

app/models/schemas.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from enum import Enum
 from typing import Any, Dict, List, Literal, Optional
 
@@ -953,3 +954,236 @@ class GoogleOAuthVerifyResponse(BaseModel):
     valid: bool
     user: Optional[GoogleOAuthUserInfo] = None
     error: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Google Maps Static API
+# ---------------------------------------------------------------------------
+
+_MAP_NAMED_COLORS: frozenset = frozenset({
+    "black", "brown", "green", "purple", "yellow", "blue", "gray", "orange", "red", "white",
+})
+
+
+class GoogleStaticMapMarkerStyle(BaseModel):
+    """Visual style descriptor for a group of static map markers."""
+
+    size: Optional[Literal["tiny", "small", "mid", "normal"]] = Field(
+        None, description="Marker size (tiny, small, mid, or default 'normal')"
+    )
+    color: Optional[str] = Field(
+        None, description="24-bit hex color (0xRRGGBB) or a predefined named color"
+    )
+    label: Optional[str] = Field(
+        None, pattern=r"^[A-Z0-9]$", description="Single uppercase alphanumeric marker label (A-Z, 0-9)"
+    )
+    icon: Optional[str] = Field(None, max_length=2000, description="Custom icon URL (PNG/JPEG/GIF)")
+    anchor: Optional[str] = Field(
+        None, max_length=50, description="Icon anchor (predefined alignment or 'x,y' point)"
+    )
+    scale: Optional[int] = Field(None, ge=1, le=4, description="Marker scale factor (1, 2 or 4)")
+
+    @field_validator("color")
+    @classmethod
+    def _validate_marker_color(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        normalized = v.lower()
+        if normalized in _MAP_NAMED_COLORS:
+            return v
+        if re.fullmatch(r"0x[0-9a-fA-F]{6}", normalized):
+            return v
+        raise ValueError("color must be a named color or a 24-bit hex value like '0x00FF00'")
+
+    @model_validator(mode="after")
+    def _check_anchor_requires_icon(self) -> "GoogleStaticMapMarkerStyle":
+        if self.anchor and not self.icon:
+            raise ValueError("anchor is only valid when a custom icon is provided")
+        return self
+
+
+class GoogleStaticMapMarker(BaseModel):
+    """One or more marker locations sharing the same visual style."""
+
+    style: Optional[GoogleStaticMapMarkerStyle] = Field(None, description="Optional marker style")
+    locations: List[str] = Field(..., min_length=1, description="One or more locations (lat,lng or address)")
+
+    @field_validator("locations")
+    @classmethod
+    def _validate_locations(cls, v: List[str]) -> List[str]:
+        cleaned = [loc.strip() for loc in v]
+        if any(not loc for loc in cleaned):
+            raise ValueError("marker locations must not contain empty values")
+        return cleaned
+
+
+class GoogleStaticMapPathStyle(BaseModel):
+    """Visual style descriptor for a static map path / polyline."""
+
+    color: Optional[str] = Field(None, description="Line color as 24-bit or 32-bit hex (0xRRGGBB or 0xRRGGBBAA)")
+    weight: Optional[int] = Field(None, ge=0, description="Path line weight in pixels")
+    fill: Optional[str] = Field(None, description="Fill color as hex (0xRRGGBB or 0xRRGGBBAA)")
+    geodesic: Optional[bool] = Field(None, description="Draw the path as a geodesic curve")
+
+    @field_validator("color", "fill")
+    @classmethod
+    def _validate_path_hex(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        if re.fullmatch(r"0x[0-9a-fA-F]{6}(?:[0-9a-fA-F]{2})?", v):
+            return v
+        raise ValueError("color must be a hex value like '0x00FF00' or '0x00FF00AA'")
+
+
+class GoogleStaticMapPath(BaseModel):
+    """A single path / polyline overlay on the static map."""
+
+    style: Optional[GoogleStaticMapPathStyle] = Field(None, description="Optional path style")
+    points: Optional[List[str]] = Field(
+        None, min_length=2, description="Two or more connected points (lat,lng or address)"
+    )
+    encoded_polyline: Optional[str] = Field(
+        None, description="Encoded polyline (without the 'enc:' prefix)"
+    )
+
+    @field_validator("points")
+    @classmethod
+    def _validate_points(cls, v: Optional[List[str]]) -> Optional[List[str]]:
+        if v is None:
+            return v
+        cleaned = [pt.strip() for pt in v]
+        if any(not pt for pt in cleaned):
+            raise ValueError("path points must not contain empty values")
+        return cleaned
+
+    @model_validator(mode="after")
+    def _check_points_or_polyline(self) -> "GoogleStaticMapPath":
+        has_points = self.points is not None and len(self.points) >= 2
+        has_polyline = bool(self.encoded_polyline)
+        if not has_points and not has_polyline:
+            raise ValueError("path requires either at least 2 points or an encoded_polyline")
+        if has_points and has_polyline:
+            raise ValueError("path accepts either points or encoded_polyline, not both")
+        return self
+
+
+class GoogleStaticMapStyleRule(BaseModel):
+    """A single custom map style rule (feature/element + style operations)."""
+
+    feature: Optional[str] = Field(None, description="Map feature to style (e.g. 'road', 'water', 'poi')")
+    element: Optional[str] = Field(None, description="Element of the feature (e.g. 'geometry', 'labels')")
+    hue: Optional[str] = Field(None, description="Hue color as hex ('#RRGGBB')")
+    lightness: Optional[float] = Field(None, ge=-100, le=100, description="Lightness adjustment (-100 to 100)")
+    saturation: Optional[float] = Field(None, ge=-100, le=100, description="Saturation adjustment (-100 to 100)")
+    gamma: Optional[float] = Field(None, ge=0.01, le=10.0, description="Gamma correction (0.01 to 10.0)")
+    invert_lightness: Optional[bool] = Field(None, description="Invert the existing lightness")
+    visibility: Optional[Literal["on", "off", "simplified"]] = Field(
+        None, description="Visibility of the element (on, off, simplified)"
+    )
+    color: Optional[str] = Field(None, description="Color as hex ('#RRGGBB')")
+    weight: Optional[int] = Field(None, ge=0, description="Feature weight in pixels")
+
+    @field_validator("color", "hue")
+    @classmethod
+    def _validate_style_hex(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        if re.fullmatch(r"#[0-9a-fA-F]{6}", v):
+            return v
+        raise ValueError("color/hue must be a hex value like '#FF0000'")
+
+    @model_validator(mode="after")
+    def _check_has_operation(self) -> "GoogleStaticMapStyleRule":
+        has_operation = any(
+            v is not None
+            for v in (
+                self.hue, self.lightness, self.saturation, self.gamma,
+                self.invert_lightness, self.visibility, self.color, self.weight,
+            )
+        )
+        if not has_operation:
+            raise ValueError("style rule must contain at least one style operation")
+        return self
+
+
+class GoogleStaticMapRequest(BaseModel):
+    """Request payload for generating a Google Maps Static API image."""
+
+    center: Optional[str] = Field(
+        None, max_length=1000, description="Center of the map: 'lat,lng' pair or a geocodable address"
+    )
+    zoom: Optional[int] = Field(None, ge=0, le=21, description="Map zoom level (0-21)")
+    size: str = Field(
+        "640x640", description="Image size as '{width}x{height}', each dimension 1-640 pixels"
+    )
+    scale: int = Field(1, ge=1, le=2, description="Pixel density scale factor (1 or 2)")
+    format: Literal["png", "png8", "png32", "gif", "jpg", "jpg-baseline"] = Field(
+        "png", description="Image format"
+    )
+    maptype: Literal["roadmap", "satellite", "hybrid", "terrain"] = Field(
+        "roadmap", description="Map type"
+    )
+    language: Optional[str] = Field(None, max_length=10, description="Language code for map labels")
+    region: Optional[str] = Field(None, max_length=2, description="Region biasing (two-character ccTLD)")
+    map_id: Optional[str] = Field(None, max_length=200, description="Map ID for cloud-based map styling")
+    markers: Optional[List[GoogleStaticMapMarker]] = Field(None, description="Marker groups to place on the map")
+    paths: Optional[List[GoogleStaticMapPath]] = Field(None, description="Paths / polylines to overlay")
+    visible: Optional[List[str]] = Field(
+        None, min_length=1, description="Locations to keep visible on the map (no markers drawn)"
+    )
+    styles: Optional[List[GoogleStaticMapStyleRule]] = Field(None, description="Custom map style rules")
+    signature_secret: Optional[str] = Field(
+        None, max_length=500, description="URL signing secret used to digitally sign the request"
+    )
+
+    @field_validator("size")
+    @classmethod
+    def _validate_size(cls, v: str) -> str:
+        match = re.fullmatch(r"(\d{1,3})x(\d{1,3})", v)
+        if not match:
+            raise ValueError("size must be of the form '{width}x{height}' (e.g. '640x480')")
+        width, height = int(match.group(1)), int(match.group(2))
+        if not (1 <= width <= 640 and 1 <= height <= 640):
+            raise ValueError("size dimensions must each be between 1 and 640 pixels")
+        return v
+
+    @field_validator("center")
+    @classmethod
+    def _validate_center(cls, v: Optional[str]) -> Optional[str]:
+        if v is None:
+            return v
+        cleaned = v.strip()
+        if not cleaned:
+            raise ValueError("center must not be empty")
+        return cleaned
+
+    @model_validator(mode="after")
+    def _check_center_and_zoom(self) -> "GoogleStaticMapRequest":
+        has_elements = bool(self.markers or self.paths or self.visible)
+        if not has_elements:
+            missing = []
+            if not self.center:
+                missing.append("center")
+            if self.zoom is None:
+                missing.append("zoom")
+            if missing:
+                raise ValueError(
+                    f"{', '.join(missing)} is/are required when no markers, paths, or visible locations are supplied"
+                )
+        return self
+
+
+class GoogleStaticMapResponse(BaseModel):
+    """Response payload for the static map generation endpoint."""
+
+    success: bool
+    time_ms: float
+    url: Optional[str] = Field(None, description="Fully built Google Static Maps URL")
+    content_type: str = Field("image/png", description="MIME type of the returned image")
+    image_base64: Optional[str] = Field(None, description="Base64-encoded image bytes")
+    width: int = Field(640, description="Image width in pixels")
+    height: int = Field(640, description="Image height in pixels")
+    scale: int = Field(1, description="Scale factor used for the image")
+    format: str = Field("png", description="Image format")
+    warning: Optional[str] = Field(None, description="Warning returned by the Maps Static API, if any")
+    error: Optional[str] = None

app/services/google_maps_service.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import base64
+import hashlib
+import hmac
 import logging
-from typing import Any, Dict, Optional
+import re
+from typing import Any, Dict, Optional, Tuple
+from urllib.parse import unquote, urlencode
 
 import httpx
 
@@ -47,6 +52,21 @@ DEFAULT_AUTOCOMPLETE_FIELD_MASK = (
     "suggestions.queryPrediction.text.text"
 )
 
+STATIC_MAP_BASE_URL = "https://maps.googleapis.com/maps/api/staticmap"
+
+STATIC_MAP_FORMAT_CONTENT_TYPES: dict[str, str] = {
+    "png": "image/png",
+    "png8": "image/png",
+    "png32": "image/png",
+    "gif": "image/gif",
+    "jpg": "image/jpeg",
+    "jpg-baseline": "image/jpeg",
+}
+
+_STYLE_OPERATION_KEYS = (
+    "hue", "lightness", "saturation", "gamma", "invert_lightness", "visibility", "color", "weight",
+)
+
 
 class GoogleMapsService:
 
@@ -55,6 +75,35 @@ class GoogleMapsService:
         self._places_base_url: str = _settings.google_maps_places_base_url
         self._timeout: int = _settings.google_maps_timeout
         self._max_retries: int = _settings.google_maps_max_retries
+        self._client: Optional[httpx.AsyncClient] = None
+        self._client_lock = asyncio.Lock()
+
+    # -----------------------------------------------------------------------
+    # HTTP client management
+    # -----------------------------------------------------------------------
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        """Return the shared connection-pooled AsyncClient, creating it lazily."""
+        if self._client is None or self._client.is_closed:
+            async with self._client_lock:
+                if self._client is None or self._client.is_closed:
+                    self._client = httpx.AsyncClient(
+                        timeout=self._timeout,
+                        follow_redirects=True,
+                        limits=httpx.Limits(
+                            max_connections=100,
+                            max_keepalive_connections=20,
+                            keepalive_expiry=30,
+                        ),
+                    )
+        return self._client
+
+    async def close(self) -> None:
+        """Close the shared AsyncClient and release pooled connections."""
+        async with self._client_lock:
+            if self._client is not None and not self._client.is_closed:
+                await self._client.aclose()
+            self._client = None
 
     # -----------------------------------------------------------------------
     # Geocoding — still uses the legacy Google Geocoding API
@@ -267,6 +316,180 @@ class GoogleMapsService:
 
         return await self._call_places_api("GET", path, params=params, field_mask=field_mask, api_key=api_key)
 
+    # -----------------------------------------------------------------------
+    # Google Maps Static API — image generation
+    # -----------------------------------------------------------------------
+
+    @staticmethod
+    def _build_marker_param(marker: Dict[str, Any]) -> str:
+        """Serialize a marker group dict into a Google `markers` parameter string."""
+        style = marker.get("style") or {}
+        parts: list[str] = []
+        for key in ("size", "color", "label", "icon", "anchor", "scale"):
+            value = style.get(key)
+            if value is not None:
+                parts.append(f"{key}:{value}")
+        parts.extend(marker.get("locations", []))
+        return "|".join(parts)
+
+    @staticmethod
+    def _build_path_param(path: Dict[str, Any]) -> str:
+        """Serialize a path dict into a Google `path` parameter string."""
+        style = path.get("style") or {}
+        parts: list[str] = []
+        if style.get("color"):
+            parts.append(f"color:{style['color']}")
+        if style.get("weight") is not None:
+            parts.append(f"weight:{style['weight']}")
+        if style.get("fill"):
+            parts.append(f"fill:{style['fill']}")
+        if style.get("geodesic"):
+            parts.append("geodesic:true")
+        if path.get("encoded_polyline"):
+            parts.append(f"enc:{path['encoded_polyline']}")
+        else:
+            parts.extend(path.get("points", []))
+        return "|".join(parts)
+
+    @staticmethod
+    def _build_style_param(rule: Dict[str, Any]) -> str:
+        """Serialize a style rule dict into a Google `style` parameter string."""
+        parts: list[str] = []
+        if rule.get("feature"):
+            parts.append(f"feature:{rule['feature']}")
+        if rule.get("element"):
+            parts.append(f"element:{rule['element']}")
+        for key in _STYLE_OPERATION_KEYS:
+            value = rule.get(key)
+            if value is None:
+                continue
+            if isinstance(value, bool):
+                parts.append(f"{key}:{str(value).lower()}")
+            else:
+                parts.append(f"{key}:{value}")
+        return "|".join(parts)
+
+    @staticmethod
+    def _sign_static_map_url(url: str, signing_secret: str) -> str:
+        """Apply a Google Maps URL digital signature (HMAC-SHA1) to a request URL."""
+        # Strip the protocol scheme and host, keeping only the path + query.
+        path_and_query = url.split("://", 1)[1].split("/", 1)[1]
+        # Remove any existing signature parameter before signing.
+        path_and_query = re.sub(r"&signature=[^&]*", "", path_and_query)
+        decoded = unquote(path_and_query)
+        key = base64.urlsafe_b64decode(signing_secret + "=" * (-len(signing_secret) % 4))
+        signature = base64.urlsafe_b64encode(
+            hmac.new(key, decoded.encode("utf-8"), hashlib.sha1).digest()
+        ).rstrip(b"=").decode("ascii")
+        return f"{url}&signature={signature}"
+
+    def build_static_map_url(self, params: Dict[str, Any], api_key: str,
+                             signature_secret: Optional[str] = None) -> Optional[str]:
+        """Build a Google Static Maps URL from validated params.
+
+        Returns ``None`` when the API key is missing. ``markers`` / ``paths`` /
+        ``styles`` entries may be either structured dicts (as produced by the
+        request schema) or pre-formatted strings (raw GET query values).
+        """
+        if not api_key or not api_key.strip():
+            return None
+
+        query: list[Tuple[str, str]] = []
+        for key in ("center", "zoom", "size", "scale", "format", "maptype", "language", "region", "map_id"):
+            value = params.get(key)
+            if value is not None:
+                query.append((key, str(value)))
+
+        for marker in params.get("markers", []):
+            query.append(("markers", marker if isinstance(marker, str) else self._build_marker_param(marker)))
+        for path in params.get("paths", []):
+            query.append(("path", path if isinstance(path, str) else self._build_path_param(path)))
+
+        visible = params.get("visible")
+        if visible:
+            if isinstance(visible, str):
+                visible = [v for v in visible.split("|") if v.strip()]
+            if visible:
+                query.append(("visible", "|".join(visible)))
+
+        for rule in params.get("styles", []):
+            query.append(("style", rule if isinstance(rule, str) else self._build_style_param(rule)))
+
+        query.append(("key", api_key))
+        url = f"{STATIC_MAP_BASE_URL}?{urlencode(query)}"
+        if signature_secret:
+            url = self._sign_static_map_url(url, signature_secret)
+        return url
+
+    @staticmethod
+    def _format_static_map_error(status_code: int, response: httpx.Response) -> str:
+        """Extract a clean, human-readable error message from a Static Maps error response."""
+        body = response.text.strip()
+        if status_code == 401:
+            return "Unauthorized. The API key may be missing or restricted."
+        if status_code == 403:
+            if "api key" in body.lower():
+                return "Invalid Google API key. Please provide a valid API key via the X-Goog-Api-Key header."
+            return "Access forbidden. The API key may not have the required APIs enabled or the request is unsigned."
+        if status_code == 429:
+            return "API rate limit exceeded. Please wait and retry."
+        if body:
+            return body[:500] if len(body) > 500 else body
+        return f"Maps Static API error (HTTP {status_code})."
+
+    async def static_map(self, params: Dict[str, Any], api_key: str) -> Dict[str, Any]:
+        """Fetch a static map image from the Google Maps Static API.
+
+        ``params`` should be the validated request payload (as a dict). Returns a
+        dict with ``success``, and on success ``content`` (image bytes),
+        ``content_type``, ``url`` and optionally ``warning``.
+        """
+        signature_secret = params.pop("signature_secret", None) if isinstance(params, dict) else None
+        url = self.build_static_map_url(params, api_key, signature_secret)
+        if url is None:
+            return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."}
+
+        last_error: Optional[str] = None
+        client = await self._get_client()
+        for attempt in range(1 + self._max_retries):
+            try:
+                response = await client.get(url)
+                if response.status_code == 200:
+                    return {
+                        "success": True,
+                        "content": response.content,
+                        "content_type": response.headers.get("content-type", "image/png"),
+                        "url": url,
+                        "warning": response.headers.get("X-Staticmap-API-Warning") or None,
+                        "error": None,
+                    }
+
+                last_error = self._format_static_map_error(response.status_code, response)
+                _logger.warning("Maps Static API error on static map: HTTP %s -> %s", response.status_code, last_error)
+
+                if response.status_code == 429:
+                    if attempt < self._max_retries:
+                        await asyncio.sleep(2.0 ** (attempt + 1))
+                    continue
+                if 400 <= response.status_code < 500:
+                    break
+
+            except httpx.TimeoutException:
+                last_error = "Request timed out"
+                _logger.warning("Maps Static API timeout (attempt %d/%d)", attempt + 1, 1 + self._max_retries)
+            except httpx.RequestError as e:
+                last_error = f"Request failed: {e}"
+                _logger.warning("Maps Static API request error: %s (attempt %d/%d)", e, attempt + 1, 1 + self._max_retries)
+            except Exception as e:
+                last_error = f"Unexpected error: {e}"
+                _logger.error("Maps Static API unexpected error: %s", last_error)
+                break
+
+            if attempt < self._max_retries:
+                await asyncio.sleep(1.0 * (attempt + 1))
+
+        return {"success": False, "error": last_error or "Unknown error"}
+
     # -----------------------------------------------------------------------
     # Internal — Legacy API (for Geocoding)
     # -----------------------------------------------------------------------
@@ -288,14 +511,14 @@ class GoogleMapsService:
     async def _call_api(self, path: str, params: Dict[str, Any]) -> Dict[str, Any]:
         url = f"{self._base_url}{path}"
         last_error: Optional[str] = None
+        client = await self._get_client()
 
         for attempt in range(1 + self._max_retries):
             try:
-                async with httpx.AsyncClient(timeout=self._timeout) as client:
-                    response = await client.get(url, params=params)
-                    response.raise_for_status()
-                    data: Dict[str, Any] = response.json()
-                    return self._normalize_response(data)
+                response = await client.get(url, params=params)
+                response.raise_for_status()
+                data: Dict[str, Any] = response.json()
+                return self._normalize_response(data)
 
             except httpx.TimeoutException:
                 last_error = "Request timed out"
@@ -390,34 +613,34 @@ class GoogleMapsService:
             headers["X-Goog-FieldMask"] = field_mask
 
         last_error: Optional[str] = None
+        client = await self._get_client()
 
         for attempt in range(1 + self._max_retries):
             try:
-                async with httpx.AsyncClient(timeout=self._timeout) as client:
-                    if method == "POST":
-                        response = await client.post(url, json=body, headers=headers)
-                    else:
-                        response = await client.get(url, params=params, headers=headers)
-
-                    if response.status_code == 200:
-                        data: Dict[str, Any] = response.json()
-                        return {"success": True, "data": data, "error": None}
-
-                    last_error = self._format_places_api_error(response.status_code, response)
-                    _logger.warning(
-                        "Places API error on %s: HTTP %s -> %s",
-                        path, response.status_code, last_error,
-                    )
+                if method == "POST":
+                    response = await client.post(url, json=body, headers=headers)
+                else:
+                    response = await client.get(url, params=params, headers=headers)
 
-                    if response.status_code == 429:
-                        backoff = 2.0 ** (attempt + 1)
-                        _logger.warning("Places API rate limited on %s, backing off %.1fs", path, backoff)
-                        if attempt < self._max_retries:
-                            await asyncio.sleep(backoff)
-                        continue
-
-                    if 400 <= response.status_code < 500:
-                        break
+                if response.status_code == 200:
+                    data: Dict[str, Any] = response.json()
+                    return {"success": True, "data": data, "error": None}
+
+                last_error = self._format_places_api_error(response.status_code, response)
+                _logger.warning(
+                    "Places API error on %s: HTTP %s -> %s",
+                    path, response.status_code, last_error,
+                )
+
+                if response.status_code == 429:
+                    backoff = 2.0 ** (attempt + 1)
+                    _logger.warning("Places API rate limited on %s, backing off %.1fs", path, backoff)
+                    if attempt < self._max_retries:
+                        await asyncio.sleep(backoff)
+                    continue
+
+                if 400 <= response.status_code < 500:
+                    break
 
             except httpx.TimeoutException:
                 last_error = "Request timed out"
@@ -434,3 +657,51 @@ class GoogleMapsService:
                 await asyncio.sleep(1.0 * (attempt + 1))
 
         return {"success": False, "error": last_error or "Unknown error"}
+
+    async def place_photo(self, photo_reference: str, max_width_px: int,
+                          max_height_px: int, api_key: Optional[str]) -> Dict[str, Any]:
+        """Fetch a place photo's image bytes from the New Places API."""
+        if not api_key or not api_key.strip():
+            return {"success": False, "error": "Google API key is missing. Provide a valid API key via the X-Goog-Api-Key header."}
+        url = f"{self._places_base_url}/{photo_reference}/media"
+        headers: Dict[str, str] = {"X-Goog-Api-Key": api_key}
+        params: Dict[str, Any] = {"maxWidthPx": max_width_px, "maxHeightPx": max_height_px}
+
+        last_error: Optional[str] = None
+        client = await self._get_client()
+        for attempt in range(1 + self._max_retries):
+            try:
+                response = await client.get(url, headers=headers, params=params)
+                if response.status_code == 200:
+                    return {
+                        "success": True,
+                        "content": response.content,
+                        "content_type": response.headers.get("content-type", "image/jpeg"),
+                        "error": None,
+                    }
+
+                last_error = self._format_places_api_error(response.status_code, response)
+                _logger.warning("Places API photo error: HTTP %s -> %s", response.status_code, last_error)
+
+                if response.status_code == 429:
+                    if attempt < self._max_retries:
+                        await asyncio.sleep(2.0 ** (attempt + 1))
+                    continue
+                if 400 <= response.status_code < 500:
+                    break
+
+            except httpx.TimeoutException:
+                last_error = "Request timed out"
+                _logger.warning("Places API photo timeout (attempt %d/%d)", attempt + 1, 1 + self._max_retries)
+            except httpx.RequestError as e:
+                last_error = f"Request failed: {e}"
+                _logger.warning("Places API photo request error: %s (attempt %d/%d)", e, attempt + 1, 1 + self._max_retries)
+            except Exception as e:
+                last_error = f"Unexpected error: {e}"
+                _logger.error("Places API photo unexpected error: %s", last_error)
+                break
+
+            if attempt < self._max_retries:
+                await asyncio.sleep(1.0 * (attempt + 1))
+
+        return {"success": False, "error": last_error or "Unknown error"}