Comment intégrer FastAPI avec des bases de données SQL et NoSQL

L'une des raisons pour lesquelles FastAPI est devenu un framework populaire dans le développement d'APIs repose sur sa flexibilité et son intégration fluide avec divers types de bases de données, qu'elles soient relationnelles (SQL) ou non relationnelles (NoSQL). Dans cette section, nous explorerons comment intégrer FastAPI avec ces systèmes de gestion de bases de données, en mettant l'accent sur les bonnes pratiques et les techniques les plus efficaces pour travailler avec des bases de données modernes.

Intégration avec SQL

L'intégration commence par la configuration de la connexion à la base de données via SQLAlchemy, en définissant les informations de connexion et en initialisant la session de la base de données. Une fois cette configuration en place, les opérations CRUD (Create, Read, Update, Delete) peuvent être facilement mises en œuvre en utilisant des modèles de données définis avec SQLAlchemy. Ces modèles permettent de décrire la structure des tables et des relations entre elles.

L'une des fonctionnalités essentielles de SQLAlchemy est la gestion des migrations de base de données. Cela permet de gérer les changements de schéma de manière sécurisée et évolutive, sans perte de données. Des outils comme Alembic sont couramment utilisés pour automatiser et faciliter ce processus.

Gestion des transactions et de la concurrence

Une autre caractéristique importante de l'intégration avec les bases de données SQL est la gestion des transactions et de la concurrence. FastAPI prend en charge les transactions via SQLAlchemy, ce qui permet de garantir l'intégrité des données, même dans des environnements à forte concurrence. Les transactions permettent d'assurer que les groupes d'opérations sur la base de données sont exécutés de manière atomique, c'est-à-dire qu'elles réussissent ou échouent ensemble.

Intégration avec NoSQL

FastAPI offre également une excellente prise en charge des bases de données NoSQL, telles que MongoDB et Redis. MongoDB, par exemple, est une base de données orientée documents qui permet de stocker des données semi-structurées sous forme de BSON (Binary JSON). FastAPI se connecte à MongoDB à l'aide de bibliothèques comme motor ou pymongo, qui offrent des interfaces Python pour interagir avec la base de données.

L'un des principaux avantages de MongoDB réside dans sa capacité à gérer de grandes quantités de données non structurées, comme celles provenant d'applications de type big data. Dans ce cas, l'utilisation de la sérialisation et de la désérialisation des données est cruciale, notamment pour garantir que les objets Python soient correctement convertis avant d'être stockés dans la base de données et vice versa.

Sécurisation des données

Dans le cadre de l'intégration avec des bases de données, la sécurité des données est une préoccupation essentielle. Il est crucial de prendre en compte des mesures telles que le chiffrement des données sensibles, que ce soit en transit (via TLS) ou au repos (via des mécanismes de chiffrement de la base de données). Par ailleurs, l'accès aux bases de données doit être restreint en fonction des rôles des utilisateurs afin de prévenir tout accès non autorisé.

L'implémentation de mécanismes de sécurité comme l'authentification et l'autorisation, à travers des protocoles tels qu'OAuth2 et JWT (JSON Web Token), est indispensable pour sécuriser les interactions avec les APIs.

Optimisation et bonnes pratiques

L'optimisation des performances est un aspect crucial lorsqu'on travaille avec des bases de données. Il est essentiel de réduire le nombre de requêtes inutiles et de privilégier des méthodes de chargement efficaces, telles que le "lazy loading" et le "eager loading". Le "lazy loading" permet de charger les données uniquement lorsque cela est nécessaire, tandis que l'"eager loading" charge toutes les données nécessaires en une seule requête.

Il est aussi recommandé d'utiliser des techniques comme le caching pour réduire la charge sur les bases de données. Redis, par exemple, est souvent utilisé comme un cache de données pour améliorer les performances des applications web.

La gestion des index est également un point clé dans l'optimisation des bases de données. En créant des index sur les colonnes fréquemment utilisées dans les requêtes, on peut accélérer considérablement les recherches, bien que cela puisse ajouter un coût en termes de temps lors des opérations d'insertion ou de mise à jour des données.

Enfin, une surveillance constante des performances est nécessaire. Des outils comme SQLAlchemy et MongoDB offrent des moyens de profiler les requêtes pour identifier les goulets d'étranglement et optimiser les performances.

Conclusion

Comment implémenter l'authentification multifacteur (MFA) et la gestion des clés API dans une application FastAPI

L'ajout de l'authentification multifacteur (MFA) dans une application renforce considérablement sa sécurité. En obligeant les utilisateurs à fournir deux facteurs de vérification distincts pour accéder à une ressource, on réduit les risques liés aux compromissions de mot de passe. Dans cette section, nous allons explorer comment intégrer la MFA dans une application FastAPI en utilisant une méthode basée sur un mot de passe à usage unique basé sur le temps (TOTP), et comment gérer l'authentification par clé API.

Implémentation de la MFA avec FastAPI

Pour implémenter la MFA dans une application FastAPI, nous allons utiliser la bibliothèque Python pyotp, qui permet de générer des mots de passe à usage unique basés sur le temps (TOTP). Ce système génère un code à six ou huit chiffres qui est valable pendant une courte période, généralement 30 secondes. Voici les étapes pour intégrer la MFA dans l'application.

Préparation

Commencez par installer la bibliothèque nécessaire :

bash
Copy code
pip install pyotp

Nous allons ensuite modifier la table des utilisateurs dans la base de données pour inclure un champ supplémentaire totp_secret, qui contiendra le secret utilisé pour générer le code TOTP.

python
Copy code
class User(Base):
    # autres champs existants
    totp_secret: Mapped[str] = mapped_column(
        nullable=True
    )

Création des fonctions d'assistance

Nous créons ensuite deux fonctions pour générer le secret TOTP et l'URI TOTP utilisée par l'authentificateur. L'URI TOTP peut être utilisée pour générer un QR code, simplifiant ainsi l'intégration de l'authentificateur.

python
Copy code
import pyotp

def generate_totp_secret():
    return pyotp.random_base32()
def generate_totp_uri(secret, user_email):
    return pyotp.totp.TOTP(secret).provisioning_uri(
        name=user_email, issuer_name="YourAppName"
    )

Activation de la MFA

Pour activer la MFA, nous créons un point de terminaison (/user/enable-mfa) qui génère un secret TOTP, l'enregistre dans la base de données et renvoie l'URI TOTP pour générer un QR code dans l'interface utilisateur. Le code temporaire généré par l'authentificateur est également retourné à titre d'exemple.

python
Copy code
from fastapi import APIRouter, Depends, HTTPException, status

from sqlalchemy.orm import Session
from db_connection import get_session
from operations import get_user
from rbac import get_current_user
from responses import UserCreateResponse
router = APIRouter()
@router.post("/user/enable-mfa")
def enable_mfa(
    user: UserCreateResponse = Depends(get_current_user),
    db_session: Session = Depends(get_session),
):
    secret = generate_totp_secret()
    db_user = get_user(db_session, user.username)
    db_user.totp_secret = secret
    db_session.add(db_user)
    db_session.commit()
    totp_uri = generate_totp_uri(secret, user.email)
    return {
        "totp_uri": totp_uri,
        "secret_numbers": pyotp.TOTP(secret).now(),
    }

Vérification du code TOTP

Une fois la MFA activée, il est nécessaire de vérifier le code TOTP fourni par l'utilisateur. Nous créons un point de terminaison (/verify-totp) qui permet à l'utilisateur de soumettre son code pour vérification.

python
Copy code
@app.post("/verify-totp")
def verify_totp(
    code: str, username: str, session: Session = Depends(get_session)
):
    user = get_user(session, username)
    if not user.totp_secret:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="MFA not activated",
        )
    totp = pyotp.TOTP(user.totp_secret)
    if not totp.verify(code):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid TOTP token",
        )
    return {"message": "TOTP token verified successfully"}

L'authentification par clé API

Gestion des clés API

Créons un module api_key.py pour gérer les clés API. Nous définissons une liste de clés API valides et une méthode pour les vérifier.

python
Copy code
from fastapi import HTTPException
from typing import Optional
VALID_API_KEYS = [
    "verysecureapikey",
    "anothersecureapi",
    "onemoresecureapi",
]

async def get_api_key(api_key: Optional[str]):

    if api_key not in VALID_API_KEYS:
        raise HTTPException(
            status_code=403, detail="Invalid API Key"
        )
    return api_key

Endpoint protégé par clé API

Ensuite, nous ajoutons un point de terminaison protégé par clé API. En ajoutant simplement une dépendance au point de terminaison, nous sécurisons l'accès aux données sensibles.

python
Copy code
from fastapi import APIRouter
router = APIRouter()
@router.get("/secure-data")

async def get_secure_data(api_key: str = Depends(get_api_key)):

    return {"message": "Access to secure data granted"}

Meilleures pratiques pour la gestion des clés API

Bien que la gestion des clés API dans cet exemple soit simple, en production, il est essentiel de suivre certaines bonnes pratiques pour assurer la sécurité de l'application :

1. Transmission sécurisée : Utilisez toujours HTTPS pour empêcher l'interception des clés API pendant leur transmission.

2. Rotation des clés : Modifiez régulièrement les clés API pour limiter les risques de compromission.

3. Permissions minimales : Appliquez le principe du moindre privilège en attribuant uniquement les permissions nécessaires à chaque clé API.

Conclusion

L'intégration de l'authentification multifacteur et de la gestion des clés API dans une application FastAPI permet de renforcer considérablement la sécurité de celle-ci. En combinant ces deux méthodes, même si le mot de passe d'un utilisateur est compromis, un attaquant ne pourra pas accéder aux données sensibles sans avoir également accès à l'appareil MFA. Cette approche garantit ainsi une meilleure protection contre les attaques externes.

Comment intégrer Elasticsearch et Redis dans une application FastAPI

L'intégration de bases de données NoSQL dans une application FastAPI peut considérablement améliorer les performances et la scalabilité de votre système. Elasticsearch, un moteur de recherche et d'analyse basé sur Lucene, est une solution puissante pour indexer et rechercher rapidement de grandes quantités de données. Redis, quant à lui, est un magasin de données en mémoire utilisé principalement pour le cache, ce qui permet de réduire la charge sur les bases de données et d'accélérer les temps de réponse des API. Cette section explore comment utiliser Elasticsearch et Redis avec FastAPI pour améliorer les performances de votre application.

Création et gestion d'index Elasticsearch

Voici un exemple de code pour créer l'index et le remplir avec des données :

python
Copy code
async def main():
    await create_index()
    await fill_elasticsearch()  # uniquement si vous l'utilisez
if __name__ == "__main__":
    import asyncio
    asyncio.run(create_index())

Construction de la requête Elasticsearch

Une fois l'index rempli, il est nécessaire de définir une requête pour récupérer des informations spécifiques, comme les 10 artistes les plus populaires dans un pays donné. La requête suivante montre comment récupérer ces informations en utilisant un champ views_per_country qui est mappé pour chaque pays.

Voici la fonction qui construit cette requête pour Elasticsearch :

python
Copy code
def top_ten_songs_query(country) -> dict:

    views_field = f"views_per_country.{country}"
    query = {
        "bool": {
            "must": {"match_all": {}},
            "filter": [
                {"exists": {"field": views_field}}
            ],
        }
    }
    sort = {views_field: {"order": "desc"}}
    source = [
        "title", 
        views_field, 
        "album.title", 
        "artist"
    ]
    return {
        "index": "songs_index",
        "query": query,
        "size": 10,
        "sort": sort,
        "source": source,
    }

Création de l'endpoint FastAPI

Une fois la requête construite, l'intégration de cette logique dans FastAPI pour créer un endpoint est une étape simple. L'exemple suivant montre comment créer un endpoint qui utilise cette requête pour retourner les 10 meilleurs artistes par pays :

python
Copy code
from fastapi import APIRouter, Depends, HTTPException
from app.db_connection import es_client
router = APIRouter(prefix="/search", tags=["search"])
def get_elasticsearch_client():
    return es_client
@router.get("/top/ten/artists/{country}")
async def top_ten_artist_by_country(country: str, es_client=Depends(get_elasticsearch_client)):
    try:
        response = await es_client.search(*top_ten_songs_query(country))
    except BadRequestError as e:
        logger.error(e)
        raise HTTPException(status_code=400, detail="Invalid country")
    return [

        {"artist": record.get("key"), "views": record.get("views", {}).get("value")}

        for record in response["aggregations"]["top_ten_artists"]["buckets"]
    ]

Ce code utilise un client Elasticsearch pour interroger l'index des chansons et renvoyer les résultats triés par le nombre de vues pour un pays donné. L'endpoint est conçu de manière asynchrone, permettant ainsi de maximiser les performances de l'application.

Intégration de Redis pour le cache

L'ajout de Redis en tant que système de cache pour les réponses des requêtes peut considérablement améliorer les performances de l'application en réduisant le nombre de requêtes envoyées à Elasticsearch. Redis peut être utilisé pour stocker temporairement les résultats des requêtes, de sorte que si la même requête est effectuée à plusieurs reprises, la réponse peut être obtenue directement à partir du cache, réduisant ainsi la latence.

Voici comment intégrer Redis dans FastAPI :

1. Connexion à Redis : Tout d'abord, installez le client Redis pour Python :

   bash
   Copy code
   pip install redis
   
   

   Ensuite, définissez le client Redis dans votre fichier de connexion à la base de données (par exemple, db_connection.py).

   python
   Copy code
   from redis import asyncio as aioredis
   redis_client = aioredis.from_url("redis://localhost")
   

2. Vérification de la connexion Redis : Vous pouvez ajouter une fonction pour vérifier la connexion à Redis lors du démarrage de l'application.

   python
   Copy code
   async def ping_redis_server():
   
       try:
           await redis_client.ping()
           logger.info("Connected to Redis")
       except Exception as e:
           logger.error(f"Error connecting to Redis: {e}")
           raise e
   

3. Utilisation de Redis pour le cache : Une fois Redis intégré, vous pouvez l'utiliser pour mettre en cache les résultats des requêtes. Par exemple, le cache peut être utilisé pour mémoriser les résultats des requêtes de recherche des artistes les plus populaires.

   Voici un exemple pour ajouter le cache à l'endpoint qui récupère les artistes :

   python
   Copy code
   @router.get("/top/ten/artists/{country}")
   
   async def top_ten_artist_by_country(country: str, es_client=Depends(get_elasticsearch_client)):
       cache_key = f"top_ten_artists_{country}"
       cached_result = await redis_client.get(cache_key)
       if cached_result:
           return cached_result  # Renvoie les résultats du cache
       try:
           response = await es_client.search(*top_ten_songs_query(country))
       except BadRequestError as e:
           logger.error(e)
           raise HTTPException(status_code=400, detail="Invalid country")
       result = [
           {"artist": record.get("key"), "views": record.get("views", {}).get("value")}
   
           for record in response["aggregations"]["top_ten_artists"]["buckets"]
   
       ]
       # Mémoriser les résultats dans Redis
       await redis_client.set(cache_key, result, ex=3600)  # Expiration du cache après 1 heure
       return result
   

Avec ce système, si une requête pour le même pays est effectuée plusieurs fois, les résultats seront d'abord récupérés à partir du cache Redis, ce qui réduit le temps de réponse et diminue la charge sur Elasticsearch.

Conclusion

L'intégration d'Elasticsearch et de Redis dans une application FastAPI offre de nombreux avantages en termes de performances et de scalabilité. Elasticsearch permet des recherches rapides et puissantes sur de grandes quantités de données, tandis que Redis améliore les performances des requêtes en servant de cache. Ces deux technologies sont des outils indispensables pour toute application qui nécessite des recherches complexes et une réponse rapide.

Comment intégrer FastAPI avec LangChain pour créer un assistant IA puissant

L'intégration de FastAPI avec LangChain permet de créer des applications puissantes et performantes en utilisant des modèles d'IA pour traiter les requêtes et générer des réponses contextuelles à partir de documents. Cette approche est particulièrement utile pour les assistants intelligents qui répondent à des questions basées sur une base de connaissances spécifique, comme des documents ou des FAQ.

Le processus commence par la création d'un module documents.py qui contient la fonction auxiliaire load_documents. Cette fonction est responsable du chargement des fichiers dans une variable. À partir de là, des bibliothèques comme langchain.text_splitter et langchain_community.document_loaders sont utilisées pour organiser les documents en morceaux plus petits, appelés "chunks". Ces morceaux sont ensuite vectorisés à l'aide d'une base de données vectorielle, comme Chroma, qui permet de rechercher des informations pertinentes en fonction d'une requête utilisateur.

L'une des étapes cruciales dans ce processus est la transformation des documents en représentations numériques appelées embeddings. Grâce à Chroma, une base de données vectorielle native à l'IA, chaque document est transformé en un vecteur de caractéristiques. Ces vecteurs sont ensuite stockés et utilisés lors d'une recherche de similarité entre la requête de l'utilisateur et les "chunks" de documents. Cette recherche permet de récupérer le contexte nécessaire pour alimenter un modèle d'IA, comme Cohere, et ainsi fournir une réponse précise à la question posée.

Le code suivant décrit la fonction load_documents qui charge les fichiers .txt depuis un répertoire donné et les divise en morceaux de texte de 100 caractères :

python
Copy code
from langchain.text_splitter import CharacterTextSplitter

from langchain_community.document_loaders import DirectoryLoader
from langchain_community.vectorstores import Chroma
async def load_documents(db: Chroma):
    text_splitter = CharacterTextSplitter(chunk_size=100, chunk_overlap=0)
    raw_documents = DirectoryLoader("docs", "*.txt").load()
    chunks = text_splitter.split_documents(raw_documents)
    await db.aadd_documents(chunks)

python
Copy code
def get_context(user_query: str, db: Chroma) -> str:
    docs = db.similarity_search(user_query)

    return "\n\n".join(doc.page_content for doc in docs)

Cette récupération du contexte est essentielle pour générer une réponse de qualité, car elle permet de limiter la portée de la réponse du modèle en ne l'alimentant qu'avec les informations pertinentes extraites des documents. Une fois le contexte récupéré, il peut être intégré dans le modèle de chaîne qui va traiter la question et renvoyer une réponse.

python
Copy code
from langchain_cohere import ChatCohere
from langchain.schema import StrOutputParser
from prompting import chat_prompt_template
model = ChatCohere(model="command-r-plus")
chain = chat_prompt_template | model | StrOutputParser()

Pour optimiser le processus de chargement des documents, une approche de gestion du cycle de vie (lifespan) est utilisée pour exécuter ce chargement uniquement au démarrage du serveur. Cette fonction garantit que le chargement des documents est effectué une seule fois lorsque le serveur est initialisé, ce qui permet d'éviter une surcharge CPU à chaque requête :

python
Copy code
from contextlib import asynccontextmanager

from fastapi import FastAPI
from langchain_community.vectorstores import Chroma
from documents import load_documents
@asynccontextmanager
async def lifespan(app: FastAPI):
    db = Chroma(embedding_function=CohereEmbeddings())
    await load_documents(db)
    yield {"db": db}
app = FastAPI(title="Ecotech AI Assistant", lifespan=lifespan)

Une fois le serveur démarré, vous pouvez tester l'API via la documentation interactive fournie par FastAPI à l'adresse http://localhost:8000/docs. Cela permet de soumettre des questions à l'API et de recevoir des réponses générées par le modèle, fondées sur les informations contenues dans les documents chargés.

Dans la mise en œuvre d'un tel système, l'un des éléments cruciaux à prendre en compte est la gestion des embeddings. Ces représentations vectorielles des documents sont au cœur du fonctionnement de la recherche de similarité et de la génération des réponses. Il est essentiel de garantir que l'indexation des documents et leur mise à jour sont effectuées efficacement afin d'éviter tout ralentissement dans les performances du système.

En outre, l'optimisation de la recherche de similarité et l'indexation des nouveaux documents sont des tâches qui nécessitent un soin particulier pour maintenir la précision et la rapidité du système d'IA. L'utilisation d'une base de données vectorielle comme Chroma est un excellent choix, mais elle doit être adaptée en fonction des besoins spécifiques de l'application.

Comment organiser et structurer votre projet FastAPI pour un développement optimal

La personnalisation de votre environnement de développement intégré (IDE) peut transformer l'expérience de codage. En ajustant les thèmes, les raccourcis clavier et les paramètres, vous adaptez l'outil à votre flux de travail personnel, augmentant ainsi à la fois votre confort et votre productivité. Une configuration bien pensée de l'IDE devient donc un atout fondamental pour une expérience de développement agréable et fluide.

L'un des aspects essentiels de tout projet logiciel est la gestion des versions. Git, associé à GitHub, constitue un duo puissant pour suivre les changements, collaborer et maintenir l'historique de vos projets. Avant de commencer, vous devez installer Git sur votre système via le site officiel git-scm.com. Ensuite, configurez Git avec votre nom d'utilisateur et votre email afin que les commits soient correctement identifiés.

Une fois Git installé, il est temps de créer votre premier projet FastAPI. Pour ce faire, il est crucial de structurer correctement votre projet dès le début. Une organisation appropriée de votre projet ne se résume pas à un code propre et bien ordonné ; elle garantit aussi la scalabilité et la maintenabilité du code au fur et à mesure que l’application évolue. Cela devient d'autant plus important lorsque le projet se développe et que de nombreuses personnes collaborent dessus.

Une fois l’environnement virtuel activé, vous pouvez installer les paquets nécessaires à votre projet, tels que FastAPI et Uvicorn. Uvicorn, étant un serveur ASGI, vous permettra de faire tourner votre application FastAPI en local. Après avoir installé ces dépendances, il est temps de coder votre première application en FastAPI. Dans un fichier appelé "main.py", vous commencez par importer le module FastAPI et instancier une application. Ensuite, vous définissez votre première route qui renverra une simple réponse "Hello World". Ce premier code constitue la base sur laquelle vous pourrez ajouter des fonctionnalités au fur et à mesure de l’avancement du projet.

Cependant, la structure de votre projet ne doit pas se limiter à quelques fichiers de base. L'organisation du code doit permettre à l'application de grandir tout en restant compréhensible et maintenable. Une approche courante consiste à diviser votre projet en plusieurs répertoires clés :

• /src : Contient le code principal de l’application. À l’intérieur de ce répertoire, vous pouvez avoir des sous-répertoires pour les différents modules de votre application, tels que les modèles de base de données, les routes FastAPI et la logique métier.

• /tests : Le répertoire des tests, séparé du code source, pour éviter que le code de test ne soit inclus dans les versions de production.

• /docs : La documentation, qu'il s'agisse de la documentation de l'API, des guides d'installation ou des instructions d'utilisation, est cruciale pour une bonne gestion du projet. Un dossier dédié à la documentation permet de centraliser ces informations de manière claire et structurée.

Un autre élément essentiel à prendre en compte dans tout projet FastAPI est la gestion des environnements virtuels. Ces environnements permettent de séparer les dépendances du projet de celles d’autres projets ou des bibliothèques système. Pour ce faire, vous pouvez utiliser la bibliothèque venv, qui crée un répertoire contenant un environnement Python isolé.

FastAPI se distingue par sa rapidité et sa facilité d'utilisation. C'est l'un des frameworks les plus performants pour construire des API en Python, grâce à son moteur sous-jacent Starlette et à Pydantic pour la gestion des données. FastAPI facilite également la création de documentation automatiquement générée à partir des annotations de types Python. Cela simplifie la tâche des développeurs et réduit les erreurs tout en accélérant le développement.

L'un des points forts de FastAPI est la prise en charge de la programmation asynchrone. Cette approche permet à votre application de gérer plusieurs requêtes simultanément, améliorant ainsi l'efficacité de l'application et sa réactivité. L'intégration de la programmation asynchrone avec FastAPI se fait via la syntaxe async/await, qui permet à votre code de traiter des tâches simultanées sans les bloquer. Cette fonctionnalité est particulièrement utile dans les applications nécessitant de gérer des tâches I/O, comme les appels API ou les requêtes de base de données.

Il est important de comprendre que la mise en place d’un projet avec FastAPI ne se limite pas à l’installation et à la configuration. Pour obtenir une application scalable et maintenable, vous devez également adopter des pratiques comme la séparation claire des responsabilités (logiciel, base de données, documentation, tests), l’utilisation systématique de la programmation asynchrone pour améliorer les performances, et l’écriture de tests unitaires pour garantir la fiabilité de l’application.