FastAPI.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# FastAPI Notebook\n",
    "\n",
    "Some getting-started stuff to the awesome world of FastAPI + Pydantic!\n",
    "\n",
    "Some indications about this notebook:\n",
    "\n",
    "- Must always run the first two cells"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Basic import of FastAPI and function to start running the API with Uvicorn\n",
    "\n",
    "import uvicorn\n",
    "from fastapi import FastAPI\n",
    "\n",
    "app = FastAPI()\n",
    "\n",
    "def run():\n",
    "    uvicorn.run(app)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Code in this cell is just for (re)starting the API on a Process, and other compatibility stuff with Jupyter cells.\n",
    "# Just ignore it!\n",
    "\n",
    "from multiprocessing import Process\n",
    "from wait4it import wait_for\n",
    "\n",
    "_api_process = None\n",
    "\n",
    "def start_api():\n",
    "    \"\"\"Stop the API if running; Start the API; Wait until API (port) is available (reachable)\"\"\"\n",
    "    global _api_process\n",
    "    if _api_process:\n",
    "        _api_process.terminate()\n",
    "        _api_process.join()\n",
    "    \n",
    "    _api_process = Process(target=run, daemon=True)\n",
    "    _api_process.start()\n",
    "    wait_for(port=8000)\n",
    "\n",
    "def delete_route(method: str, path: str):\n",
    "    \"\"\"Delete the given route from the API. This must be called on cells that re-define a route\"\"\"\n",
    "    [app.routes.remove(route) for route in app.routes if method in route.methods and route.path == path]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## First endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5304]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:51916 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51916 - \"GET /openapi.json HTTP/1.1\" 200 OK\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Shutting down\n",
      "INFO:     Waiting for application shutdown.\n",
      "INFO:     Application shutdown complete.\n",
      "INFO:     Finished server process [5304]\n"
     ]
    }
   ],
   "source": [
    "@app.get(\"/\")\n",
    "def get_root():\n",
    "    return {\"Hello\": \"World\"}\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/get_root__get\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/get_root__get\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Endpoint with URL params\n",
    "\n",
    "- Use Python string composition chars (`{abcd}`) as placeholders for URL params, in the endpoint route string\n",
    "- Define them as function params (with the same name as used on the route)\n",
    "- Typing the params will limit their input type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5334]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:51942 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51942 - \"GET /openapi.json HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51950 - \"GET /people/123abc HTTP/1.1\" 422 Unprocessable Entity\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Shutting down\n",
      "INFO:     Waiting for application shutdown.\n",
      "INFO:     Application shutdown complete.\n",
      "INFO:     Finished server process [5334]\n"
     ]
    }
   ],
   "source": [
    "@app.get(\"/people/{person_id}\")\n",
    "def get_person(person_id: int):\n",
    "    return {\"person_id\": person_id}\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/get_person_people__person_id__get\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/get_person_people__person_id__get\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Status code: 422\n",
      "Response: {'detail': [{'loc': ['path', 'person_id'], 'msg': 'value is not a valid integer', 'type': 'type_error.integer'}]}\n"
     ]
    }
   ],
   "source": [
    "# Trying to set param as text with letters, when it MUST be int (unparseable)\n",
    "import requests\n",
    "\n",
    "r = requests.get(\"http://localhost:8000/people/123abc\")\n",
    "print(\"Status code:\", r.status_code)\n",
    "print(\"Response:\", r.json())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Endpoint with query params\n",
    "\n",
    "Function args that are not dict or Pydantic models, nor set on the URL, are treated as query params"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5362]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:51974 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51980 - \"GET /people?surname=foolanito HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51974 - \"GET /openapi.json HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:51988 - \"GET /people HTTP/1.1\" 422 Unprocessable Entity\n",
      "INFO:     127.0.0.1:51994 - \"GET /people?surname=foolanito HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:52000 - \"GET /people HTTP/1.1\" 422 Unprocessable Entity\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Shutting down\n",
      "INFO:     Waiting for application shutdown.\n",
      "INFO:     Application shutdown complete.\n",
      "INFO:     Finished server process [5362]\n"
     ]
    }
   ],
   "source": [
    "from typing import Optional\n",
    "\n",
    "@app.get(\"/people\")\n",
    "def list_people(surname: str, city: Optional[str] = None, limit: int = 20):\n",
    "    return {\n",
    "        \"filters\": {\n",
    "            \"surname\": surname,\n",
    "            \"city\": city\n",
    "        },\n",
    "        \"limit\": limit\n",
    "    }\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/list_people_people_get\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/list_people_people_get\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Status code: 200\n",
      "Response: {'filters': {'surname': 'foolanito', 'city': None}, 'limit': 20}\n"
     ]
    }
   ],
   "source": [
    "# Get with \"surname\" param only\n",
    "import requests\n",
    "\n",
    "r = requests.get(\"http://localhost:8000/people?surname=foolanito\")\n",
    "print(\"Status code:\", r.status_code)\n",
    "print(\"Response:\", r.json())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Status code: 422\n",
      "Response: {'detail': [{'loc': ['query', 'surname'], 'msg': 'field required', 'type': 'value_error.missing'}]}\n"
     ]
    }
   ],
   "source": [
    "# Get without params (\"surname\" is required)\n",
    "import requests\n",
    "\n",
    "r = requests.get(\"http://localhost:8000/people\")\n",
    "print(\"Status code:\", r.status_code)\n",
    "print(\"Response:\", r.json())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Endpoint with body + model\n",
    "\n",
    "- Using Pydantic BaseModel\n",
    "- Set as function param (typing it as the BaseModel class)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5399]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:52024 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:52024 - \"GET /openapi.json HTTP/1.1\" 200 OK\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Shutting down\n",
      "INFO:     Waiting for application shutdown.\n",
      "INFO:     Application shutdown complete.\n",
      "INFO:     Finished server process [5399]\n"
     ]
    }
   ],
   "source": [
    "from pydantic import BaseModel\n",
    "from typing import Optional\n",
    "\n",
    "class Person(BaseModel):\n",
    "    name: str\n",
    "    age: Optional[int]\n",
    "    profession: str = \"Student\"\n",
    "\n",
    "@app.post(\"/people\")\n",
    "def create_person(person: Person):\n",
    "    return person.dict()\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Try to:\n",
    "\n",
    "- Send age = \"5\"\n",
    "- Send age = \"5x\"\n",
    "- Send no age\n",
    "- Send no name\n",
    "- Send no profession\n",
    "- Send custom profession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/create_person_people_post\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/create_person_people_post\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Improving models & their documentation\n",
    "\n",
    "- Add a description as docstring\n",
    "- Set the value of fields to Field objects - to define the field description, constraints and more!\n",
    "\n",
    "Try to:\n",
    "\n",
    "- Send an age that is less than 18 years old\n",
    "- Send an empty string as age or profession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5425]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:52050 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:52050 - \"GET /openapi.json HTTP/1.1\" 200 OK\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Shutting down\n",
      "INFO:     Waiting for application shutdown.\n",
      "INFO:     Application shutdown complete.\n",
      "INFO:     Finished server process [5425]\n"
     ]
    }
   ],
   "source": [
    "from pydantic import BaseModel, Field\n",
    "from typing import Optional\n",
    "\n",
    "delete_route(\"POST\", \"/people\")\n",
    "\n",
    "class Person(BaseModel):\n",
    "    \"\"\"Just a human being, with a name, age and profession\"\"\"\n",
    "    name: str = Field(..., description=\"The name of this person\", example=\"Anna\", min_length=1)\n",
    "    age: Optional[int] = Field(None, description=\"The age of this person in years. Must be 18 years or older\", example=18, ge=18)  # (g)reater or (e)qual 18\n",
    "    profession: str = Field(\"Student\", description=\"The profession of this person\", example=\"Engineer\", min_length=1)\n",
    "\n",
    "@app.post(\"/people\")\n",
    "def create_person(person: Person):\n",
    "    return person.dict()\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/create_person_people_post\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/create_person_people_post\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Improving query params\n",
    "\n",
    "Query params can also be fully documented and have additional validators."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:     Started server process [5451]\n",
      "INFO:     Waiting for application startup.\n",
      "INFO:     Application startup complete.\n",
      "INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "INFO:     127.0.0.1:52076 - \"GET /docs HTTP/1.1\" 200 OK\n",
      "INFO:     127.0.0.1:52082 - \"GET /people?surname=Fulanito&limit=0 HTTP/1.1\" 422 Unprocessable Entity\n",
      "INFO:     127.0.0.1:52088 - \"GET /people?surname=Fulanito&limit=1001 HTTP/1.1\" 422 Unprocessable Entity\n"
     ]
    }
   ],
   "source": [
    "from typing import Optional\n",
    "from fastapi import Query\n",
    "\n",
    "delete_route(\"GET\", \"/people\")\n",
    "\n",
    "@app.get(\"/people\")\n",
    "def list_people(\n",
    "        surname: str = Query(..., description=\"Surname to filter by\"),\n",
    "        city: Optional[str] = Query(None, description=\"City to filter by\"),\n",
    "        limit: float = Query(20, description=\"Max results returned\", ge=1, le=1000)\n",
    "):\n",
    "    return {\n",
    "        \"filters\": {\n",
    "            \"surname\": surname,\n",
    "            \"city\": city\n",
    "        },\n",
    "        \"limit\": limit\n",
    "    }\n",
    "\n",
    "start_api()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe src=\"http://localhost:8000/docs#/default/list_people_people_get\" width=\"800\" height=\"600\"></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe src=\"http://localhost:8000/docs#/default/list_people_people_get\" width=\"800\" height=\"600\"></iframe>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "limit=0\n",
      "\tStatus code: 422\n",
      "\tResponse: {'detail': [{'loc': ['query', 'limit'], 'msg': 'ensure this value is greater than or equal to 1', 'type': 'value_error.number.not_ge', 'ctx': {'limit_value': 1}}]}\n",
      "limit=1001\n",
      "\tStatus code: 422\n",
      "\tResponse: {'detail': [{'loc': ['query', 'limit'], 'msg': 'ensure this value is less than or equal to 1000', 'type': 'value_error.number.not_le', 'ctx': {'limit_value': 1000}}]}\n"
     ]
    }
   ],
   "source": [
    "# Get with limit=0 and limit=1001\n",
    "import requests\n",
    "\n",
    "for limit in [0, 1001]:\n",
    "    r = requests.get(f\"http://localhost:8000/people?surname=Fulanito&limit={limit}\")\n",
    "    print(f\"limit={limit}\")\n",
    "    print(\"\\tStatus code:\", r.status_code)\n",
    "    print(\"\\tResponse:\", r.json())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}