2020 is a challenging year. First, we have social unrest in Hong Kong since June 2019. Then we have COVID-19 pandemic around the world. Some people lost their beloved ones, some people lost their jobs, some people had to postpone their dreams. During that very season, nobody is safe. I canceled two of my trips and got some financial loss as well. Frustrated? Yes. But what we can do currently are, stay positive and keep going. Last time, we created a Cheap Flights Checker console program with Skyscanner API. I would like to build a web app using the same API. Then, we need an Airport Seeker with Skyscanner API.
The Cheap Flights Web App
Yes, we are talking about the Cheap Flights Web App. So why do we need an Airport Seeker? Since we are making a Cheap Flights Web App, the UI picture should look like:
Where we have <Market>, <Currency>, <From> and <To> select options.
We can get <Market> and <Currency> listings from Skyscanner APIs. But there is no such thing for getting airport listings on <From> and <To> options.
In order to make the Cheap Flights Web App, we have to solve the airport listing problem. That is why we need the Airport Seeker.
Airport Seeker using Skyscanner API
First, let us check what Skyscanner API can help with the airport list. Well, they have a “Places API” which accepts location as input (e.g. “Stockholm”) and returns places related to the input. (we can see the output sample below)
We can use country name as our input, then use the returned PlaceId as IATA code and PlaceName as the airport name to make our airport list.
Limitation on Skyscanner API
Last time, we mentioned a limitation of the Skyscanner API — we can only make 50 requests within 1 minute. There are around 190 countries in the world. That means we need to run around 190 requests to build the airport list.
By not exceeding the “1 minute with 50 requests” rule, then we can record every API request starting time. If 50 requests have been run within 1 minute, we let the program sleep for certain seconds after the 50th or the number multiplied by 50th request. Assume we can handle 50 requests per minute, we will need 4 minutes to make our airport list.
It is Coding Time
Input? Check. Output? Check. Limitation? Check. So what time is it? IT IS CODING TIME!
And the good thing is, we can use most of the code from our last Cheap Flights Checker. So we install requests and TinyDB Python modules again, then create “Profile”, “Countries” and “Airports” tables.
import requests
import os, sys, json, time
from tinydb import TinyDB, Query
db = TinyDB('skyscanner_iata.json')
Profile = db.table('Profile')
Countries = db.table('Countries')
Airports = db.table('Airports')
ENDPOINT_PREFIX = "https://skyscanner-skyscanner-flight-search-v1.p.rapidapi.com/apiservices/"
MARKET = "US"
CURRENCY = "USD"
COUNTRY_ENDPOINT = "http://country.io/names.json"
SLEEP_BUFFER = 5
“Profile” is the place where we store the API key (see the previous post if you forgot how had we handled it), “Countries” is a list of countries while “Airports” is the table we wanted.
Okay, another question, where do we get the list of countries? We can’t find it from Skyscanner API, but we can get it from other web sites. For our case, we use country.io ‘s JSON response.
Before we go further, let’s create our common functions, which are almost the same as the ones used in our previous exercise.
def initProfileDB():
if "SKYSCAN_RAPID_API_KEY" in os.environ:
API_KEY = os.environ['SKYSCAN_RAPID_API_KEY']
Profile.upsert({'api_key':API_KEY}, Query().api_key.exists())
else:
API_KEY = Profile.search(Query().api_key)
if API_KEY == []:
sys.exit("No API key found")
API_KEY = API_KEY[0]['api_key']
profile_dict = {
"API_KEY": API_KEY,
}
return profile_dict
def handleAPIException(responseText, apiname):
print(json.dumps(json.loads(responseText), indent=3, sort_keys=True))
sys.exit(f"API exception on [{apiname}]")
Now it is the time we add something new. First, the country list.
def getCountries():
country_list = []
ss_countries = Countries.search(Query())
if len(ss_countries) == 0:
response = requests.request("GET", COUNTRY_ENDPOINT)
if response.status_code != 200: handleAPIException(response.text, "getCountries")
country_json = json.loads(response.text)
Countries.insert(country_json)
ss_countries = [country_json]
for country in ss_countries[0]:
country_list.append(ss_countries[0][country])
return country_list
Airport, Airport and Airport
Then we add our core airport finding logic here.
def getIataCode(place_string, request_time_list):
url = ENDPOINT_PREFIX+f"autosuggest/v1.0/{MARKET}/{CURRENCY}/en-US/"
querystring = {"query":place_string}
request_start_time = time.time()
request_time_list.append(request_start_time)
if (len(request_time_list) % 50 == 0):
the_first_request_time = request_time_list[0]
second_from_1st_request = round(time.time() - the_first_request_time)
print(f"Hit the 50th request mark, wait for 60-{second_from_1st_request} seconds")
time.sleep(60-second_from_1st_request+SLEEP_BUFFER)
request_time_list =[] #clean up the request start time list
response = requests.request("GET", url, headers=headers, params=querystring)
if response.status_code != 200: handleAPIException(response.text, "getIataCode")
place_json = json.loads(response.text)
for place in place_json["Places"]:
if ((len(place['PlaceId']) == 7) and (place['CountryName']==place_string)):
print(f"{place['PlaceName']}, {place['CountryName']} - {place['PlaceId'][:3]}")
place_dict = {
"PlaceName": place['PlaceName'],
"CountryName": place['CountryName'],
"Iata": place['PlaceId'][:3],
}
Airports.upsert(place_dict, Query().Iata == place['PlaceId'][:3])
return request_time_list
In order to handle the limitation of 50 requests per minute, we create a list called request_time_list. We record the starting time of every request. When we hit the 50th request, we check rather it passes the 1 minute mask. If everything is fine, let it process more requests. Otherwise, just let it wait for the remaining seconds within the 1 minute mask. Please note that there are (optional) 5 seconds added there, to ensure it won’t break the limit.
The rest of the codes are straight forward, they are mostly JSON handling.
print("Start processing your request...")
profile_dict = initProfileDB()
headers = {
'x-rapidapi-host': "skyscanner-skyscanner-flight-search-v1.p.rapidapi.com",
'x-rapidapi-key': profile_dict["API_KEY"]
}
airports = Airports.search(Query().Iata)
if airports == []:
print("No Airport found, start requesting from Skyscanner API...")
country_list = getCountries()
request_time_list = []
for country in country_list:
request_time_list = getIataCode(country, request_time_list)
print("Got Airport from DB")
print(Airports.search(Query()))
To run it, all we need to do are:
>export SKYSCAN_RAPID_API_KEY=abcde (for Linux/MacOS)
or
>$Env:SKYSCAN_RAPID_API_KEY="abcde" (for Win10 powershell, double quotes are needed)
or
>SET SKYSCAN_RAPID_API_KEY=abcde (for Win10 command prompt)
Then run the program
>python getairportcode.pyThe program should then capture airport information from Skyscanner, store it in our TinyDB. The result set should be used in our next project — the Cheap Flights Web App.
What have we learned in this post?
- Use time and sleep to limit the number of our API request
- Think positive and keep preparing!
(the complete source can be found at GitHub: https://github.com/codeastar/easy_iata_airport_seeker )