Terraform - Déployer une API sur AWS Lambda - Partie 1

Publié le : 18 novembre 2021 | Mis à jour le : 22 janvier 2023

J’ai déjà écris plusieurs API durant ces dernières années et j’ai pu voir l’évolution des framework Python. J’ai commencé avec Flask, puis je suis passé à Connexion, développé par Zalando et depuis peu je m’intéresse à FasTAPI. Ce billet fait parti d’une série dont l’objectif est de déployer une API sur AWS en utilisant entre autre AWS Lambda, API Gateway et d’autres produits.

Ce premier billet abordera juste comment démarrer avec FastAPI pour écrire une API basique. FastAPI est un framework Python 3, apparu fin 2018, qui est facile à prendre en main. En effet, il intègre toutes les fonctionnalités attendues pour développer une API et permet donc de se concentrer sur l’essentiel qui est l’intégration de la partie fonctionnelle de votre API. Cerise sur le gâteau il est annoncé comme étant aussi rapide que les backend Javascript.

FastAPI repose sur deux librairies Python qui sont :

Starlette, un micro framework Web ASGI
Pydantic qui gère la validation, la sérialisation et la documentation des données manipulées par FastAPI.

Démarrer le développement d’une API avec FastAPI

Je vais m’appuyer sur un template, fastapi-nano, qui intègre l’authentification Oauth2, mais surtout offre une structure de répertoire facilitant le support. Il propose aussi :

un makefile gérant tout le cycle de vie de notre API : installation, test, lancement local, … make help
des fichiers Dockerfile et Docker-compose qui vont nous servir à tester mais aussi à déployer notre API dans une Lambda AWS
la gestion des requirements de votre API avec pip-tools

git clone --depth=1 --branch=master https://github.com/rednafi/fastapi-nano.git firstapi
rm -rf firstapi/.git
cd firstapi && tree

.
├── app
│   ├── apis
│   │   ├── api_a
│   │   │   ├── __init__.py
│   │   │   ├── mainmod.py
│   │   │   └── submod.py
│   │   └── api_b
│   │       ├── __init__.py
│   │       ├── mainmod.py
│   │       └── submod.py
│   ├── core
│   │   ├── auth.py
│   │   ├── config.py
│   │   └── __init__.py
│   ├── __init__.py
│   ├── main.py
│   ├── routes
│   │   └── views.py
│   └── tests
│       ├── __init__.py
│       ├── test_apis.py
│       └── test_functions.py
├── Caddyfile
├── docker-compose.yml
├── Dockerfile
├── LICENSE
├── makefile
├── pyproject.toml
├── README.md
├── requirements-dev.in
├── requirements-dev.txt
├── requirements.in
├── requirements.txt
└── scripts
    ├── docker_ci.sh
    └── update_deps.sh

Créons notre environnement virtuel avec pyenv :

pyenv install 3.9.7
pyenv virtualenv 3.9.7 firstapi
pyenv local firstapi

Maintenant installons fastapi en utilisant pip-tools :

pipx install pip-tools
pip-compile requirements.in
pip-compile requirements-dev.in
pip install -r requirements.txt && pip install -r requirements-dev.txt

Lançons l’API une première fois :

uvicorn app.main:app --port 5000 --reload

INFO:     Will watch for changes in these directories: ['/home/vagrant/Projets/firstapi']
INFO:     Uvicorn running on http://127.0.0.1:5000 (Press CTRL+C to quit)
INFO:     Started reloader process [80878] using statreload
INFO:     Started server process [80924]

Allez sur la page de la documentation openapi et aussi la page redoc.

fasapi openapi

Lançons notre api dans un container :

sudo curl -L "https://github.com/docker/compose/releases/download/1.29.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
docker volume create --name=caddy_data
make run-container

Voila on a tout pour démarrer et se concentrez sur le code de notre API.

Faisons le tour du boilerplate Fastapi nano

Instancions une API

Lorsque vous avez lancé la commande uvicorn app.main:app --port 5000 --reload nous avons fait appel au code python du fichier main.py du répertoire app dont voici le contenu :

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

from app.core import auth
from app.routes import views

app = FastAPI()

# Set all CORS enabled origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(views.router)

On instancie une API se nommant app, on pourrait l’appeler autrement. On lui ajoute la gestion des CORS avec app.add_middleware.

Gestion des routes de notre API

On créé deux routes auth et views se trouvant respectivement dans les fichiers app/core/auth.pyet app/routes/views.py.

from app.core import auth
from app.routes import views

app.include_router(auth.router)
app.include_router(views.router)

Ouvrons le fichier app/routes/views.py :

from __future__ import annotations

from fastapi import APIRouter, Depends

from app.apis.api_a.mainmod import main_func as main_func_a
from app.apis.api_b.mainmod import main_func as main_func_b
from app.core.auth import get_current_user

router = APIRouter()


@router.get("/api_a/{num}", tags=["api_a"])
async def view_a(
    num: int,
    auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(num)


@router.get("/api_b/{num}", tags=["api_b"])
async def view_b(
    num: int,
    auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_b(num)

Une route ou chemin (path) est instanciée en utilisant APIRouter :

from fastapi import APIRouter`
router = APIRouter()

Puis on déclare la route en utilisant le décorateur @router.<methode>. Methode pouvant être : get, post, put, delete (voir options, head, patch ou trace).

@router.get("/api_a/{num}", tags=["api_a"])

Ce décorateur prend des arguments optionnels comme ici tags qui permet de classer les endpoints par catégories dans la documentation.

Ensuite on définit la fonction, avec ces arguments et la méthode qui sera utilisée pour traiter la demande.

@router.get("/api_a/{num}", tags=["api_a"])
async def view_a(
    num: int,
    auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(num)

Paramètre(s) de la route

Le paramètre de la route num est déclaré de type entier. En fait cela va permettre à fastAPI de parser et de contrôler que la valeur envoyée dans la requête est bien du type attendu (donc vous n’aurez pas à le gérer).

Attention si vous devez définir des routes avec des paramètres fixes il faudra les déclarer en premier.


@router.get("/api_a/1", tags=["api_a"])
async def view_a_1(
  auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(1)

@router.get("/api_a/{num}", tags=["api_a"])
async def view_a(
    num: int,
    auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(num)

Pour fixer les valeurs possibles d’un paramètre il faudra utiliser les Enums de python.

from enum import Enum

class ModelName(str, Enum):
    one   = "1"
    two   = "2"
    three = "3"

@app.get("/api_a/{num}")
async def view_a(num: ModelName):
    if model_name == ModelName.one:
        ...
    if model_name.value == "2":
        ...

Paramètre(s) de la requête

Quand vous déclarez des paramètres dans votre fonction de chemin qui ne font pas partie des paramètres indiqués dans le chemin associé, ces paramètres sont automatiquement considérés comme des paramètres de “requête”. Prenons l’exemple de l’url suivante : http://127.0.0.1:8000/api_a/1?skip=0&limit=10. On trouve deux paramètres de requêtes skip et limit qui sont déclarés dans le corps de la fonction ainsi :

@router.get("/api_a/1", tags=["api_a"])
async def view_a_1(
  skip: int = 0,
  limit: int = 10,
  query: str = None,
  auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(1)

On peut attribuer des valeurs par défaut pour les rendre optionnels limit: int = 10, mais aussi ainsi : query: str = None. Pour les rendre obligatoire il suffit simplement de les déclarer sans valeur par défaut.

Déclaration des méthodes

La fonction de la route retourne le résultat fournit par la méthode main_func_a(num) qui retourne un dictionnaire.

Cette méthode est déclarée ainsi :

from app.apis.api_a.mainmod import main_func as main_func_a

Elle se trouve donc dans le fichier app/apis/api_a/main_mod.py.

Création des méthodes

Étudions la déclarations des méthodes, pour cela ouvrons le fichier app/apis/api_a/main_mod.py :

from __future__ import annotations

from .submod import rand_gen


def main_func(num: int) -> dict[str, int]:
    d = rand_gen(num)
    return d

On voit que l’on fait appel à des fonctions déclarées dans des librairies. Ici ./submod.py :

from .submod import rand_gen
    d = rand_gen(num)

from __future__ import annotations
import random


def rand_gen(num: int) -> dict[str, int]:
    num = int(num)
    d = {
        "seed": num,
        "random_first": random.randint(0, num),
        "random_second": random.randint(0, num),
    }
    return d

Gestion de l’authentification

Comme dis précédemment la partie authentification est géré par la route auth :

from app.core import auth

app.include_router(auth.router)

Elle se trouve donc dans le répertoire app/core/auth.py. Je ne vais pas recopier le contenu mais m’attarder sur comment cela fonctionne.

Le user et mot de passe est défini dans le fichier config.py :

import os

from dotenv import load_dotenv

load_dotenv("./.env")


API_USERNAME = os.environ["API_USERNAME"]
API_PASSWORD = os.environ["API_PASSWORD"]

# Auth configs.
API_SECRET_KEY = os.environ["API_SECRET_KEY"]
API_ALGORITHM = os.environ["API_ALGORITHM"]
API_ACCESS_TOKEN_EXPIRE_MINUTES = int(
    os.environ["API_ACCESS_TOKEN_EXPIRE_MINUTES"]
)  # infinity

Il récupère les informations stockées dans des variables d’environnement présentes dans le fichier .env :

HOST="0.0.0.0" # localhost
PORT="5000"    # port to access the app

# App config.

API_USERNAME="ubuntu"
API_PASSWORD="debian"

# To get a string like this run:
# openssl rand -hex 32
API_SECRET_KEY="09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
API_ALGORITHM="HS256"
API_ACCESS_TOKEN_EXPIRE_MINUTES="5256000000"  # infinity

Si vous voulez modifier ces valeurs ca se fera donc dans ce fichier. Si vous voulez implémenter d’autres méthodes d’authentification elle sont assez bien documenté sur le site de FastAPI.

Pour restreindre des endpoints il suffit d’ajouter dans la définition de la fonction auth: pointant sur la méthode de contrôle.


from app.core.auth import get_current_user

@router.get("/api_a/1", tags=["api_a"])
async def view_a_1(
  skip: int = 0,
  limit: int = 10,
  query: str = None,
  auth: Depends = Depends(get_current_user),
) -> dict[str, int]:
    return main_func_a(1)

La suite dans un prochain billet ou nous aborderons les tests unitaires en autre.

Stéphane ROBERT

Tous les billets sur :