from __future__ import annotations import base64 import json import time from typing import Any, Literal, Optional from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response from app.core.logger import get_logger from app.models.schemas import ( GoogleAutocompleteMatchedSubstring, GoogleAutocompletePrediction, GoogleAutocompleteStructuredFormatting, GoogleAutocompleteTerm, GoogleGeocodeRequest, GoogleGeocodeResponse, GoogleGeocodeResult, GooglePlaceAutocompleteRequest, GooglePlaceAutocompleteResponse, GooglePlaceDetailsRequest, GooglePlaceDetailsResponse, GooglePlacePhoto, GooglePlaceResult, GooglePlacesNearbyRequest, GooglePlacesNearbyResponse, GooglePlacesSearchRequest, GooglePlacesSearchResponse, GoogleQueryAutocompleteRequest, GoogleQueryAutocompleteResponse, GoogleReverseGeocodeRequest, GoogleStaticMapRequest, GoogleStaticMapResponse, ) from app.services.google_maps_service import GoogleMapsService, PRICE_LEVEL_MAP router = APIRouter(prefix="/google", tags=["Google Maps"]) _logger = get_logger(__name__) _maps_service = GoogleMapsService() def get_maps_service() -> GoogleMapsService: return _maps_service async def close_maps_service() -> None: await _maps_service.close() # --------------------------------------------------------------------------- # Legacy response builders (for Geocoding — still on Google Geocoding API) # --------------------------------------------------------------------------- def _build_geocode_results(data: dict) -> list[GoogleGeocodeResult]: results: list[GoogleGeocodeResult] = [] for r in data.get("results", []): geometry = r.get("geometry", {}) location = geometry.get("location", {}) results.append(GoogleGeocodeResult( formatted_address=r.get("formatted_address", ""), place_id=r.get("place_id", ""), latitude=location.get("lat", 0.0), longitude=location.get("lng", 0.0), location_type=geometry.get("location_type", ""), address_components=r.get("address_components", []), types=r.get("types", []), )) return results # --------------------------------------------------------------------------- # New Places API response builders # --------------------------------------------------------------------------- def _parse_photos(photos_raw: list[dict]) -> list[GooglePlacePhoto]: photos: list[GooglePlacePhoto] = [] for p in photos_raw: photos.append(GooglePlacePhoto( photo_reference=p.get("name", ""), height=p.get("heightPx", 0), width=p.get("widthPx", 0), html_attributions=[ a.get("displayName", "") for a in p.get("authorAttributions", []) ], )) return photos def _build_place_results_new(data: dict) -> list[GooglePlaceResult]: results: list[GooglePlaceResult] = [] for p in data.get("places", []): loc = p.get("location", {}) price_level_str = p.get("priceLevel", "") results.append(GooglePlaceResult( place_id=p.get("id", ""), name=p.get("displayName", {}).get("text", ""), formatted_address=p.get("formattedAddress", ""), latitude=loc.get("latitude", 0.0), longitude=loc.get("longitude", 0.0), rating=p.get("rating"), user_ratings_total=p.get("userRatingCount"), price_level=PRICE_LEVEL_MAP.get(price_level_str) if price_level_str else None, types=p.get("types", []), vicinity=p.get("shortFormattedAddress", ""), business_status=p.get("businessStatus", ""), photos=_parse_photos(p.get("photos", [])), plus_code=p.get("plusCode", {}).get("globalCode") if p.get("plusCode") else None, icon=p.get("iconMaskBaseUri"), opening_hours=p.get("regularOpeningHours") or p.get("currentOpeningHours"), website=p.get("websiteUri"), formatted_phone_number=p.get("nationalPhoneNumber"), international_phone_number=p.get("internationalPhoneNumber"), google_maps_uri=p.get("googleMapsUri"), )) return results def _build_place_detail_new(data: dict) -> Optional[GooglePlaceResult]: if not data or not data.get("id"): return None loc = data.get("location", {}) price_level_str = data.get("priceLevel", "") return GooglePlaceResult( place_id=data.get("id", ""), name=data.get("displayName", {}).get("text", ""), formatted_address=data.get("formattedAddress", ""), latitude=loc.get("latitude", 0.0), longitude=loc.get("longitude", 0.0), rating=data.get("rating"), user_ratings_total=data.get("userRatingCount"), price_level=PRICE_LEVEL_MAP.get(price_level_str) if price_level_str else None, types=data.get("types", []), vicinity=data.get("shortFormattedAddress", ""), business_status=data.get("businessStatus", ""), photos=_parse_photos(data.get("photos", [])), plus_code=data.get("plusCode", {}).get("globalCode") if data.get("plusCode") else None, icon=data.get("iconMaskBaseUri"), opening_hours=data.get("regularOpeningHours") or data.get("currentOpeningHours"), website=data.get("websiteUri"), formatted_phone_number=data.get("nationalPhoneNumber"), international_phone_number=data.get("internationalPhoneNumber"), google_maps_uri=data.get("googleMapsUri"), ) def _build_autocomplete_predictions_new(data: dict) -> list[GoogleAutocompletePrediction]: predictions: list[GoogleAutocompletePrediction] = [] for s in data.get("suggestions", []): pp = s.get("placePrediction") or s.get("queryPrediction") if not pp: continue text = pp.get("text", {}) description = text.get("text", "") if isinstance(text, dict) else str(text) place_id = pp.get("placeId", "") sf_raw = pp.get("structuredFormat", {}) or pp.get("structuredFormatting", {}) sf = None if sf_raw: main_text = sf_raw.get("mainText", {}) if isinstance(main_text, dict): main_text = main_text.get("text", "") secondary_text = sf_raw.get("secondaryText", {}) if isinstance(secondary_text, dict): secondary_text = secondary_text.get("text", "") main_matches_raw = sf_raw.get("mainTextMatchedSubstrings", []) secondary_matches_raw = sf_raw.get("secondaryTextMatchedSubstrings", []) sf = GoogleAutocompleteStructuredFormatting( main_text=main_text, main_text_matched_substrings=[ GoogleAutocompleteMatchedSubstring(**m) for m in main_matches_raw ], secondary_text=secondary_text, secondary_text_matched_substrings=[ GoogleAutocompleteMatchedSubstring(**m) for m in secondary_matches_raw ], ) terms = [ GoogleAutocompleteTerm(offset=0, value=description) ] types = pp.get("types", []) predictions.append(GoogleAutocompletePrediction( description=description, place_id=place_id, structured_formatting=sf, terms=terms, types=types, matched_substrings=[], distance_meters=pp.get("distanceMeters"), )) return predictions def _build_autocomplete_predictions_legacy(data: dict) -> list[GoogleAutocompletePrediction]: predictions: list[GoogleAutocompletePrediction] = [] for p in data.get("predictions", []): sf_raw = p.get("structured_formatting") sf = None if sf_raw: main_matches = [ GoogleAutocompleteMatchedSubstring(**m) for m in sf_raw.get("main_text_matched_substrings", []) ] secondary_matches = [ GoogleAutocompleteMatchedSubstring(**m) for m in sf_raw.get("secondary_text_matched_substrings", []) ] sf = GoogleAutocompleteStructuredFormatting( main_text=sf_raw.get("main_text", ""), main_text_matched_substrings=main_matches, secondary_text=sf_raw.get("secondary_text", ""), secondary_text_matched_substrings=secondary_matches, ) terms = [GoogleAutocompleteTerm(**t) for t in p.get("terms", [])] matched_subs = [ GoogleAutocompleteMatchedSubstring(**m) for m in p.get("matched_substrings", []) ] predictions.append(GoogleAutocompletePrediction( description=p.get("description", ""), place_id=p.get("place_id", ""), structured_formatting=sf, terms=terms, types=p.get("types", []), matched_substrings=matched_subs, distance_meters=p.get("distance_meters"), )) return predictions # --------------------------------------------------------------------------- # Routes — Autocomplete # --------------------------------------------------------------------------- @router.post("/places/autocomplete", response_model=GooglePlaceAutocompleteResponse, summary="Get place predictions for autocomplete input (New Places API)") async def place_autocomplete( body: GooglePlaceAutocompleteRequest, 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() result = await service.place_autocomplete( input=body.input, offset=body.offset, origin=body.origin, location=body.location, radius=body.radius, language=body.language, types=body.types, components=body.components, strictbounds=body.strictbounds, sessiontoken=body.sessiontoken, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlaceAutocompleteResponse(success=False, time_ms=elapsed_ms, predictions=[], count=0, error=error) data = result.get("data", {}) predictions = _build_autocomplete_predictions_new(data) return GooglePlaceAutocompleteResponse(success=True, time_ms=elapsed_ms, predictions=predictions, count=len(predictions)) @router.get("/places/autocomplete", response_model=GooglePlaceAutocompleteResponse, summary="Get place predictions for autocomplete input (GET, New Places API)") async def place_autocomplete_get( input: str = Query(..., min_length=1, max_length=1000, description="Text string to search for"), offset: Optional[int] = Query(None, ge=1), origin: Optional[str] = Query(None, pattern=r"^-?\d+\.?\d*,-?\d+\.?\d*$"), location: Optional[str] = Query(None, pattern=r"^-?\d+\.?\d*,-?\d+\.?\d*$"), radius: Optional[int] = Query(None, ge=1, le=50000), language: Optional[str] = Query(None, max_length=10), types: Optional[str] = Query(None), components: Optional[str] = Query(None), strictbounds: Optional[bool] = Query(None), sessiontoken: Optional[str] = Query(None), 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() result = await service.place_autocomplete( input=input, offset=offset, origin=origin, location=location, radius=radius, language=language, types=types, components=components, strictbounds=strictbounds, sessiontoken=sessiontoken, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlaceAutocompleteResponse(success=False, time_ms=elapsed_ms, predictions=[], count=0, error=error) data = result.get("data", {}) predictions = _build_autocomplete_predictions_new(data) return GooglePlaceAutocompleteResponse(success=True, time_ms=elapsed_ms, predictions=predictions, count=len(predictions)) @router.post("/places/queryautocomplete", response_model=GoogleQueryAutocompleteResponse, summary="Get query predictions for autocomplete input (New Places API)") async def query_autocomplete( body: GoogleQueryAutocompleteRequest, 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() result = await service.query_autocomplete( input=body.input, offset=body.offset, location=body.location, radius=body.radius, language=body.language, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleQueryAutocompleteResponse(success=False, time_ms=elapsed_ms, predictions=[], count=0, error=error) data = result.get("data", {}) predictions = _build_autocomplete_predictions_new(data) return GoogleQueryAutocompleteResponse(success=True, time_ms=elapsed_ms, predictions=predictions, count=len(predictions)) @router.get("/places/queryautocomplete", response_model=GoogleQueryAutocompleteResponse, summary="Get query predictions for autocomplete input (GET, New Places API)") async def query_autocomplete_get( input: str = Query(..., min_length=1, max_length=1000, description="Text string to search for"), offset: Optional[int] = Query(None, ge=1), location: Optional[str] = Query(None, pattern=r"^-?\d+\.?\d*,-?\d+\.?\d*$"), radius: Optional[int] = Query(None, ge=1, le=50000), language: Optional[str] = Query(None, max_length=10), 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() result = await service.query_autocomplete( input=input, offset=offset, location=location, radius=radius, language=language, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleQueryAutocompleteResponse(success=False, time_ms=elapsed_ms, predictions=[], count=0, error=error) data = result.get("data", {}) predictions = _build_autocomplete_predictions_new(data) return GoogleQueryAutocompleteResponse(success=True, time_ms=elapsed_ms, predictions=predictions, count=len(predictions)) # --------------------------------------------------------------------------- # Routes — Geocoding (still uses legacy API) # --------------------------------------------------------------------------- @router.post("/geocode", response_model=GoogleGeocodeResponse, summary="Forward geocode an address") async def geocode( body: GoogleGeocodeRequest, 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() result = await service.geocode( address=body.address, region=body.region, language=body.language, bounds=body.bounds, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleGeocodeResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_geocode_results(data) return GoogleGeocodeResponse(success=True, time_ms=elapsed_ms, results=results, count=len(results)) @router.get("/geocode", response_model=GoogleGeocodeResponse, summary="Forward geocode an address (GET)") async def geocode_get( address: str = Query(..., min_length=1, max_length=1000, description="Street address to geocode"), region: Optional[str] = Query(None, max_length=2), language: Optional[str] = Query(None, max_length=10), bounds: Optional[str] = Query(None), 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() result = await service.geocode(address=address, region=region, language=language, bounds=bounds, api_key=x_goog_api_key) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleGeocodeResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_geocode_results(data) return GoogleGeocodeResponse(success=True, time_ms=elapsed_ms, results=results, count=len(results)) @router.post("/reverse-geocode", response_model=GoogleGeocodeResponse, summary="Reverse geocode coordinates to an address") async def reverse_geocode( body: GoogleReverseGeocodeRequest, 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() result = await service.reverse_geocode( latlng=body.latlng, language=body.language, result_type=body.result_type, location_type=body.location_type, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleGeocodeResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_geocode_results(data) return GoogleGeocodeResponse(success=True, time_ms=elapsed_ms, results=results, count=len(results)) @router.get("/reverse-geocode", response_model=GoogleGeocodeResponse, summary="Reverse geocode coordinates to an address (GET)") async def reverse_geocode_get( latlng: str = Query(..., pattern=r"^-?\d+\.?\d*,-?\d+\.?\d*$", description="Latitude,Longitude"), language: Optional[str] = Query(None, max_length=10), result_type: Optional[str] = Query(None), location_type: Optional[str] = Query(None), 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() result = await service.reverse_geocode( latlng=latlng, language=language, result_type=result_type, location_type=location_type, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GoogleGeocodeResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_geocode_results(data) return GoogleGeocodeResponse(success=True, time_ms=elapsed_ms, results=results, count=len(results)) # --------------------------------------------------------------------------- # Routes — Places Search / Nearby / Details (New Places API) # --------------------------------------------------------------------------- @router.post("/places/search", response_model=GooglePlacesSearchResponse, summary="Search for places using a text query (New Places API)") async def places_search( body: GooglePlacesSearchRequest, 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() result = await service.places_search( query=body.query, region=body.region, language=body.language, min_price=body.min_price, max_price=body.max_price, open_now=body.open_now, type_filter=body.type, radius=body.radius, page_token=body.page_token, page_size=body.page_size, min_rating=body.min_rating, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlacesSearchResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_place_results_new(data) return GooglePlacesSearchResponse( success=True, time_ms=elapsed_ms, results=results, count=len(results), next_page_token=data.get("nextPageToken"), ) @router.post("/places/nearby", response_model=GooglePlacesNearbyResponse, summary="Search for places near a location (New Places API)") async def places_nearby( body: GooglePlacesNearbyRequest, 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() result = await service.places_nearby( location=body.location, radius=body.radius, keyword=body.keyword, language=body.language, min_price=body.min_price, max_price=body.max_price, open_now=body.open_now, type_filter=body.type, page_token=body.page_token, page_size=body.page_size, rank_preference=body.rank_preference, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlacesNearbyResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_place_results_new(data) return GooglePlacesNearbyResponse( success=True, time_ms=elapsed_ms, results=results, count=len(results), next_page_token=data.get("nextPageToken"), ) @router.get("/places/search", response_model=GooglePlacesSearchResponse, summary="Search for places using a text query (GET, New Places API)") async def places_search_get( query: str = Query(..., min_length=1, max_length=1000, description="Text query for place search"), region: Optional[str] = Query(None, max_length=2), language: Optional[str] = Query(None, max_length=10), type_filter: Optional[str] = Query(None, alias="type"), radius: Optional[int] = Query(None, ge=1, le=50000), min_price: Optional[int] = Query(None, ge=0, le=4, description="Minimum price level (0=free, 4=most expensive)"), max_price: Optional[int] = Query(None, ge=0, le=4, description="Maximum price level (0=free, 4=most expensive)"), open_now: Optional[bool] = Query(None, description="Only return places that are open now"), page_token: Optional[str] = Query(None, description="Token for pagination"), page_size: Optional[int] = Query(None, ge=1, le=200, description="Number of results per page"), min_rating: Optional[float] = Query(None, ge=0.0, le=5.0, description="Minimum rating filter"), 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() result = await service.places_search( query=query, region=region, language=language, type_filter=type_filter, radius=radius, min_price=min_price, max_price=max_price, open_now=open_now, page_token=page_token, page_size=page_size, min_rating=min_rating, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlacesSearchResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_place_results_new(data) return GooglePlacesSearchResponse( success=True, time_ms=elapsed_ms, results=results, count=len(results), next_page_token=data.get("nextPageToken"), ) @router.get("/places/nearby", response_model=GooglePlacesNearbyResponse, summary="Search for places near a location (GET, New Places API)") async def places_nearby_get( location: str = Query(..., pattern=r"^-?\d+\.?\d*,-?\d+\.?\d*$", description="Latitude,Longitude"), radius: int = Query(1000, ge=1, le=50000), keyword: Optional[str] = Query(None, max_length=500), language: Optional[str] = Query(None, max_length=10), type_filter: Optional[str] = Query(None, alias="type"), min_price: Optional[int] = Query(None, ge=0, le=4, description="Minimum price level"), max_price: Optional[int] = Query(None, ge=0, le=4, description="Maximum price level"), open_now: Optional[bool] = Query(None, description="Only return places that are open now"), page_token: Optional[str] = Query(None, description="Token for pagination"), page_size: Optional[int] = Query(None, ge=1, le=200, description="Number of results per page"), rank_preference: Optional[str] = Query(None, pattern="^(POPULARITY|DISTANCE)$", description="Ranking preference"), 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() result = await service.places_nearby( location=location, radius=radius, keyword=keyword, language=language, type_filter=type_filter, min_price=min_price, max_price=max_price, open_now=open_now, page_token=page_token, page_size=page_size, rank_preference=rank_preference, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlacesNearbyResponse(success=False, time_ms=elapsed_ms, results=[], count=0, error=error) data = result.get("data", {}) results = _build_place_results_new(data) return GooglePlacesNearbyResponse( success=True, time_ms=elapsed_ms, results=results, count=len(results), next_page_token=data.get("nextPageToken"), ) @router.post("/places/details", response_model=GooglePlaceDetailsResponse, summary="Get detailed information about a place (New Places API)") async def place_details( body: GooglePlaceDetailsRequest, 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() result = await service.place_details( place_id=body.place_id, region=body.region, language=body.language, fields=body.fields, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlaceDetailsResponse(success=False, time_ms=elapsed_ms, result=None, error=error) data = result.get("data", {}) detail = _build_place_detail_new(data) return GooglePlaceDetailsResponse(success=True, time_ms=elapsed_ms, result=detail) @router.get("/places/details", response_model=GooglePlaceDetailsResponse, summary="Get detailed information about a place (GET, New Places API)") async def place_details_get( place_id: str = Query(..., min_length=1, max_length=500, description="Google Place ID"), region: Optional[str] = Query(None, max_length=2), language: Optional[str] = Query(None, max_length=10), fields: Optional[str] = Query(None), 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() result = await service.place_details( place_id=place_id, region=region, language=language, fields=fields, api_key=x_goog_api_key, ) elapsed_ms = round((time.perf_counter() - start) * 1000, 2) error = result.get("error") if error: return GooglePlaceDetailsResponse(success=False, time_ms=elapsed_ms, result=None, error=error) data = result.get("data", {}) detail = _build_place_detail_new(data) return GooglePlaceDetailsResponse(success=True, time_ms=elapsed_ms, result=detail) @router.get("/places/photo", summary="Get a place photo by photo reference (New Places API)") async def place_photo( photo_reference: str = Query(..., min_length=1, description="Photo reference name (e.g. 'places/ChIJ.../photos/...')"), max_width_px: int = Query(400, ge=1, le=4800), max_height_px: int = Query(400, ge=1, le=4800), x_goog_api_key: str = Header(..., alias="X-Goog-Api-Key", description="Your Google API key"), service: GoogleMapsService = Depends(get_maps_service), ): 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, )