Abstract API/Abstract_API_Get_IP_Geolocation.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9e7cc5aa-233b-439d-aaae-7d47490db8c3",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "<img width=\"8%\" alt=\"Abstract API.png\" src=\"https://raw.githubusercontent.com/jupyter-naas/awesome-notebooks/master/.github/assets/logos/Abstract%20API.png\" style=\"border-radius: 15%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2d3a8f73-d1f0-45dd-9aaf-757b578d6af3",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "# Abstract API - Get IP Geolocation\n",
    "<a href=\"https://bit.ly/3JyWIk6\">Give Feedback</a> | <a href=\"https://github.com/jupyter-naas/awesome-notebooks/issues/new?assignees=&labels=bug&template=bug_report.md&title=Abstract+API+-+Get+IP+Geolocation:+Error+short+description\">Bug report</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9b95e2ba-a05b-4b71-a430-c890a5b54fba",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "**Tags:** #api #abstract-api #ip #geolocation #stream #multithread #queues #operations #dataprocessing #automation"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "14ffd274-1141-467f-95a0-af19740f7ed6",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "**Author:** [Maxime Jublou](https://www.linkedin.com/in/maximejublou)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f88f9d2-0e31-41e4-a368-fecacb787ce0",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "**Last update:** 2023-04-12 (Created: 2023-02-24)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "naas-description",
   "metadata": {
    "papermill": {},
    "tags": [
     "description"
    ]
   },
   "source": [
    "**Description:** This notebook provides a way to get the geolocation of an IP address using the AbstractAPI service."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3e99d497-5b3e-40f8-b54d-107f9339062d",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "## Input"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0e88c228-22d1-4a09-9f20-74e1f2b14bb7",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b1354e4c-eeff-43c1-a38a-d17f5c1f8101",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "import threading\n",
    "import queue\n",
    "import time\n",
    "import requests\n",
    "import pandas as pd\n",
    "import json\n",
    "try:\n",
    "    from ratelimiter import RateLimiter\n",
    "except:\n",
    "    ! pip install --user ratelimiter\n",
    "    from ratelimiter import RateLimiter"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "148e85eb-514e-4874-a4b1-4a8c8c9db5bc",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Setup Variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be642235-c61a-40f7-ba4f-d28fd9ba05ed",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "API_KEY = \"...\"\n",
    "CALL_PER_SECOND = 500\n",
    "\n",
    "# Define number of workers.\n",
    "# You will learn by running this notebook the right numbers for your machine.\n",
    "MAX_WORKERS = 50\n",
    "MIN_WORKERS = 45\n",
    "\n",
    "# File where the results from the api will be streamed.\n",
    "RESULT_TMP_FILE = \"/tmp/results.txt\"\n",
    "\n",
    "# File where we will store the resulting CSV file.\n",
    "CSV_FILE = \"results.csv\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d871136a-dde3-4266-b61a-71efa6c92168",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-08-26T09:32:30.286092Z",
     "iopub.status.busy": "2022-08-26T09:32:30.285779Z",
     "iopub.status.idle": "2022-08-26T09:32:30.289343Z",
     "shell.execute_reply": "2022-08-26T09:32:30.288607Z",
     "shell.execute_reply.started": "2022-08-26T09:32:30.286059Z"
    },
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Create list of IP addresses\n",
    "\n",
    "You can replace this code to load your own dataframe.\n",
    "Only requirements are:\n",
    "- The dataframe variable must be named `df`\n",
    "- The dataframe should have an `ip` columns containing ip addresses."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c0050892-14bc-4809-a53a-1416c271c2bb",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Generate a pandas dataframe containing ip addresses\n",
    "ip_addresses = []\n",
    "\n",
    "for x in range(1, 41):\n",
    "    for y in range(1, 251):\n",
    "        ip_addresses.append({\"ip\": f\"77.205.{x}.{y}\"})\n",
    "\n",
    "df = pd.DataFrame(ip_addresses)\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf99a140-4a9b-4e97-bf7c-3fb11cba834c",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-08-26T09:06:52.577011Z",
     "iopub.status.busy": "2022-08-26T09:06:52.576738Z",
     "iopub.status.idle": "2022-08-26T09:06:52.580907Z",
     "shell.execute_reply": "2022-08-26T09:06:52.579938Z",
     "shell.execute_reply.started": "2022-08-26T09:06:52.576976Z"
    },
    "papermill": {},
    "tags": []
   },
   "source": [
    "## Model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0cf82cda-82e2-4272-8eb8-341e091f06c7",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Create queues\n",
    "The goal is to stream events (here IP addresses), instead of batching all of them in one process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "93c90d74-d688-4e1f-ae2b-e2838a1ee66c",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "# This queue will contain the list of IP addresses.\n",
    "job_queue = queue.Queue()\n",
    "\n",
    "# This queue will contain the results from the workers.\n",
    "results_queue = queue.Queue()\n",
    "\n",
    "# This queue will be used to count the number of redrived jobs. (Jobs on error that are pushed back on job_queue for new processing)\n",
    "redrive_queue = queue.Queue()\n",
    "\n",
    "# This queue will contain ended jobs (Either successful or pushed in dead_letter_queue)\n",
    "ended_queue = queue.Queue()\n",
    "\n",
    "# This queue will be used to send failing job to it, that way we will be able to look at it to see if we have any errors.\n",
    "dead_letter_queue = queue.Queue()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ac5c925-5876-4e3f-b5e0-d7860b876c27",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Set timer\n",
    "\n",
    "This is the cadence at which we will call the Abstract API."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9cd53327-878a-4293-a4f6-cc5679572d9a",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "timer = (1000 / CALL_PER_SECOND) / 1000\n",
    "timer"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eca4e61e-903e-4b4a-9f8f-d7551357dedf",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Create job worker function\n",
    "\n",
    "This function will be the worker taking jobs and calling the api."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f176b89b-3928-4013-b56b-61d0346c7c8b",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "@RateLimiter(max_calls=1, period=timer)\n",
    "def limit_me():\n",
    "    pass\n",
    "\n",
    "\n",
    "# We pre compute the url to only append the ip in the worker to not \"format\" the string everytime, doing this is 10 time faster.\n",
    "pre_computed_url = (\n",
    "    \"https://ipgeolocation.abstractapi.com/v1/?api_key={API_KEY}&ip_address=\".format(\n",
    "        API_KEY=API_KEY\n",
    "    )\n",
    ")\n",
    "\n",
    "\n",
    "def job_worker():\n",
    "    # Initialize requests session.\n",
    "    s = requests.Session()\n",
    "\n",
    "    while True:\n",
    "        # Wait for a job to be availabe.\n",
    "        item = job_queue.get()\n",
    "\n",
    "        # Limit\n",
    "        limit_me()\n",
    "\n",
    "        try:\n",
    "            # Query the api with our existing session.\n",
    "            response = s.get(pre_computed_url + item)\n",
    "\n",
    "            # Raise for status if we have an error.\n",
    "            response.raise_for_status()\n",
    "\n",
    "            # Send the result to the results_queue.\n",
    "            results_queue.put(response.text)\n",
    "        except Exception as e:\n",
    "            if response.status_code == 429:\n",
    "                job_queue.put(item)\n",
    "                redrive_queue.put(1)\n",
    "            else:\n",
    "                print(e)\n",
    "                dead_letter_queue.put({\"item\": item, \"error\": e})\n",
    "                ended_queue.put(1)\n",
    "        job_queue.task_done()\n",
    "\n",
    "\n",
    "workers = []\n",
    "\n",
    "\n",
    "def start_a_new_worker():\n",
    "    print(\"🚀 Starting a new worker\")\n",
    "    workers.append(threading.Thread(target=job_worker, daemon=True).start())\n",
    "\n",
    "\n",
    "# Start minimal workers count.\n",
    "for i in range(0, MIN_WORKERS):\n",
    "    start_a_new_worker()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "856c4548-bcad-4b97-ade8-9ec6ca356f80",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Create result worker function\n",
    "\n",
    "This function is the code that will be run by our result worker function.\n",
    "Its role is to take the results from all workers and store them.\n",
    "\n",
    "**Note**: If we want to get a lot of jobs done then we will need to offload results to a file instead of keeping everything in memory. We could stream json representations to a file then read it later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7848951a-a515-4c4c-892f-14d417bd6fe8",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "def result_worker():\n",
    "    f = open(RESULT_TMP_FILE, \"w\", encoding=\"utf-8\")\n",
    "    previous_time = None\n",
    "    while True:\n",
    "        try:\n",
    "            result = results_queue.get(block=True, timeout=10)\n",
    "            f.write(result + \"\\n\")\n",
    "            results_queue.task_done()\n",
    "            ended_queue.put(1)\n",
    "        # We should except _queue.empty here instead of catching a generic exception.\n",
    "        # It's fine for now.2\n",
    "        except Exception as e:\n",
    "            print(\"Leaving result_worker\")\n",
    "            break\n",
    "    f.close()\n",
    "\n",
    "\n",
    "# Start the result worker\n",
    "threading.Thread(target=result_worker, daemon=True).start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc819cae-0b61-4028-8f86-121e649f48c9",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Create queue monitor function\n",
    "\n",
    "This function is here to provide live progress during the execution.\n",
    "It can also start new workers if we are not reaching the `CALL_PER_SECOND` target."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "efb482ca-ad22-401d-bd39-57f518954b67",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "@RateLimiter(max_calls=1, period=1)\n",
    "def monitor_limiter():\n",
    "    pass\n",
    "\n",
    "\n",
    "def queue_monitor():\n",
    "    prev_queue_size = 0\n",
    "    while True:\n",
    "        ended_queue_size = ended_queue.qsize()\n",
    "        jobs_per_seq = ended_queue_size - prev_queue_size\n",
    "\n",
    "        workers_number = len(workers)\n",
    "        if jobs_per_seq < CALL_PER_SECOND and workers_number < MAX_WORKERS:\n",
    "            start_a_new_worker()\n",
    "\n",
    "        print(\n",
    "            f\"#jobqsize[{job_queue.qsize()}] #resultsqsize[{results_queue.qsize()}] #redriveqsize[{redrive_queue.qsize()}] #deadletterqsize[{dead_letter_queue.qsize()}] | ✅ {ended_queue.qsize()}/{nbr_of_jobs} jobs completed. ⚡ {jobs_per_seq}/sec. 👷‍♂️ Running workers: {workers_number}\"\n",
    "        )\n",
    "\n",
    "        prev_queue_size = ended_queue_size\n",
    "\n",
    "        if ended_queue_size >= nbr_of_jobs:\n",
    "            print(\"Jobs completed. Shuting down monitor.\")\n",
    "            return\n",
    "\n",
    "        monitor_limiter()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "daa7f9cc-0614-4bf6-ba8e-26dbca47bca0",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Run API calls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b575e2d5-e765-4784-b22e-232ea9e1699c",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Pushing all jobs in the queue.\n",
    "nbr_of_jobs = 0\n",
    "for i in df.ip:\n",
    "    job_queue.put(i)\n",
    "    nbr_of_jobs += 1\n",
    "\n",
    "# Start monitor function\n",
    "threading.Thread(target=queue_monitor, daemon=True).start()\n",
    "\n",
    "# Block until all tasks are done.\n",
    "job_queue.join()\n",
    "\n",
    "# Making sure we have handled all the jobs.\n",
    "while True:\n",
    "    if nbr_of_jobs == ended_queue.qsize():\n",
    "        break\n",
    "    print(nbr_of_jobs, ended_queue.qsize())\n",
    "    time.sleep(0.1)\n",
    "\n",
    "print(\"🎉 All API calls has been completed!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dad715c7-31b8-442f-81a5-fa4e94bce215",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2022-08-26T09:06:54.938567Z",
     "iopub.status.busy": "2022-08-26T09:06:54.938297Z",
     "iopub.status.idle": "2022-08-26T09:06:54.942393Z",
     "shell.execute_reply": "2022-08-26T09:06:54.941317Z",
     "shell.execute_reply.started": "2022-08-26T09:06:54.938514Z"
    },
    "papermill": {},
    "tags": []
   },
   "source": [
    "## Output\n",
    "\n",
    "Load all results from the `RESULT_TMP_FILE` and load them in a Pandas DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "308ae4b9-d179-4b80-ab76-c88ff93f23b5",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "results = []\n",
    "\n",
    "# Open file\n",
    "with open(RESULT_TMP_FILE, \"r\") as results_file:\n",
    "    # Read line by line\n",
    "    for line in results_file:\n",
    "        # Load json into results list\n",
    "        results.append(json.loads(line))\n",
    "\n",
    "# Create DataFrame from results\n",
    "results_df = pd.DataFrame(results)\n",
    "\n",
    "# Display DataFrame\n",
    "results_df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1f9b2d9-da61-42fc-a01b-b9854532e176",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "source": [
    "### Store the results as a CSV file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "801a19e5-f646-4a84-b151-37312b7a39e8",
   "metadata": {
    "papermill": {},
    "tags": []
   },
   "outputs": [],
   "source": [
    "results_df.to_csv(CSV_FILE)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.9.6"
  },
  "naas": {
   "notebook_id": "8d77689e05cc016ea0eed7a995b0ef55adc9998684802cc44b3cccf702baed41",
   "notebook_path": "Abstract API/Abstract_API_Get_IP_Geolocation.ipynb"
  },
  "papermill": {
   "default_parameters": {},
   "environment_variables": {},
   "parameters": {},
   "version": "2.4.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}