FastAPI ist ein asynchrones Framework zum Erstellen von Python-WEB-APIs. Dies bedeutet, dass Sie problemlos Hochleistungs-APIs erstellen können.

Versuchen Sie es unter Tutorial.

1. Dateistruktur der zu erstellenden App

Erstellen Sie eine CRUD für die Benutzertabelle. Ich erstelle ungefähr 6 Dateien mit jeweils einer kleinen Menge Code.

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

source

(1) db.py

sqlalchemy wird verwendet, um Modelle zu definieren und Abfragen zu generieren. Verwenden Sie Datenbanken, um auf die Datenbank zuzugreifen.

`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

Verwenden Sie das Modell in pydantic, um das Modell auch für Rohöl zu definieren.

`schemas.py`


from pydantic import BaseModel

#Modell zum Einfügen anfordern. Ich würde(automatische Nummerierung)Ist nicht definiert, da keine Eingabe erforderlich ist.
class UserCreate(BaseModel):
    username: str
    email: str
    password: str
    is_active: bool
    is_superuser: bool

#Modell zur Aktualisierung anfordern
class UserUpdate(BaseModel):
    id : int
    username: str
    email: str
    password: str
    is_active: bool
    is_superuser: bool

#Modell zur Auswahl anfordern. In select ist kein Passwort erforderlich, daher ist es nicht definiert.
class UserSelect(BaseModel):
    username: str
    email: str
    is_active: bool
    is_superuser: bool

(4) users/router.py

Es wird der Hauptteil von Rohöl sein. Alle Auswahl-, Einfüge-, Aktualisierungs- und Löschvorgänge werden in weniger als 10 Zeilen implementiert. Es ist sehr schön, dass der Codierungsaufwand gering ist.

`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()

#Das eingegebene Passwort (Klartext) wird gehasht und zurückgegeben.
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

#Durchsuchen Sie alle Benutzer und geben Sie die Liste "User Select" als json zurück.
@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)

#Suchen Sie Benutzer nach ID und geben Sie "User Select" als json zurück.
@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)

#Registrieren Sie neue Benutzer.
@router.post("/users/create", response_model=UserSelect)
async def users_create(user: UserCreate, database: Database = Depends(get_connection)):
    #Validator weggelassen
    query = users.insert()
    values = get_users_insert_dict(user)
    ret = await database.execute(query, values)
    return {**user.dict()}

#Benutzer aktualisieren.
@router.post("/users/update", response_model=UserSelect)
async def users_update(user: UserUpdate, database: Database = Depends(get_connection)):
    #Validator weggelassen
    query = users.update().where(users.columns.id==user.id)
    values=get_users_insert_dict(user)
    ret = await database.execute(query, values)
    return {**user.dict()}

#Benutzer löschen.
@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

#Verbindung auf Anfrage von Middleware gespeichert(Datenbankobjekt)Kehrt zurück.
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()

#Stellen Sie beim Start eine Verbindung zur Datenbank her.
@app.on_event("startup")
async def startup():
    await database.connect()

#Trennen Sie die Datenbank, wenn Sie fertig sind.
@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

#Registrieren Sie den Router des Benutzers.
app.include_router(userrouter)

# middleware state.Legen Sie das Datenbankobjekt in Verbindung.
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    request.state.connection = database
    response = await call_next(request)
    return response

3. Starten und bestätigen Sie

Geben Sie "uvicorn main: app --reload" von der Power Shell ein, um einzutreten

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. Fragen Sie bei Swagger nach

Sie können die API mit der Swagger-Benutzeroberfläche testen. Es ist bequem.

Versuchen Sie, mit Chrome auf http://127.0.0.1:8000/docs zuzugreifen.

5. Schließlich

Ich hatte den Eindruck, dass FastAPI eine auf API spezialisierte Freame-Arbeit ist, aber Sie können auch eine Template-Engine mit jinja2 usw. verwenden und sie hat auch eine Authentifizierungsfunktion wie oauth2. FastAPI Es fühlt sich gut an.

[PYTHON] Erstellen Sie die CRUD-API mit der Fast API

1. Dateistruktur der zu erstellenden App

db.py

models.py

schemas.py

router.py

dbutils.py

main.py