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.

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

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.

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.

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

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

db.py

models.py

schemas.py

router.py

dbutils.py

main.py