FastApi nasce come framework per lo sviluppo di applicazioni di API in ambiente Python. Il suo principale punto di forza di basa sullo sfruttamento dei meccanismi di esecuzione asincrona basata su coroutines messi a disposizione dalla versione 3.6 del linguaggio. In questo articolo vedremo come creare una semplice API CRUD (Create, Read, Update, Delete) utilizzando l'ecosistema di FastApi.
Nel caso specifico, andremo a creare un'API per la gestione di promemoria, ciascuno caratterizzato da un titolo e una descrizione.
Iniziamo!
Per il nostro progetto avremo bisogno di alcune dipendenze:
Iniziamo creando una cartella per il progetto e un ambiente virtuale Python in cui installare le dipendenze
# Creazione della cartella
$ mkdir fastapi-items
$ cd fastapi-items
# Inizializzazione dell'ambiente virtuale
$ python3 -m venv env
# Avvio dell'ambiente virtuale
$ source env/bin/activate
Installiamo ora le dipendenze necessarie
(env) $ pip install fastapi "uvicorn[standard]" sqlalchemy
Creare la seguente struttura di file e cartelle
.
└── env
├── items_app
├── __init__.py
├── crud.py
├── database.py
├── main.py
├── models.py
└── schemas.py
Questo lo scopo dei diversi file
Nel nostro esempio verrà utilizzato un database SQLite, ma l'approccio è generalizzabile con modifiche minime a qualsiasi DBMS.
# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Gli elementi definiti in questo file sono
Nel nostro caso, andremo a definire un solo modello di dati, con i seguenti campi
Utilizzando gli elementi forniti da SQLAlchemy, la definizione del nostro modello dati è la seguente
# models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String)
Attenzione al fatto che la classe
Item
estende la classeBase
che abbiamo creato precedentemente nel filedatabase.py
. Questo è necessario per far si che l'ORM funzioni correttamente e che le tabelle vengano create nel database all'avvio dell'applicazione
La natura delle diverse operazioni esposte dall'API richiede che alcuni attributi di un oggetto siano utilizzati in precisi contesti piuttosto che in altri. Ad esempio, l'attributo id
che identifica un oggetto solitamente viene generato dal database, e non è quindi possibile per il client inviarlo al server nell'operazione di creazione di un oggetto.
Nel nostro esempio, definiremo tre schemi, ovvero tre forme di un oggetto Item
, per tre usi diversi:
Item
che un client chiede di creareItem
che un client chiede di aggiornare# schemas.py
from typing import Optional
from pydantic import BaseModel
class ItemCreate(ItemBase):
title: str
description: Optional[str] = None
class ItemUpdate(ItemBase):
title: str
description: Optional[str] = None
class Item(ItemBase):
id: int
title: str
description: Optional[str] = None
class Config:
orm_mode = True
Lo schema Item
è configurato per operare in orm_mode
: questo configura pydantic
e permette un'integrazione diretta con SQLAlchemy
Questi "schemi" hanno un ruolo parzialmente equivalente a quello di un
serializer
indjango-rest-framework
Andiamo ora a definire le operazioni fondamentali di un modello CRUD, ovvero
L'implementazione tiene conto del fatto che sia l'utente dell'API ad utilizzarle: i parametri delle funzioni modellano gli oggetti con schemi
(ad esempio ItemUpdate
o ItemCreate
), mentre i valori di ritorno sono classi del modello dati (Item
definita nel file models.py
)
Il primo parametro di ciascuna funzione (db
) rappresenta la connessione al database, ed è necessario per le diverse operazioni. Gli altri parametri variano caso per caso. Ad esempio l'operazione di create richiede come parametro i dati per creare la nuova istanza di Item
.
# crud.py
from sqlalchemy.orm import Session
from .models import Item
from .schemas import ItemCreate, ItemUpdate
from typing import Union
def list_items(db: Session, skip: int = 0, limit: int = 100):
return db.query(Item).offset(skip).limit(limit).all()
def get_item(db: Session, id: int):
return db.query(Item).get(id)
def create_item(db: Session, data: ItemCreate):
db_item = Item(**data.dict())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
def drop_item(db: Session, item_id: int):
db.query(Item).filter(Item.id == item_id).delete()
db.commit()
return None
def update_item(db: Session, item: Union[int, Item], data: ItemUpdate):
if isinstance(item, int):
item = get_item(db, item)
if item is None:
return None
for key, value in data:
setattr(item, key, value)
db.commit()
return item
L'implementazione dell'api vera e propria si avvale di tutti gli elementi che abbiamo definito nei punti precedenti:
Questa l'implementazione delle operazioni CRUD sugli oggetti di tipo Item
# main.py
from typing import List, Optional
from fastapi import FastAPI, HTTPException
from fastapi.params import Depends
from sqlalchemy.orm import Session
from . import models, crud, schemas
from .database import engine, SessionLocal
# Creazione automatica delle tabelle nel database
# Le tabelle sono create solo se non già esistenti, quindi questo comando
# può essere eseguito in sicurezza ad ogni avvio dell'applicazione
# Per una gestione migliore delle modifiche strutturali sul database,
# si consiglia di utilizzare uno strumento apposito, come Alembic
models.Base.metadata.create_all(bind=engine)
# Inizializzazione dell'applicazione
app = FastAPI()
# Questa funzione rappresenta una dipendenza dell'api
# FastApi contiene un meccanismo di dependency injection, che permette a ciascun
# endpoint di richiedere i servizi necessari in modo dichiarativo. Questa funzione
# modella la connessione al db come un servizio su richiesta, inizializzato solo
# se realmente necessario
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# LIST
# Questo endpoint ritorna una lista di oggetti di tipo `Item` in base allo schema `Item` che
# è stato definito in schemas.py. Gli oggetti ritornati sono istanze di `models.Item` che vengono
# validati e serializzati in base alla definizione dello schema `schemas.Item`
@app.get("/items", response_model=List[schemas.Item])
def items_action_list(limit: int = 100, offset: int = 0, db: Session = Depends(get_db)):
items = crud.list_items(db, offset, limit)
return items
# RETRIEVE
# Questo endpoint ritorna uno spefico `Item`, noto il valore del campo `id` che viene passato
# come parametro nel path. Può ritornare anche una condizione di errore in caso in cui l'id
# richiesto non corrisponda a nessun oggetto
@app.get("/items/{item_id}", response_model=schemas.Item)
def items_action_retrieve(item_id: int, db: Session = Depends(get_db)):
item = crud.get_item(db, item_id)
if item is None:
raise HTTPException(status_code=404)
return item
# CREATE
# Questo endpoint permette di creare un nuovo `Item`. I dati necessari sono letti dal corpo della
# richiesta e validati in base allo schema `schemas.ItemCreate` che noi abbiamo definito sopra
@app.post("/items", response_model=schemas.Item)
def item_action_create(data: schemas.ItemCreate, db: Session = Depends(get_db)):
item = crud.create_item(db, data)
return item
# UPDATE
# Questo endpoint permette di aggiornare un `Item` esistente, identificato dalla chiave primaria
# passata come parametro nel path. I dati necessari all'aggiornamento sono letti dal corpo della
# richiesta e validati in base allo schema `schemas.ItemUpdate` che noi abbiamo definito sopra
@app.put("/items/{item_id}", response_model=schemas.Item)
def items_action_retrieve(item_id: int, data: schemas.ItemUpdate, db: Session = Depends(get_db)):
item = crud.update_item(db, item_id, data)
if item is None:
raise HTTPException(status_code=404)
return item
# DELETE
# Questo endpoint permette di eliminare un item, identificato dalla chiave primaria inserita
# nel path della richiesta. Da notare che in caso di successo il codice di stato della risposta
# è HTTP 204 No Content, in quanto il corpo della risposta è vuoto
@app.delete("/items/{item_id}", status_code=204)
def items_action_retrieve(item_id: int, db: Session = Depends(get_db)):
crud.drop_item(db, item_id)
return None
Per eseguire la nostra API è sufficiente il seguente comando, eseguito nella directory radice del progetto
(env) $ uvicorn items_app.main:app --reload
A questo punto l'api è in ascolto sulla porta 8000, ed è possibile interagirvi con qualsiasi tool per il testing di API Rest, come Postman o Insomnia
I contenuti di questo articolo sono ispirati dal tutorial ufficiale di FastApi. Agli interessati consigliamo vivamente la lettura della documentazione, data la sua completezza e facilità di lettura
Rispetto ai framework Python tradizionali come Django o Flask, FastApi utilizza standard più moderni (ASGI vs WSGI) e rispetto ai benchmark si dimostra molto più performante ed efficiente nell'utilizzo delle risorse. Inoltre, i componenti base di FastApi utilizzano le più moderne caratteristiche del linguaggio, come la tipizzazione, che conferiscono al codice maggiore leggibilità e chiarezza, oltre a migliorare il supporto da parte degli IDE e a semplificare la ricerca di errori
Il codice sorgente completo (ed eseguibile) che abbiamo sviluppato in questo articolo è disponibile su questo repository, e può essere utilizzato come punto di partenza per lo sviluppo di una generica API REST.
Il caso che abbiamo valutato è volutamente semplice, in modo da poter costruire una solida base concettuale su cui poi è possibile introdurre nuovi concetti, come elementi di sicurezza o una gestione più ordinata delle modifiche al modello dati per mezzo di migrazioni. Se l'articolo vi è piaciuto o se vi interessa saperne di più, fatecelo sapere! Ci trovate su facebook e twitter.