Upload 2 files

planmate/clients/amadeus_client.py

@@ -0,0 +1,76 @@
+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)