04_update_RDS_database.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### WBS Coding School\n",
    "___\n",
    "# Data Engineering Project\n",
    "\n",
    "This is a data engineering project, in which I use Python, MySQL and AWS Services to create and automatically update an online database.\n",
    "\n",
    "It is a learning project, in which I practise several data engineering techniques, such as API calls and AWS Lambda functions.\n",
    "\n",
    "Our tasks are:\n",
    "- [x] Collect data\n",
    "- [x] Clean data\n",
    "- [x] Create a database\n",
    "- [x] Update the database with the latest data\n",
    "- [ ] Move the data pipeline to the Cloud (AWS)\n",
    "\n",
    "\n",
    "#### Next Steps\n",
    "The next task is to transfer the updating of the dynamic tables to the Cloud and automate it.\n",
    "Therefore, I pasted the code from this script to an AWS Lambda function. Subsequently, I set up an AWS EventBridge Schedule that triggers my Lambda function on a daily basis.\n",
    "\n",
    "Et voilà, there you have an automated Data Pipeline running in the Cloud.\n",
    "\n",
    "___"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Update the Cloud database\n",
    "This script fetches the latest data for the dynamic tables (`weather`, `arrivals`) and pushes them to the Cloud database."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Table of contents:\n",
    "- [Connect to the Cloud MySQL instance](#connect)\n",
    "- [Update `weather` table](#update_weather)\n",
    "- [Update `arrivals` table](#update_arrivals)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import Libraries & Dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import requests\n",
    "from datetime import datetime, timedelta, date\n",
    "import config_file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"connect\"></a>\n",
    "#### Connect to the Cloud MySQL instance\n",
    "(set up a MySQL database on an AWS RDS instance beforehand)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Cloud MySQL server connection information\n",
    "user = config_file.AWS_DATABASE_USER\n",
    "password = config_file.AWS_DATABASE_PASSWORD\n",
    "host = config_file.AWS_DATABASE_HOST\n",
    "port = config_file.AWS_DATABASE_PORT\n",
    "schema = config_file.AWS_DATABASE_SCHEMA\n",
    "\n",
    "# Connect to RDS MySQL server\n",
    "connection = f'mysql+pymysql://{user}:{password}@{host}:{port}/{schema}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"update_weather\"></a>\n",
    "## Update `weather` table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_weather_response(city: str, cnt = 50):\n",
    "    base_url = \"https://api.openweathermap.org/data/2.5/forecast?\"\n",
    "    API_key = config_file.OPEN_WEATHER_API_KEY\n",
    "    units = \"metric\"\n",
    "    language = \"en\"\n",
    "    weather_response = requests.get(f\"{base_url}appid={API_key}&q={city}&cnt={cnt}&units={units}&lang={language}\")\n",
    "    return weather_response\n",
    "\n",
    "\n",
    "def create_weather_dataframe(response: requests.models.Response):\n",
    "    # Save the weather data:\n",
    "    weather_json = response.json()\n",
    "    weather_df = pd.json_normalize(weather_json[\"list\"]) # Column \"weather\" contains a dictionary ...\n",
    "    weather_df[\"city\"] = weather_json[\"city\"][\"name\"]\n",
    "    weather_exploded_df = pd.json_normalize(weather_df.explode(\"weather\")[\"weather\"]) # ... we therefore explode that column ...\n",
    "    weather_exploded_df[\"city\"] = weather_json[\"city\"][\"name\"]\n",
    "    weather_exploded_df[\"dt\"] = weather_df[\"dt\"]\n",
    "    weather_all_df = weather_exploded_df.merge(weather_df, on=[\"city\", \"dt\"]) # ... and re-merge the two dataframes.\n",
    "    weather_clean_df = weather_all_df[[\"city\", \"dt_txt\", \"main\", \"description\", \"main.temp\", \"main.feels_like\", \"wind.speed\"]]\n",
    "    weather_clean_df = weather_clean_df.rename(columns={\n",
    "        \"dt_txt\": \"forecast_time\",\n",
    "        \"main\": \"outlook\",\n",
    "        \"description\": \"outlook_description\",\n",
    "        \"main.temp\": \"temperature\",\n",
    "        \"main.feels_like\": \"feels_like\",\n",
    "        \"wind.speed\": \"wind_speed\"\n",
    "    })\n",
    "    # Save city data:\n",
    "    city = weather_json['city']['name']\n",
    "    latitude = weather_json['city']['coord']['lat']\n",
    "    longitude = weather_json['city']['coord']['lon']\n",
    "    country = weather_json['city']['country']\n",
    "    city_data = {\n",
    "        \"city\": city,\n",
    "        \"latitude\": latitude,\n",
    "        \"longitude\": longitude,\n",
    "        \"country\": country\n",
    "    }\n",
    "    city_data_df = pd.DataFrame(city_data, index=[0])\n",
    "    # Join the two dataframes:\n",
    "    weather_city_df = pd.merge(weather_clean_df, city_data_df, on=\"city\", how=\"inner\")\n",
    "    return weather_city_df\n",
    "\n",
    "\n",
    "def get_weather_data(cities: list):\n",
    "    weather_cities_df = pd.DataFrame()\n",
    "    for city in cities:\n",
    "        weather_response = create_weather_response(city)\n",
    "        weather_city_df = create_weather_dataframe(weather_response)\n",
    "        weather_cities_df = pd.concat([weather_cities_df, weather_city_df])\n",
    "    return weather_cities_df.reset_index(drop=True)\n",
    "\n",
    "\n",
    "def clean_weather(weather_df: pd.DataFrame, city_ids: int):\n",
    "    weather_df_clean = weather_df.copy()\n",
    "    # Create city_id column:\n",
    "    weather_df_clean[\"city_id\"] = weather_df_clean[\"city\"].map(city_ids)\n",
    "    columns_to_keep = [\"city_id\", \"forecast_time\", \"outlook\", \"outlook_description\", \"temperature\", \"feels_like\", \"wind_speed\"]\n",
    "    weather_df_clean = weather_df_clean[columns_to_keep]\n",
    "    # Change forecast_time column type:\n",
    "    weather_df_clean['forecast_time']= pd.to_datetime(weather_df_clean['forecast_time'])\n",
    "    return weather_df_clean"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get cities and `city_ids`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "gans_cities_data = pd.read_sql_table(\"cities\", con=connection, columns=['city_id', 'city_name'])\n",
    "cities_of_interest = list(gans_cities_data[\"city_name\"])\n",
    "city_ids = dict(zip(gans_cities_data['city_name'], gans_cities_data['city_id']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get up-to-date `weather` data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "weather_df = get_weather_data(cities_of_interest)\n",
    "weather_df_clean = clean_weather(weather_df, city_ids)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Push `weather` data to Cloud database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "240"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Update weather data\n",
    "weather_df_clean.to_sql(\"weather\", if_exists=\"append\", con=connection, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Update `arrivals` table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_date_range():\n",
    "    date_range_dict = {}\n",
    "    morning1 = (datetime.combine(date.today(), datetime.min.time()) + timedelta(1)).strftime('%Y-%m-%dT%H:%M')\n",
    "    evening1 = (datetime.combine(date.today(), datetime.min.time()) + timedelta(1.5)).strftime('%Y-%m-%dT%H:%M')\n",
    "    morning2 = (datetime.combine(date.today(), datetime.min.time()) + timedelta(1.5)).strftime('%Y-%m-%dT%H:%M')\n",
    "    evening2 = (datetime.combine(date.today(), datetime.min.time()) + timedelta(2)).strftime('%Y-%m-%dT%H:%M')\n",
    "    date_range_dict[\"morning\"] = [morning1, morning2]\n",
    "    date_range_dict[\"evening\"] = [morning2, evening2]\n",
    "    return date_range_dict\n",
    "\n",
    "\n",
    "def get_arrivals_response(icao_code: str, date_range: dict, day_time: str):\n",
    "    url = f\"https://aerodatabox.p.rapidapi.com/flights/airports/icao/{icao_code}/{date_range[day_time][0]}/{date_range[day_time][1]}\"\n",
    "    querystring = {\n",
    "\t    \"withLeg\": \"false\", \"direction\": \"Arrival\", \n",
    "\t    \"withCancelled\": \"false\", \"withCodeshared\": \"false\", \n",
    "\t    \"withCargo\": \"false\", \"withPrivate\": \"false\", \"withLocation\": \"false\"\n",
    "    }\n",
    "    headers = {\n",
    "    \t\"X-RapidAPI-Key\": config_file.AERODATABOX_API_KEY,\n",
    "    \t\"X-RapidAPI-Host\": \"aerodatabox.p.rapidapi.com\"\n",
    "    }\n",
    "    response = requests.get(url, headers=headers, params=querystring)\n",
    "    return response\n",
    "\n",
    "\n",
    "def create_arrivals_df(response: requests.models.Response, icao_code: str):\n",
    "    arrivals_json = response.json()[\"arrivals\"]\n",
    "    arrivals_df = pd.json_normalize(arrivals_json)\n",
    "    arrivals_df[\"icao\"] = icao_code\n",
    "    new_columns_dict = {\n",
    "        \"icao\": \"icao\",\n",
    "        \"number\": \"flight_number\", \n",
    "        \"movement.scheduledTime.utc\": \"arrival_time_utc\", \n",
    "        \"movement.scheduledTime.local\": \"arrival_time_local\",\n",
    "        \"airline.name\": \"airline\", \n",
    "        \"movement.airport.name\": \"departing_airport\", \n",
    "        \"movement.airport.icao\": \"departing_airport_icao\"\n",
    "    }\n",
    "    old_columns = list(new_columns_dict.keys())\n",
    "    arrivals_df_clean = arrivals_df[old_columns].rename(columns=new_columns_dict)\n",
    "    return arrivals_df_clean\n",
    "\n",
    "\n",
    "def get_arrivals_data(icao_codes: list):\n",
    "    # Initiate empty list:\n",
    "    arrivals_list = []\n",
    "    # Get start & end date:\n",
    "    date_range = create_date_range()\n",
    "\n",
    "    for icao in icao_codes:\n",
    "        for day_time in ['morning', 'evening']:\n",
    "            response = get_arrivals_response(icao_code=icao, date_range=date_range, day_time=day_time)\n",
    "            if response.status_code != 200:\n",
    "                continue\n",
    "            arrivals_df = create_arrivals_df(response, icao_code=icao)\n",
    "            arrivals_list.append(arrivals_df)\n",
    "        arrivals = pd.concat(arrivals_list)\n",
    "    return arrivals.reset_index(drop=True)\n",
    "\n",
    "\n",
    "def clean_arrivals(arrivals_df: pd.DataFrame):\n",
    "    arrivals_df_clean = arrivals_df.copy().rename(columns={\n",
    "        \"icao\": \"arrival_icao\", \n",
    "        \"arrival_time_local\": \"arrival_time\", \n",
    "        \"departing_airport_icao\": \"departure_icao\"\n",
    "    })[[\"flight_number\", \"arrival_icao\", \"arrival_time\", \"departure_icao\"]]\n",
    "    arrivals_df_clean[\"arrival_time\"] = pd.to_datetime(arrivals_df_clean[\"arrival_time\"].str[:-6]) # remove the addition (e.g. +01:00)\n",
    "    return arrivals_df_clean"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get airports and `icao_codes`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'connection' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb Cell 19\u001b[0m line \u001b[0;36m1\n\u001b[0;32m----> <a href='vscode-notebook-cell:/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb#X14sZmlsZQ%3D%3D?line=0'>1</a>\u001b[0m gans_airports_data \u001b[39m=\u001b[39m pd\u001b[39m.\u001b[39mread_sql_table(\u001b[39m\"\u001b[39m\u001b[39mairports\u001b[39m\u001b[39m\"\u001b[39m, con\u001b[39m=\u001b[39mconnection, columns\u001b[39m=\u001b[39m[\u001b[39m'\u001b[39m\u001b[39mairport_icao\u001b[39m\u001b[39m'\u001b[39m])\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb#X14sZmlsZQ%3D%3D?line=1'>2</a>\u001b[0m icao_codes \u001b[39m=\u001b[39m \u001b[39mlist\u001b[39m(gans_airports_data[\u001b[39m\"\u001b[39m\u001b[39mairport_icao\u001b[39m\u001b[39m\"\u001b[39m])\n",
      "\u001b[0;31mNameError\u001b[0m: name 'connection' is not defined"
     ]
    }
   ],
   "source": [
    "gans_airports_data = pd.read_sql_table(\"airports\", con=connection, columns=['airport_icao'])\n",
    "icao_codes = list(gans_airports_data[\"airport_icao\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get up-to-date `arrivals` data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'get_arrivals_data' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb Cell 21\u001b[0m line \u001b[0;36m1\n\u001b[0;32m----> <a href='vscode-notebook-cell:/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb#X15sZmlsZQ%3D%3D?line=0'>1</a>\u001b[0m arrivals_df \u001b[39m=\u001b[39m get_arrivals_data(icao_codes)\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/moritzbaumann/Documents/Courses/Data_Science_Bootcamp/Chapter_5_Data_Pipelines_Cloud/Cloud_Data_Engineering/04_update_RDS_database.ipynb#X15sZmlsZQ%3D%3D?line=1'>2</a>\u001b[0m arrivals_df_clean \u001b[39m=\u001b[39m clean_arrivals(arrivals_df)\n",
      "\u001b[0;31mNameError\u001b[0m: name 'get_arrivals_data' is not defined"
     ]
    }
   ],
   "source": [
    "arrivals_df = get_arrivals_data(icao_codes)\n",
    "arrivals_df_clean = clean_arrivals(arrivals_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Push `arrivals` data to the Cloud database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3544"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "arrivals_df_clean.to_sql(\"arrivals\", if_exists=\"append\", con=connection, index=False)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}