Keyboard shortcuts

Press or to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Web development with Python FastAPI

Install FastAPI

  • FastAPI to build web application back-ends that serve JSON or other data formats.

  • Python 3.8 or newer is needed

  • In order to get started one needs to install FastAPI manually, or, as we can see on the next page add it as part of the requirements.txt file.

pip install "fastapi[all]"

FastAPI - Hello World

  • FastAPI

  • get

  • virtualenv

  • The requirements include both FastAPI and pytest so we can write and run tests for our web application.

fastapi[all]
pytest
  • It is recommended to set up a virtual environment or use some other way to separate environments.
virtualenv -p python3 venv
source venv/bin/activate
pip install -r requirements.txt
  • We need a single file in which we import the FastAPI class.
  • We then create an instance of it. We can call it any name, but most of the examples are using app so we'll use that as well.
  • For each URL path we need to create a mapping to indicate which function needs to be executed when someone access that path.
  • The functions are defined as async so FastAPI can handle several requests at the same time. The name of the function does not matter.
  • The decorator @app.get("/") is doing the mapping.
  • The function needs to return some Python data structure that will be converted to JSON when it is sent back to the client.
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}

  • In order to see this working launch the development web server that comes with the installation of FastAPI.
fastapi dev main.py

Then visit http://localhost:8000/

You can also visit some other pages on this site:

  • http://localhost:8000/docs to see the documentation generated by Swagger UI

  • http://localhost:8000/redoc to see the documentation generated by Redoc

  • http://localhost:8000/openapi.json

  • path - endpoint - route

FastAPI - Test Hello World

  • TestClient
  • assert

Writing the web application is nice, but we better also write tests that verify the application works properly. This will make it easier to verify that none of the changes we introduce later on will break parts that have been working and tested already.

The key here is to have a file that starts with test_ that has a function name that starts with test_ that uses assert to check values.

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.headers["content-type"] == "application/json"
    assert response.json() == {"message": "Hello World"}

Just run pytest to execute the tests.

pytest

FastAPI with Docker compose

{% embed include file="src/examples/fastapi/Dockerfile)

{% embed include file="src/examples/fastapi/docker-compose.yml)

fastapi[all]

pytest
requests

motor
docker compose up
docker exec -it fastapi_app_1 bash
uvicorn main:app --reload --host=0.0.0.0

FastAPI - Dynamic response

  • datetime

  • A small step ahead generating part of the content of the response dynamically.

from fastapi import FastAPI
import datetime

app = FastAPI()


@app.get("/")
async def root():
    return {"message": f"Hello World at {datetime.datetime.now()}"}

FastAPI - Echo GET - Query Parameters

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root(text):
    return {"message": f"You wrote: '{text}'"}

http://localhost:8000/?text=Foo%20Bar

FastAPI - Echo GET - Query Parameters - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 422
    assert response.json() == {
        'detail': [
            {
                'loc': ['query', 'text'],
                'msg': 'field required',
                'type': 'value_error.missing'
            }]}

def test_main_param():
    response = client.get("/?text=Foo Bar")
    assert response.status_code == 200
    assert response.json() == {'message': "You wrote: 'Foo Bar'"}

FastAPI - Echo POST - request body

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    text: str

@app.post("/")
async def root(data: Item):
    return {"message": f"You wrote: '{data.text}'"}


http://localhost:8000/?text=Foo%20Bar
curl -d '{"text":"Foo Bar"}' -H "Content-Type: application/json" -X POST http://localhost:8000/
import requests
res = requests.post('http://localhost:8000/',
    headers = {
        #'User-agent'  : 'Internet Explorer/2.0',
        'Content-type': 'application/json'
    },
    json = {"text": "Fast API"},
)
#print(res.headers['content-type'])
print(res.text)

FastAPI - Echo POST - request body - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)

def test_main_param():
    response = client.post("/", json={"text": "Foo Bar"})
    assert response.status_code == 200
    assert response.json() == {'message': "You wrote: 'Foo Bar'"}

FastAPI - Calculator GET

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def main(a: int, b: int):
    return {"message": a+b}
http://localhost:8000/?a=2&b=3

FastAPI - Calculator GET - Test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_main():
    response = client.get("/?a=2&b=3")
    assert response.status_code == 200
    assert response.json() == {'message': 5}

    response = client.get("/?a=2&b=x")
    assert response.status_code == 422
    assert response.json() == {
        'detail': [
            {
                'loc': ['query', 'b'],
                'msg': 'value is not a valid integer',
                'type': 'type_error.integer'
            }]}

FastAPI - Path Parameters - str

from fastapi import FastAPI

app = FastAPI()


@app.get("/user/{user_name}")
async def root(user_name: str):
    return {'msg': f"user '{user_name}'"}
http://localhost:8000/user/foobar
http://localhost:8000/user/foo bar


http://localhost:8000/user/ - 404

FastAPI - Path Parameters - str - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_foobar():
    response = client.get("/user/foobar")
    assert response.status_code == 200
    assert response.json() == {'msg': "user 'foobar'"}

def test_foo_bar():
    response = client.get("/user/foo bar")
    assert response.status_code == 200
    assert response.json() == {'msg': "user 'foo bar'"}


def test_user():
    response = client.get("/user/")
    assert response.status_code == 404
    assert response.json() == {'detail': 'Not Found'}

FastAPI - Path Parameters - int

from fastapi import FastAPI

app = FastAPI()


@app.get("/user/{user_id}")
async def root(user_id: int):
    return {'user': user_id}
http://localhost:8000/user/23      works
http://localhost:8000/user/foo     422 error
http://localhost:8000/user/2.3     422 error

FastAPI - Path Parameters - int - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_int():
    response = client.get("/user/23")
    assert response.status_code == 200
    assert response.json() == {'user': 23}

def test_str():
    response = client.get("/user/foo")
    assert response.status_code == 422
    assert response.json() == {
        'detail': [
            {
                'loc': ['path', 'user_id'],
                'msg': 'value is not a valid integer',
                'type': 'type_error.integer'
            }]}

def test_float():
    response = client.get("/user/2.3")
    assert response.status_code == 422
    assert response.json() == {
        'detail': [
            {
                'loc': ['path', 'user_id'],
                'msg': 'value is not a valid integer',
                'type': 'type_error.integer'
            }]}

def test_nothing():
    response = client.get("/user/")
    assert response.status_code == 404
    assert response.json() == {'detail': 'Not Found'}

FastAPI - Path Parameters - specific values with enum

from enum import Enum
from fastapi import FastAPI


class CarTypeName(str, Enum):
    tesla = "Tesla"
    volvo = "Volvo"
    fiat  = "Fiat"



app = FastAPI()


@app.get("/car/{car_type}")
async def get_car(car_type: CarTypeName):
    print(car_type) # CarTypeName.tesla
    if car_type == CarTypeName.tesla:
        print("in a Tesla")
    return {'car_type': car_type}
http://localhost:8000/car/Volvo

http://localhost:8000/car/volvo     error

FastAPI - Path Parameters - specific values with enum - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_Volvo():
    response = client.get("/car/Volvo")
    assert response.status_code == 200
    assert response.json() == {'car_type': 'Volvo'}

def test_volvo():
    response = client.get("/car/volvo")
    assert response.status_code == 422
    assert response.json() == {
        'detail': [
            {
                'ctx': {'enum_values': ['Tesla', 'Volvo', 'Fiat']},
                'loc': ['path', 'car_type'],
                'msg': 'value is not a valid enumeration member; permitted: ' "'Tesla', 'Volvo', 'Fiat'",
                'type': 'type_error.enum'
            }]}


FastAPI - Path containing a directory path

from fastapi import FastAPI

app = FastAPI()

@app.get("/shallow/{filepath}")
async def get_filename(filepath: str):
    return {'shallow': filepath}

@app.get("/deep/{filepath:path}")
async def get_path(filepath: str):
    return {'deep': filepath}
http://localhost:8000/shallow/a.txt    works
http://localhost:8000/shallow/a/b.txt  not found

http://localhost:8000/deep/a.txt       works
http://localhost:8000/deep/a/b.txt     works

FastAPI - Path containing a directory path - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_shallow_one():
    response = client.get("/shallow/a.txt")
    assert response.status_code == 200
    assert response.json() == {'shallow': 'a.txt'}

def test_shallow_more():
    response = client.get("/shallow/a/b.txt")
    assert response.status_code == 404
    assert response.json() == {'detail': 'Not Found'}


def test_deep_one():
    response = client.get("/deep/a.txt")
    assert response.status_code == 200
    assert response.json() == {'deep': 'a.txt'}

def test_deep_more():
    response = client.get("/deep/a/b.txt")
    assert response.status_code == 200
    assert response.json() == {'deep': 'a/b.txt'}

Return main HTML page

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/")
async def root():
    data = '<a href="/hello">hello</a>'
    return Response(content=data, media_type="text/html")


@app.get("/hello")
async def hello():
    return {"message": "Hello World"}

Return main HTML page - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.content == b'<a href="/hello">hello</a>'

def test_hello():
    response = client.get("/hello")
    assert response.status_code == 200
    assert response.json() == {'message': 'Hello World'}

Return main HTML file

from fastapi import FastAPI, Response
import os
root = os.path.dirname(os.path.abspath(__file__))

app = FastAPI()

@app.get("/")
async def main():
    #print(root)
    with open(os.path.join(root, 'index.html')) as fh:
        data = fh.read()
    return Response(content=data, media_type="text/html")

@app.get("/hello")
async def hello():
    return {"message": "Hello World"}

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1, user-scalable=yes">

  <title>Demo FastAPI</title>
</head>
<body>
<h1>Main subject</h1>


<a href="/hello">hello</a>

</body>
</html>

Send 400 error

  • user/{id} but we don't have that specific id.
  • abort(400) with some status code
from fastapi import FastAPI, Response, status

app = FastAPI()


@app.get("/user/{user_id}")
async def root(user_id: int, response: Response):
    if user_id > 40:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {'detail': 'User does not exist'}
    return {'user': user_id}

Send 400 error - test

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_good():
    response = client.get("/user/23")
    assert response.status_code == 200
    assert response.json() == {'user': 23}

def test_bad():
    response = client.get("/user/42")
    assert response.status_code == 400
    assert response.json() == {'detail': 'User does not exist'}

FastAPI - in memory counter

from fastapi import FastAPI

app = FastAPI()

counter = 0

@app.get("/")
async def main():
    global counter
    counter += 1
    return {"cnt": counter}

FastAPI - on disk counter

from fastapi import FastAPI
import os
root = os.path.dirname(os.path.abspath(__file__))
filename = os.path.join(root, 'counter.txt')

app = FastAPI()

@app.get("/")
async def main():
    if os.path.exists(filename):
        with open(filename) as fh:
            counter = int(fh.read())
    else:
        counter = 0
    counter += 1
    with open(filename, 'w') as fh:
        fh.write(str(counter))

    return {"cnt": counter}

FastAPI - on disk multi-counter uising JSON

from fastapi import FastAPI, Response
import json
import os
root = os.path.dirname(os.path.abspath(__file__))
filename = os.path.join(root, 'counter.json')

app = FastAPI()

@app.get("/{name}")
async def count(name):
    counters = load_counters()
    if name not in counters:
        counters[name] = 0

    counters[name] += 1

    with open(filename, 'w') as fh:
        json.dump(counters, fh)

    return {"cnt": counters[name]}

@app.get("/")
async def main():
    counters = load_counters()
    if counters:
        html = '<table>\n'
        for name in sorted(counters.keys()):
            html += f'<tr><td><a href="/{name}">{name}</a></td><td>{counters[name]}</td></tr>\n'
        html += '</table>\n'
    else:
        html = 'Try accessing <a href="/foo">/foo</a>';
    return Response(content=html, media_type="text/html")


def load_counters():
    if os.path.exists(filename):
        with open(filename) as fh:
            counters = json.load(fh)
    else:
        counters = {}

    return counters

FastAPI - get header from request

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/")
async def main(request: Request):
    print(request.headers)
    # Headers({
    #     'host': 'testserver',
    #     'user-agent': 'testclient',
    #     'accept-encoding': 'gzip, deflate',
    #     'accept': '*/*',
    #     'connection': 'keep-alive',
    #     'x-some-field': 'a value'
    # })

    #print(request.client)
    print(request.client.host) # testclient
    print(request.client.port) # 50000
    return {"message": "Hello World"}

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/", headers={"x-some-field": "a value"})
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}
    assert response.headers == {'content-length': '25', 'content-type': 'application/json'}

FastAPI - set arbitrary header in response

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/")
async def main(response: Response):
    response.headers['X-something-else'] = "some value"
    return {"message": "Hello World"}

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}
    #assert response.headers == {'content-length': '25', 'content-type': 'application/json'}
    assert response.headers == {'content-length': '25', 'content-type': 'application/json', 'x-something-else': 'some value'}

FastAPI - serve static files - JavaScript example

import os

from fastapi import FastAPI, Response
from fastapi.staticfiles import StaticFiles

root = os.path.dirname(os.path.abspath(__file__))

app = FastAPI()

app.mount("/js", StaticFiles(directory=os.path.join(root, 'js')), name="js")

@app.get("/")
async def main():
    with open(os.path.join(root, 'index.html')) as fh:
        data = fh.read()
    return Response(content=data, media_type="text/html")


function demo() {
    console.log("demo");
    document.getElementById("content").innerHTML = "Written by JavaScript";
}

demo();
<h1>Static HTML</h2>

<div id="content"></div>

<script src="/js/demo.js"></script>

FastAPI mounted sub-applications

import os

from fastapi import FastAPI, Response
from api_v1 import api_v1

root = os.path.dirname(os.path.abspath(__file__))

app = FastAPI()

app.mount("/api/v1", api_v1)

@app.get("/")
async def main():
    return Response(content='main <a href="/api/v1">/api/v1</a>', media_type="text/html")

from fastapi import FastAPI

api_v1 = FastAPI()


@api_v1.get("/")
def main():
    return {"message": "Hello World from API v1"}



uvicorn main:api_v1 --reload --host=0.0.0.0
uvicorn main:main --reload --host=0.0.0.0

FastAPI - todo

  • Access to MongoDB
  • Access to PostgreSQL
  • Access to SQLite
  • Session ???
  • Can we have the query parameters and the internal variables be different (e.g. paramName and param_name ?)
  • handel exceptions (500 errors)
pip install "uvicorn[standard]"
uvicorn main:app --reload