Create nasa_neo_data_fetcher.py

tools/nasa_neo_data_fetcher.py

@@ -0,0 +1,68 @@
+import os
+import requests
+import json
+from calendar import monthrange
+
+class NASANeoDataFetcher:
+    def __init__(self):
+        self.api_key = os.getenv("NASA_API_KEY")
+        self.root_url = "https://api.nasa.gov/neo/rest/v1/feed?"
+
+    def get_nasa_neo_data(self, start_date: str, end_date: str) -> dict:
+        """A function to get Near Earth Object data from NASA API. Try this function first before
+        calling the fetch_neo_data_in_chunks function.
+        Args:
+            start_date: A string representing the start date of the data to be fetched.
+            end_date: A string representing the end date of the data to be fetched.
+        Returns: The data fetched from the API as a JSON-like dictionary.
+        """
+        params = f"start_date={start_date}&end_date={end_date}&api_key={self.api_key}"
+        url = self.root_url + params
+        response = requests.get(url)
+        if repr(response.status_code) == "200":
+            data = response.json()
+            return data
+        else:
+            print("Error: ", response.status_code)
+
+    def split_month_into_chunks(self, year: int, month: int) -> list:
+        days_in_month = monthrange(year, month)[1]
+        first_day = 1
+        last_day = 7
+        week_dates = []
+        while last_day <= days_in_month:
+            start_date = f"{year}-{month}-{first_day}"
+            end_date = f"{year}-{month}-{last_day}"
+            week_dates.append((start_date, end_date))
+            first_day += 7
+            last_day += 7
+        remaining_days = days_in_month - first_day + 1
+        if remaining_days > 0:
+            start_date = f"{year}-{month}-{first_day}"
+            end_date = f"{year}-{month}-{days_in_month}"
+            week_dates.append((start_date, end_date))
+        return week_dates
+
+    def fetch_neo_data_in_chunks(
+        self, start_year: int, end_year: int
+    ) -> list[tuple[str, list[dict]]]:
+        """A function to fetch Near Earth Object data from NASA API in chunks. Call this function if the API
+        returns an error when you try to pull data e.g. a 400 or 429.
+        Args:
+            start_year: An integer representing the start year of the data to be fetched.
+            end_year: An integer representing the end year of the data to be fetched.
+        Returns: A list of Near Earth Object data fetched from the API as a JSON-like dictionary.
+        """
+        neo_data = []
+        for year in range(start_year, end_year + 1):
+            for month in range(1, 13):
+                for chunk in self.split_month_into_chunks(year, month):
+                    start_date, end_date = chunk
+                    print(f"Fetching data for {start_date} to {end_date}")
+                    data = self.get_nasa_neo_data(
+                        start_date=start_date, end_date=end_date
+                    )
+                    if data and "near_earth_objects" in data:
+                        neo_data.extend(data["near_earth_objects"].items())
+        neo_data_json = json.dumps(neo_data)
+        return neo_data_json