Créer une API CRUD à l'aide de l'API rapide

FastAPI est un framework asynchrone pour la création d'API WEB python, ce qui signifie que vous pouvez facilement créer des API hautes performances.

Web Framework Benchmarks

Essayez-le en vous référant à Tutoriel.

1. Structure des fichiers de l'application à créer

Créez un CRUD pour la table des utilisateurs. Je crée environ 6 fichiers, chacun avec une petite quantité de code.

fastapi-crud-example
│  db.py
│  main.py
│  
├─users
│  │  models.py
│  │  router.py
│  └─ schemas.py
│          
└─utils
      dbutils.py
  1. source

(1) db.py

sqlalchemy est utilisé pour définir des modèles et générer des requêtes. Utilisez Databases pour accéder à la base de données.

db.py


import databases
import sqlalchemy

DATABASE = 'postgresql'
USER = 'testuser'
PASSWORD = 'secret'
HOST = 'localhost'
PORT = '5432'
DB_NAME = 'testdb'

DATABASE_URL = '{}://{}:{}@{}:{}/{}'.format(DATABASE, USER, PASSWORD, HOST, PORT, DB_NAME)

# databases
database = databases.Database(DATABASE_URL, min_size=5, max_size=20)

ECHO_LOG = False

engine = sqlalchemy.create_engine(DATABASE_URL, echo=ECHO_LOG)

metadata = sqlalchemy.MetaData()

(2) users/models.py

models.py


import sqlalchemy
from db import metadata, engine

users = sqlalchemy.Table(
    "users",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, index=True),
    sqlalchemy.Column("username", sqlalchemy.String, index=True),
    sqlalchemy.Column("email", sqlalchemy.String, index=True),
    sqlalchemy.Column("hashed_password", sqlalchemy.String),
    sqlalchemy.Column("is_active", sqlalchemy.Boolean(), default=True),
    sqlalchemy.Column("is_superuser", sqlalchemy.Boolean(), default=False)
)

metadata.create_all(bind=engine)

(3) users/schemas.py

Utilisez le modèle dans pydantic pour définir également le modèle pour crud.

schemas.py


from pydantic import BaseModel

#modèle de demande pour l'insert. id(numérotation automatique)N'est pas défini car aucune entrée n'est requise.
class UserCreate(BaseModel):
    username: str
    email: str
    password: str
    is_active: bool
    is_superuser: bool

#modèle de demande de mise à jour
class UserUpdate(BaseModel):
    id : int
    username: str
    email: str
    password: str
    is_active: bool
    is_superuser: bool

#modèle de demande pour sélectionner. Dans select, aucun mot de passe n'est requis, il n'est donc pas défini.
class UserSelect(BaseModel):
    username: str
    email: str
    is_active: bool
    is_superuser: bool

(4) users/router.py

Ce sera la partie principale de crud. Tous les éléments de sélection, d'insertion, de mise à jour et de suppression sont implémentés en moins de 10 lignes. C'est très bien que la quantité de codage soit petite.

router.py


import hashlib

from fastapi import APIRouter, Depends
from typing import List
from starlette.requests import Request

from .models import users
from .schemas import UserCreate, UserUpdate, UserSelect

from databases import Database

from utils.dbutils import get_connection

router = APIRouter()

#Le mot de passe saisi (texte brut) est haché et renvoyé.
def get_users_insert_dict(user):
    pwhash=hashlib.sha256(user.password.encode('utf-8')).hexdigest()
    values=user.dict()
    values.pop("password")
    values["hashed_password"]=pwhash
    return values

#Recherchez tous les utilisateurs et renvoyez la liste de "Sélection d'utilisateur" en tant que json.
@router.get("/users/", response_model=List[UserSelect])
async def users_findall(request: Request, database: Database = Depends(get_connection)):
    query = users.select()
    return await database.fetch_all(query)

#Recherchez les utilisateurs par identifiant et renvoyez "User Select" en tant que json.
@router.get("/users/find", response_model=UserSelect)
async def users_findone(id: int, database: Database = Depends(get_connection)):
    query = users.select().where(users.columns.id==id)
    return await database.fetch_one(query)

#Enregistrez de nouveaux utilisateurs.
@router.post("/users/create", response_model=UserSelect)
async def users_create(user: UserCreate, database: Database = Depends(get_connection)):
    #validateur omis
    query = users.insert()
    values = get_users_insert_dict(user)
    ret = await database.execute(query, values)
    return {**user.dict()}

#Mettre à jour les utilisateurs.
@router.post("/users/update", response_model=UserSelect)
async def users_update(user: UserUpdate, database: Database = Depends(get_connection)):
    #validateur omis
    query = users.update().where(users.columns.id==user.id)
    values=get_users_insert_dict(user)
    ret = await database.execute(query, values)
    return {**user.dict()}

#Supprimer des utilisateurs.
@router.post("/users/delete")
async def users_delete(user: UserUpdate, database: Database = Depends(get_connection)):
    query = users.delete().where(users.columns.id==user.id)
    ret = await database.execute(query)
    return {"result": "delete success"}

(5) utils/dbutils.py

dbutils.py


from starlette.requests import Request

#Connexion stockée dans la demande par le middleware(Objet de base de données)Retour.
def get_connection(request: Request):
    return request.state.connection

(6) main.py

main.py


from fastapi import FastAPI
from db import database
from users.router import router as userrouter
from starlette.requests import Request

app = FastAPI()

#Connectez-vous à la base de données au démarrage.
@app.on_event("startup")
async def startup():
    await database.connect()

#Déconnectez la base de données lorsque vous avez terminé.
@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

#Enregistrez le routeur des utilisateurs.
app.include_router(userrouter)

# middleware state.Définissez l'objet de base de données en connexion.
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    request.state.connection = database
    response = await call_next(request)
    return response

3. Démarrez et confirmez

Entrez "uvicorn main: app --reload" à partir du shell d'alimentation pour entrer

uvicorn main:app --reload
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [14108]
INFO: Started server process [21068]
INFO: Waiting for application startup.
INFO: Connected to database postgresql://testuser:********@localhost:5432/fastapidb

4. Vérifiez avec Swagger

Vous pouvez tester l'API avec Swagger UI. C'est pratique.

Essayez d'accéder à http://127.0.0.1:8000/docs avec chrome.

image.png

5. Enfin

J'ai eu l'impression que FastAPI est un travail freame spécialisé pour api, mais vous pouvez également utiliser un moteur de template utilisant jinja2 etc., et il a aussi une fonction d'authentification telle que oauth2. FastAPI Ça fait du bien.

Recommended Posts

Créer une API CRUD à l'aide de l'API rapide
Créons une API REST en utilisant SpringBoot + MongoDB
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 1 ~
Créer un pseudo serveur d'API REST à l'aide de pages GitHub
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 2 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 3 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 4 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 5 ~
Créer une interface graphique python à l'aide de tkinter
Créer un dictionnaire imbriqué à l'aide de defaultdict
Essayez d'utiliser l'API de Pleasant (python / FastAPI)
Créez un robot de réponse automatique en temps réel à l'aide de l'API Twitter Streaming
Créez une API en utilisant hug avec mod_wsgi
Créez une application CRUD simple en utilisant la vue de classe générique de Django
Créez un wrapper de langage C à l'aide de Boost.Python
Créer une API qui renvoie les données d'un modèle à l'aide de turicreate
Créez facilement un TalkBot en utilisant Discord.py et l'API Talk d'A3RT (pya3rt).
Créer un graphique à l'aide du module Sympy
Créer une application à l'aide de l'API Spotify
Créer un bloc de données à partir d'Excel à l'aide de pandas
Créer un fichier GIF en utilisant Pillow en Python
Créez facilement un serveur API à l'aide du module go-json-rest
Créer un arbre phylogénétique à partir de Biopyton en utilisant ClustalW2
Créer une carte Web en utilisant Python et GDAL
Créer un système de notification des visiteurs à l'aide de Raspberry Pi
Comment créer une API Rest dans Django
Créez un fichier MIDI en Python en utilisant pretty_midi
Préparer un pseudo serveur API à l'aide d'actions GitHub
Créez une interface graphique sur le terminal à l'aide de curses
Créez une API REST à l'aide du modèle appris dans Lobe et TensorFlow Serving.
Créez une base de données propre pour les tests avec FastAPI et effectuez le test Unittest de l'API avec pytest
Créer un planning Django
Créer un bot de collecte de données en Python à l'aide de Selenium
Créez un capteur de couleur à l'aide d'une tarte à la râpe et d'une caméra
Essayez de dessiner un graphe social à l'aide de l'API Twitter v2
[API LINE Messaging] Créez un menu riche avec Python
[Python] Générer ValueObject avec un constructeur complet à l'aide de classes de données
Construire un modèle seq2seq en utilisant la vue d'ensemble de l'API fonctionnelle de Keras
Enregistrez des tickets avec l'API de Redmine en utilisant des requêtes Python
Créez une carte thermique de tweet avec l'API Google Maps
Apprendre un réseau neuronal à l'aide de Chainer
Créer un LV amorçable
Créer un environnement Python
Rechercher la table à l'aide de sqlalchemy et créer un dictionnaire
Créer un environnement de développement d'API rapide léger à l'aide de Docker
Créer un bot slack
J'ai essayé de créer une API RESTful en connectant l'API rapide du framework Python explosif à MySQL.
Créez instantanément un diagramme de données 2D à l'aide de matplotlib de python
Je vais créer un jeu pour contrôler le puzzle et les dragons en utilisant pygame
Construire un modèle seq2seq en utilisant la création et l'apprentissage du modèle d'API fonctionnelle de Keras
Créez un serveur Web API à une vitesse explosive en utilisant HUG
Tornado - Créons une API Web qui renvoie facilement JSON avec JSON
Créez une API Web capable de fournir des images avec Django
Créer une instance GCE à partir d'une image Docker GCR à l'aide de terraform
Créez rapidement une API avec Python, lambda et API Gateway à l'aide d'AWS SAM
Utilisez wxpython pour créer une fenêtre de type d'image d'arrière-plan transparent semblable à un gadget
L'histoire de la création d'une base de données à l'aide de l'API Google Analytics
J'ai créé Chatbot en utilisant l'API LINE Messaging et Python