App ad alte prestazioni con Python: un tutorial FastAPI

Pubblicato: 2022-03-11

I buoni framework del linguaggio di programmazione semplificano la produzione di prodotti di qualità più velocemente. Ottimi framework rendono anche piacevole l'intera esperienza di sviluppo. FastAPI è un nuovo framework Web Python potente e divertente da usare. Le seguenti funzionalità rendono FastAPI da provare:

  • Velocità: FastAPI è uno dei framework Web Python più veloci. In effetti, la sua velocità è alla pari con Node.js e Go. Controlla questi test delle prestazioni.
  • Documenti per sviluppatori dettagliati e facili da usare
  • Digita un suggerimento per il tuo codice e ottieni la convalida e la conversione dei dati gratuite.
  • Crea facilmente plugin usando l'iniezione di dipendenza.

Costruire un'app TODO

Per esplorare le grandi idee alla base di FastAPI, creiamo un'app TODO, che imposta elenchi di cose da fare per i suoi utenti. La nostra piccola app fornirà le seguenti funzionalità:

  • Iscriviti e accedi
  • Aggiungi nuovo elemento TODO
  • Ottieni un elenco di tutti i TODO
  • Elimina/aggiorna un elemento TODO

SQLAlchemy per i modelli di dati

La nostra app ha solo due modelli: User e TODO. Con l'aiuto di SQLAlchemy, il toolkit di database per Python, possiamo esprimere i nostri modelli in questo modo:

 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan") class TODO(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="todos")

Una volta che i nostri modelli sono pronti, scriviamo il file di configurazione per SQLAlchemy in modo che sappia come stabilire una connessione con il database.

 import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Scatena il potere dei suggerimenti di tipo

Una parte considerevole di qualsiasi progetto API riguarda le cose di routine come la convalida e la conversione dei dati. Affrontiamolo in anticipo prima di passare alla scrittura dei gestori delle richieste. Con FastAPI, esprimiamo lo schema dei nostri dati in entrata/uscita utilizzando modelli pydantic e quindi utilizziamo questi modelli pydantic per digitare hint e usufruire gratuitamente della convalida e della conversione dei dati. Tieni presente che questi modelli non sono correlati al flusso di lavoro del nostro database e specificano solo la forma dei dati che fluiscono in entrata e in uscita dalla nostra interfaccia REST. Per scrivere modelli avventurosi, pensa a tutti i modi in cui le informazioni sull'utente e su TODO fluiranno dentro e fuori.

Tradizionalmente, un nuovo utente si iscriverà al nostro servizio TODO e un utente esistente effettuerà l'accesso. Entrambe queste interazioni riguardano le informazioni dell'utente, ma la forma dei dati sarà diversa. Abbiamo bisogno di più informazioni dagli utenti durante la registrazione e minime (solo e-mail e password) durante l'accesso. Ciò significa che abbiamo bisogno di due modelli pydantic per esprimere queste due diverse forme di informazioni sull'utente.

Nella nostra app TODO, tuttavia, sfrutteremo il supporto OAuth2 integrato in FastAPI per un flusso di accesso basato su JSON Web Tokens (JWT). Dobbiamo solo definire qui uno schema UserCreate per specificare i dati che fluiranno nel nostro endpoint di registrazione e uno schema UserBase da restituire come risposta nel caso in cui il processo di registrazione abbia esito positivo.

 from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

Qui, abbiamo contrassegnato il cognome, il nome e la password come una stringa, ma può essere ulteriormente rafforzata utilizzando stringhe vincolate pydantic che abilitano controlli come lunghezza minima, lunghezza massima e regex.

Per supportare la creazione e l'elenco degli articoli TODO, definiamo il seguente schema:

 class TODOCreate(BaseModel): text: str completed: bool

Per supportare l'aggiornamento di un elemento TODO esistente, definiamo un altro schema:

 class TODOUpdate(TODOCreate): id: int

Con questo, abbiamo finito con la definizione degli schemi per tutti gli scambi di dati. Ora rivolgiamo la nostra attenzione ai gestori delle richieste in cui questi schemi verranno utilizzati per eseguire gratuitamente tutte le operazioni pesanti di conversione e convalida dei dati.

Consenti agli utenti di registrarsi

Innanzitutto, consentiamo agli utenti di registrarsi, poiché tutti i nostri servizi devono essere accessibili da un utente autenticato. Scriviamo il nostro primo gestore di richieste utilizzando lo schema UserCreate e UserBase definito sopra.

 @app.post("/api/users", response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): """add new user""" user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail="Email already registered.") signedup_user = crud.create_user(db, user_data) return signedup_user

C'è molto da fare in questo breve pezzo di codice. Abbiamo utilizzato un decoratore per specificare il verbo HTTP, l'URI e lo schema delle risposte riuscite. Per garantire che l'utente abbia inviato i dati corretti, abbiamo digitato hint il corpo della richiesta con uno schema UserCreate definito in precedenza. Il metodo definisce un altro parametro per ottenere un handle sul database: questa è l'iniezione di dipendenza in azione ed è discussa più avanti in questo tutorial.

Mettere in sicurezza la nostra API

Vogliamo le seguenti funzionalità di sicurezza nella nostra app:

  • Hashing delle password
  • Autenticazione basata su JWT

Per l'hashing delle password, possiamo usare Passlib. Definiamo le funzioni che gestiscono l'hashing delle password e controllano se una password è corretta.

 from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

Per abilitare l'autenticazione basata su JWT, è necessario generare JWT e decodificarli per ottenere le credenziali dell'utente. Definiamo le seguenti funzioni per fornire questa funzionalità.

 # install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

Emettere token in caso di accesso riuscito

Ora definiremo un endpoint di accesso e implementeremo il flusso di password OAuth2. Questo endpoint riceverà un'e-mail e una password. Verificheremo le credenziali rispetto al database e, in caso di successo, rilasceremo un token Web JSON all'utente.

Per ricevere le credenziali, utilizzeremo OAuth2PasswordRequestForm , che fa parte delle utilità di sicurezza di FastAPI.

 @app.post("/api/token", response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): """generate access token for valid credentials""" user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={"sub": user.email}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer"}

Utilizzo dell'inserimento delle dipendenze per accedere a DB e proteggere gli endpoint

Abbiamo impostato l'endpoint di accesso che fornisce un JWT a un utente dopo l'accesso riuscito. L'utente può salvare questo token nella memoria locale e mostrarlo al nostro back-end come intestazione di autorizzazione. Gli endpoint che prevedono l'accesso solo dagli utenti che hanno effettuato l'accesso possono decodificare il token e scoprire chi è il richiedente. Questo tipo di lavoro non è legato a un particolare endpoint, ma è una logica condivisa utilizzata in tutti gli endpoint protetti. È meglio impostare la logica di decodifica dei token come una dipendenza che può essere utilizzata in qualsiasi gestore di richieste.

In linguaggio FastAPI, le nostre funzioni di operazione sul percorso (gestori di richieste) dipenderebbero quindi da get_current_user . La dipendenza get_current_user deve avere una connessione al database e collegarsi alla logica OAuth2PasswordBearer di OAuth2PasswordBearer per ottenere un token. Risolveremo questo problema facendo in modo che get_current_user dipenda da altre funzioni. In questo modo, possiamo definire catene di dipendenza, che è un concetto molto potente.

 def get_db(): """provide db session to path operation functions""" try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get("/api/me", response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): """return user settings for current user""" return current_user

Gli utenti che hanno effettuato l'accesso possono CRUD TODOs

Prima di scrivere le funzioni dell'operazione di percorso per TODO Create, Read, Update, Delete (CRUD), definiamo le seguenti funzioni di supporto per eseguire CRUD effettivo sul db.

 def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

Queste funzioni a livello di database verranno utilizzate nei seguenti endpoint REST:

 @app.get("/api/mytodos", response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """return a list of TODOs owned by current user""" todos = crud.get_user_todos(db, current_user.id) return todos @app.post("/api/todos", response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """add a TODO""" todo = crud.create_meal(db, current_user, meal_data) return todo @app.put("/api/todos/{todo_id}", response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """update and return TODO for given id""" todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete("/api/todos/{todo_id}") def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """delete TODO of given id""" crud.delete_meal(db, todo_id) return {"detail": "TODO Deleted"}

Scrivi test

Scriviamo alcuni test per la nostra API TODO. FastAPI fornisce una classe TestClient basata sulla popolare libreria Requests e possiamo eseguire i test con Pytest.

Per assicurarci che solo gli utenti che hanno effettuato l'accesso possano creare un TODO, possiamo scrivere qualcosa del genere:

 from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False) response = client.post("/api/todos", data=todo) assert response.status_code == 401

Il test seguente verifica il nostro endpoint di accesso e genera un JWT se presentato con credenziali di accesso valide.

 def test_user_can_obtain_auth_token(): response = client.post("/api/token", data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

Riassumendo

Abbiamo finito di implementare un'app TODO molto semplice utilizzando FastAPI. A questo punto, hai visto la potenza dei suggerimenti sui tipi messa a frutto per definire la forma dei dati in entrata e in uscita attraverso la nostra interfaccia REST. Definiamo gli schemi in un punto e lasciamo a FastAPI l'applicazione della convalida e della conversione dei dati. L'altra caratteristica degna di nota è l'iniezione di dipendenza. Abbiamo utilizzato questo concetto per impacchettare la logica condivisa per ottenere una connessione al database, decodificare il JWT per ottenere l'utente attualmente connesso e implementare semplice OAuth2 con password e bearer. Abbiamo anche visto come le dipendenze possono essere concatenate.

Possiamo facilmente applicare questo concetto per aggiungere funzionalità come l'accesso basato sui ruoli. Inoltre, stiamo scrivendo un codice conciso e potente senza imparare le peculiarità di un framework. In parole semplici, FastAPI è una raccolta di potenti strumenti che non devi imparare perché sono solo moderni Python. Divertiti.