[main] Prod Creds
All checks were successful
Build, Push, and Deploy to Nomad / docker-nomad (push) Successful in 1m10s

This commit is contained in:
Henrik Jess
2025-01-14 19:51:29 +01:00
parent 5de92fa7a1
commit e1207362de
4 changed files with 630 additions and 18 deletions

View File

@@ -0,0 +1,95 @@
import json
from datetime import datetime
from amadeus import Client, ResponseError
from pprint import pprint
# OopCompanion:suppressRename
class AmadeusClient:
def __init__(self, client_id: str, client_secret: str):
"""
Initialize the Amadeus API client.
"""
self.client = Client(
client_id = client_id,
client_secret = client_secret
)
def find_cheapest_flights(self, origin: str, destination: str, departure_date: str, adults: int = 1):
"""
Find the cheapest flights from the origin to the destination on a specific date.
:param origin: The IATA code for the origin location.
:param destination: The IATA code for the destination location.
:param departure_date: The date of departure in YYYY-MM-DD format.
:param adults: Number of adult passengers.
:return: Flight offers data.
"""
try:
response = self.client.shopping.flight_offers_search.get(
originLocationCode = origin,
destinationLocationCode = destination,
departureDate = departure_date,
adults = adults,
max=250,
nonStop="true"
)
return response.data
except ResponseError as error:
raise error
# Example usage
if __name__ == "__main__":
client_id = 'uxDqIh36xPAUvpXnXynwAnH86pGBdIch'
client_secret = 'xTSLooNZpJWantb5'
amadeus_client = AmadeusClient( client_id, client_secret )
try:
#flights = amadeus_client.find_cheapest_flights(
# origin = 'LIS',
# destination = 'CPH',
# departure_date = '2025-01-15',
# adults = 1,
#)
#pprint( flights )
#with open( "flights.json", "w" ) as file:
# json.dump( flights, file, indent = 4 )
with open( "flights.json", "r" ) as file:
flights = json.load( file )
for flight_offer in flights:
itineraries = flight_offer.get( 'itineraries', [] )
result = []
price = flight_offer.get( 'price', { } ).get( 'total', 0 )
for itinerary in itineraries:
segments = itinerary.get( 'segments', [] )
for segment in segments:
numberOfStops = segment.get( 'numberOfStops', -1 )
if numberOfStops == 0:
departure = segment.get( 'departure', { } )
arrival = segment.get( 'arrival', { } )
travel_time = datetime.fromisoformat( str(arrival['at']) ) - datetime.fromisoformat( str(departure['at']) )
result.append( {
"departure": {
"iataCode": departure.get( "iataCode" ),
"at": departure.get( "at" ),
},
"arrival": {
"iataCode": arrival.get( "iataCode" ),
"at": arrival.get( "at" ),
}
,
"travel_time": {
"time":travel_time,
}
} )
pprint( result )
except ResponseError as error:
print( f"An error occurred: {error}" )