Aplicaciones de alto rendimiento con Python: un tutorial de FastAPI

Publicado: 2022-03-11

Los buenos marcos de lenguaje de programación facilitan la producción de productos de calidad más rápido. Los grandes marcos incluso hacen que toda la experiencia de desarrollo sea agradable. FastAPI es un nuevo marco web de Python que es poderoso y agradable de usar. Las siguientes funciones hacen que valga la pena probar FastAPI:

  • Velocidad: FastAPI es uno de los marcos web de Python más rápidos. De hecho, su velocidad está a la par con Node.js y Go. Compruebe estas pruebas de rendimiento.
  • Documentos para desarrolladores detallados y fáciles de usar
  • Escriba su código y obtenga validación y conversión de datos gratis.
  • Cree complementos fácilmente usando la inyección de dependencia.

Creación de una aplicación TODO

Para explorar las grandes ideas detrás de FastAPI, construyamos una aplicación TODO, que configura listas de tareas para sus usuarios. Nuestra pequeña aplicación proporcionará las siguientes características:

  • Regístrese e inicie sesión
  • Agregar nuevo elemento TODO
  • Obtener una lista de todas las TODO
  • Eliminar/Actualizar un elemento TODO

SQLAlchemy para modelos de datos

Nuestra aplicación tiene solo dos modelos: Usuario y TODO. Con la ayuda de SQLAlchemy, el conjunto de herramientas de base de datos para Python, podemos expresar nuestros modelos de esta manera:

 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 vez que nuestros modelos estén listos, escribamos el archivo de configuración para SQLAlchemy para que sepa cómo establecer una conexión con la base de datos.

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

Libere el poder de las sugerencias tipográficas

Una parte considerable de cualquier proyecto de API se refiere a cosas rutinarias como la validación y conversión de datos. Abordémoslo desde el principio antes de escribir controladores de solicitudes. Con FastAPI, expresamos el esquema de nuestros datos entrantes/salientes usando modelos pydantic y luego usamos estos modelos pydantic para escribir sugerencias y disfrutar de validación y conversión de datos gratis. Tenga en cuenta que estos modelos no están relacionados con el flujo de trabajo de nuestra base de datos y solo especifican la forma de los datos que entran y salen de nuestra interfaz REST. Para escribir modelos dinámicos, piense en todas las formas en que la información del Usuario y TODO fluirá hacia adentro y hacia afuera.

Tradicionalmente, un nuevo usuario se registrará en nuestro servicio TODO y un usuario existente iniciará sesión. Ambas interacciones tratan con la información del Usuario, pero la forma de los datos será diferente. Necesitamos más información de los usuarios durante el registro y mínima (solo correo electrónico y contraseña) al iniciar sesión. Esto significa que necesitamos dos modelos dinámicos para expresar estas dos formas diferentes de información del usuario.

En nuestra aplicación TODO, sin embargo, aprovecharemos la compatibilidad integrada con OAuth2 en FastAPI para un flujo de inicio de sesión basado en JSON Web Tokens (JWT). Solo necesitamos definir un esquema UserCreate aquí para especificar los datos que fluirán hacia nuestro punto final de registro y un esquema UserBase para devolver como respuesta en caso de que el proceso de registro sea exitoso.

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

Aquí, marcamos el apellido, el nombre y la contraseña como una cadena, pero se puede ajustar aún más mediante el uso de cadenas con restricciones pydantic que permiten verificaciones como la longitud mínima, la longitud máxima y las expresiones regulares.

Para respaldar la creación y el listado de elementos TODO, definimos el siguiente esquema:

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

Para admitir la actualización de un elemento TODO existente, definimos otro esquema:

 class TODOUpdate(TODOCreate): id: int

Con esto, hemos terminado con la definición de esquemas para todos los intercambios de datos. Ahora dirigimos nuestra atención a los controladores de solicitudes donde estos esquemas se utilizarán para hacer todo el trabajo pesado de conversión y validación de datos de forma gratuita.

Permitir que los usuarios se registren

Primero, permitamos que los usuarios se registren, ya que un usuario autenticado debe acceder a todos nuestros servicios. Escribimos nuestro primer controlador de solicitudes utilizando el esquema UserCreate y UserBase definido anteriormente.

 @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

Están sucediendo muchas cosas en este breve fragmento de código. Hemos usado un decorador para especificar el verbo HTTP, el URI y el esquema de respuestas exitosas. Para asegurarnos de que el usuario haya enviado los datos correctos, hemos escrito sugerencias en el cuerpo de la solicitud con un esquema UserCreate definido anteriormente. El método define otro parámetro para controlar la base de datos: se trata de la inyección de dependencia en acción y se analiza más adelante en este tutorial.

Asegurando nuestra API

Queremos las siguientes características de seguridad en nuestra aplicación:

  • Hash de contraseña
  • Autenticación basada en JWT

Para el hashing de contraseñas, podemos usar Passlib. Definamos funciones que manejen el hashing de contraseñas y verifiquen si una contraseña es correcta.

 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

Para habilitar la autenticación basada en JWT, necesitamos generar JWT y decodificarlos para obtener las credenciales de usuario. Definimos las siguientes funciones para proporcionar esta funcionalidad.

 # 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

Emitir tokens en el inicio de sesión exitoso

Ahora, definiremos un punto final de inicio de sesión e implementaremos el flujo de contraseña de OAuth2. Este punto final recibirá un correo electrónico y una contraseña. Comprobaremos las credenciales con la base de datos y, en caso de éxito, emitiremos un token web JSON para el usuario.

Para recibir las credenciales, haremos uso de OAuth2PasswordRequestForm , que forma parte de las utilidades de seguridad de 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"}

Uso de la inyección de dependencia para acceder a la base de datos y proteger los puntos finales

Hemos configurado el punto final de inicio de sesión que proporciona un JWT a un usuario al iniciar sesión correctamente. El usuario puede guardar este token en el almacenamiento local y mostrárselo a nuestro back-end como un encabezado de Autorización. Los puntos finales que esperan acceso solo de los usuarios registrados pueden decodificar el token y averiguar quién es el solicitante. Este tipo de trabajo no está vinculado a un punto final en particular, sino que es una lógica compartida que se utiliza en todos los puntos finales protegidos. Lo mejor es configurar la lógica de decodificación de tokens como una dependencia que se puede usar en cualquier controlador de solicitudes.

En FastAPI-speak, nuestras funciones de operación de rutas (controladores de solicitudes) dependerían entonces de get_current_user . La dependencia get_current_user debe tener una conexión a la base de datos y conectarse a la lógica OAuth2PasswordBearer de OAuth2PasswordBearer para obtener un token. Resolveremos este problema haciendo que get_current_user dependa de otras funciones. De esta manera, podemos definir cadenas de dependencia, que es un concepto muy poderoso.

 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

Los usuarios registrados pueden CRUD TODO

Antes de escribir las funciones de operación de ruta para TODO Create, Read, Update, Delete (CRUD), definimos las siguientes funciones auxiliares para realizar CRUD real en la base de datos.

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

Estas funciones de nivel de base de datos se utilizarán en los siguientes puntos finales 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"}

escribir pruebas

Escribamos algunas pruebas para nuestra API TODO. FastAPI proporciona una clase TestClient que se basa en la popular biblioteca Requests y podemos ejecutar las pruebas con Pytest.

Para asegurarnos de que solo los usuarios registrados puedan crear TODO, podemos escribir algo como esto:

 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

La siguiente prueba verifica nuestro punto final de inicio de sesión y genera un JWT si se presenta con credenciales de inicio de sesión válidas.

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

Resumiendo

Terminamos de implementar una aplicación TODO muy simple usando FastAPI. A estas alturas, ya ha visto el poder de las sugerencias de tipo que se utilizan para definir la forma de los datos entrantes y salientes a través de nuestra interfaz REST. Definimos los esquemas en un lugar y dejamos que FastAPI aplique la validación y conversión de datos. La otra característica notable es la inyección de dependencia. Usamos este concepto para empaquetar la lógica compartida de obtener una conexión a la base de datos, decodificar el JWT para obtener el usuario conectado actualmente e implementar OAuth2 simple con contraseña y portador. También vimos cómo las dependencias se pueden encadenar.

Podemos aplicar fácilmente este concepto para agregar funciones como el acceso basado en roles. Además, estamos escribiendo código conciso y poderoso sin aprender las peculiaridades de un framework. En palabras simples, FastAPI es una colección de herramientas poderosas que no tiene que aprender porque son solo Python moderno. Divertirse.