flight_search_token.py

import os
import requests
from dotenv import load_dotenv
from datetime import datetime, timedelta
from sheety_data_manager import SheetyDataManager

# Load environment variables from the .env file
load_dotenv()

# Define API URL to get the access token
url_token = "https://test.api.amadeus.com/v1/security/oauth2/token"

# Get today's date and calculate the date 6 months later
today = datetime.now()
six_months_later = today + timedelta(days=6 * 30)
departure_date = today.strftime("%Y-%m-%d")
six_month = six_months_later.strftime("%Y-%m-%d")


class GetToken:
    """
    Class to get API token from Amadeus using client credentials.
    """

    def __init__(self):
        """
        Initializes the necessary credentials for requesting the token.
        """
        self.request = None
        self.api_key = os.environ["API_KEY"]  # Get API Key from environment
        self.api_secret = os.environ["API_SECRET"]  # Get API Secret from environment
        self.header = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        self.data = {
            "grant_type": "client_credentials",
            "client_id": self.api_key,
            "client_secret": self.api_secret
        }

    def returned_token(self):
        """
        Send POST request to get the token and return the access token.
        :return: Access token as a string
        """
        self.request = requests.post(url=url_token, data=self.data, headers=self.header)
        return self.request.json().get("access_token", None)


# Example usage of GetToken class to fetch the access token
token_instance = GetToken()
print(token_instance.returned_token())  # Print the token for testing

# Define the flight data API endpoint
url_data_flight = "https://test.api.amadeus.com/v2/shopping/flight-offers"


class GetDataFlight:
    """
    Class to fetch flight data using the Amadeus API.
    """

    def __init__(self, token_id: GetToken, data: SheetyDataManager):
        """
        Initialize the class with token and data manager for retrieving flight data.
        :param token_id: Instance of GetToken class for fetching token
        :param data: Instance of SheetyDataManager to manage sheet data
        """
        self.request = {}
        self.token_id = token_id.returned_token()  # Get the token using GetToken instance
        self.data = data.get_sheet_data()  # Fetch data from the Sheety manager

        # Set authorization header for API requests
        self.header = {
            "Authorization": f"Bearer {self.token_id}"
        }

    def check_flights(self, from_city, to_city, tomorrow, after_six_month):
        """
        Search for flights between two cities with given dates using Amadeus API.
        :param from_city: IATA code for the departure city
        :param to_city: IATA code for the destination city
        :param tomorrow: Departure date in 'YYYY-MM-DD' format
        :param after_six_month: Return date in 'YYYY-MM-DD' format
        :return: Flight offers in JSON format or None if there is an error
        """
        parameters = {
            "originLocationCode": from_city,  # IATA code for departure city
            "destinationLocationCode": to_city,  # IATA code for destination city
            "departureDate": tomorrow,  # Departure date
            "returnDate": after_six_month,  # Return date
            "nonStop": "true",  # Only nonstop flights
            "currencyCode": "GBP",  # Currency for price
            "adults": 1,  # Number of adults
            "max": 10  # Maximum number of flight offers to return
        }

        # Make a GET request to the flight search API
        response = requests.get(url=url_data_flight, headers=self.header, params=parameters)

        if response.status_code != 200:
            # Print error message if the API request fails
            print(f"Error in check_flights(): {response.status_code}")
            print("There was a problem with the flight search.\n"
                  "For details on status codes, check the API documentation:\n"
                  "https://developers.amadeus.com/self-service/category/flights/api-doc/flight-offers-search/api"
                  "-reference")
            print("Response body:", response.text)
            return None

        # Return the response in JSON format
        return response.json()


# Example usage of the GetDataFlight class
if __name__ == "__main__":
    # Initialize the necessary objects
    token_instance = GetToken()  # Get the token from GetToken
    sheet_manager = SheetyDataManager()  # Sheety data manager for sheet data

    # Create an instance of GetDataFlight
    flight_data_instance = GetDataFlight(token_id=token_instance, data=sheet_manager)

    # Call the check_flights method to search for flights
    flight_results = flight_data_instance.check_flights(
        from_city="LON",  # Example IATA code for London
        to_city="PAR",  # Example IATA code for Paris
        tomorrow=departure_date,  # Use today's date as departure
        after_six_month=six_month  # Use the calculated date for 6 months later as return date
    )

    # Print the flight results (if any)
    print(flight_results)