Spaces:
Sleeping
Sleeping
| import time | |
| import requests | |
| from typing import Dict | |
| from ..config import AMADEUS_BASE, get_amadeus_credentials | |
| class AmadeusClient: | |
| def __init__(self): | |
| self.client_id, self.client_secret = get_amadeus_credentials() | |
| self._token = None | |
| self._exp = 0 | |
| def _auth(self): | |
| if self._token and time.time() < self._exp - 10: | |
| return | |
| url = f"{AMADEUS_BASE}/v1/security/oauth2/token" | |
| data = { | |
| "grant_type": "client_credentials", | |
| "client_id": self.client_id, | |
| "client_secret": self.client_secret, | |
| } | |
| headers = {"Content-Type": "application/x-www-form-urlencoded"} | |
| r = requests.post(url, data=data, headers=headers, timeout=20) | |
| r.raise_for_status() | |
| j = r.json() | |
| self._token = j["access_token"] | |
| self._exp = time.time() + int(j.get("expires_in", 1799)) | |
| def _get(self, path: str, params: Dict): | |
| self._auth() | |
| url = f"{AMADEUS_BASE}{path}" | |
| headers = {"Authorization": f"Bearer {self._token}"} | |
| r = requests.get(url, params=params, headers=headers, timeout=30) | |
| if r.status_code == 401: | |
| # retry once | |
| self._token = None | |
| self._auth() | |
| headers = {"Authorization": f"Bearer {self._token}"} | |
| r = requests.get(url, params=params, headers=headers, timeout=30) | |
| r.raise_for_status() | |
| return r.json() | |
| def locations(self, keyword: str, subtypes="CITY,AIRPORT", page_limit=10): | |
| return self._get( | |
| "/v1/reference-data/locations", | |
| { | |
| "keyword": keyword, | |
| "subType": subtypes, | |
| "page[limit]": page_limit, | |
| }, | |
| ) | |
| def flight_offers( | |
| self, | |
| origin: str, | |
| dest: str, | |
| depart: str, | |
| ret: str, | |
| adults=1, | |
| currency="PKR", | |
| non_stop=False, | |
| max_results=20, | |
| ): | |
| params = { | |
| "originLocationCode": origin, | |
| "destinationLocationCode": dest, | |
| "departureDate": depart, | |
| "adults": adults, | |
| "currencyCode": currency, | |
| "max": max_results, | |
| } | |
| if ret: | |
| params["returnDate"] = ret | |
| if non_stop: | |
| params["nonStop"] = "true" | |
| return self._get("/v2/shopping/flight-offers", params) | |