Skip to main content

Sabre Integration

This document describes the integration with Sabre GDS for accommodation search and booking.

Authentication​

Sabre uses OAuth 2.0 for authentication. The token procedure involves:

  1. Encode Credentials — Base64 encode client_id:client_secret
  2. Request Token — POST to token endpoint
  3. Store Token — Cache token until expiry
  4. Refresh — Automatically refresh before expiry

Token Request​

class SabreClient:
    async def get_access_token(self) -> str:
        """Obtain OAuth access token from Sabre."""
        credentials = base64.b64encode(
            f"{self.client_id}:{self.client_secret}".encode()
        ).decode()
        
        response = await self.http_client.post(
            f"{self.base_url}/v2/auth/token",
            headers={
                "Authorization": f"Basic {credentials}",
                "Content-Type": "application/x-www-form-urlencoded",
            },
            data={"grant_type": "client_credentials"},
        )
        
        data = response.json()
        return data["access_token"]

Token Caching​

class SabreTokenManager:
    """Manages Sabre OAuth tokens with caching."""
    
    def __init__(self, client: SabreClient, cache: Redis) -> None:
        self.client = client
        self.cache = cache
    
    async def get_token(self) -> str:
        """Get valid token, refreshing if needed."""
        cached = await self.cache.get("sabre:access_token")
        if cached:
            return cached
        
        token = await self.client.get_access_token()
        # Cache with 5 minute buffer before expiry
        await self.cache.setex("sabre:access_token", 3600 - 300, token)
        return token

API Capabilities​

async def search_hotels(
    self,
    location: str,
    checkin: date,
    checkout: date,
    guests: int = 1,
) -> list[SabreHotel]:
    """Search for hotels via Sabre GDS."""
    token = await self.token_manager.get_token()
    
    response = await self.http_client.get(
        f"{self.base_url}/v3.0.0/shop/hotels",
        headers={"Authorization": f"Bearer {token}"},
        params={
            "location": location,
            "checkInDate": checkin.isoformat(),
            "checkOutDate": checkout.isoformat(),
            "guests": guests,
        },
    )
    
    return [SabreHotel(**h) for h in response.json()["hotels"]]

Rate Details​

async def get_rates(
    self,
    hotel_id: str,
    checkin: date,
    checkout: date,
) -> list[SabreRate]:
    """Get detailed rates for a specific hotel."""
    token = await self.token_manager.get_token()
    
    response = await self.http_client.get(
        f"{self.base_url}/v3.0.0/shop/hotels/{hotel_id}/rates",
        headers={"Authorization": f"Bearer {token}"},
        params={
            "checkInDate": checkin.isoformat(),
            "checkOutDate": checkout.isoformat(),
        },
    )
    
    return [SabreRate(**r) for r in response.json()["rates"]]

Data Mapping​

Map Sabre responses to internal entities:

class SabreMapper:
    """Maps Sabre API responses to domain entities."""
    
    def map_hotel(self, sabre_hotel: SabreHotel) -> Accommodation:
        return Accommodation(
            external_id=f"sabre:{sabre_hotel.id}",
            name=sabre_hotel.name,
            address=sabre_hotel.address.street,
            city=sabre_hotel.address.city,
            country_code=sabre_hotel.address.country_code,
            latitude=sabre_hotel.location.latitude,
            longitude=sabre_hotel.location.longitude,
            star_rating=sabre_hotel.category,
            chain=sabre_hotel.chain_name,
        )
    
    def map_rate(self, sabre_rate: SabreRate, accommodation_id: UUID) -> Product:
        return Product(
            accommodation_id=accommodation_id,
            name=sabre_rate.room_type,
            price_per_night=sabre_rate.rate_per_night,
            currency=sabre_rate.currency,
            is_refundable=sabre_rate.refundable,
        )

Configuration​

# Environment variables
SABRE_API_URL=https://api.sabre.com
SABRE_CLIENT_ID=your-client-id
SABRE_CLIENT_SECRET=your-client-secret
SABRE_ENVIRONMENT=production  # or "certification" for testing

Error Handling​

class SabreClient:
    async def _make_request(self, method: str, url: str, **kwargs) -> dict:
        try:
            response = await self.http_client.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 401:
                # Token expired, refresh and retry
                await self.token_manager.refresh()
                return await self._make_request(method, url, **kwargs)
            elif e.response.status_code == 429:
                raise SabreRateLimitError("Rate limit exceeded")
            else:
                raise SabreAPIError(f"Sabre API error: {e.response.status_code}")

Testing​

For development and testing, use the Sabre certification environment:

# Use certification environment for testing
SABRE_ENVIRONMENT=certification
SABRE_API_URL=https://api-crt.cert.sabre.com