Aprendiendo FastAPI con PostgreSQL

FastAPI

Sirva este artículo como tutorial básico para trabajar con FastAPI, He decidido usar este framework/ORM para un side project que estoy desarrollando y que trata temas como Kafka, Python, microservicios, webscrapping.

El artículo es un resumen en español de la documentación oficial de FastAPI (que está realmente bien detallada), pero orientado a usar una base de datos Postgre, en mi opinión la parte que le falta a la documentación. Por eso hago incapié en el tema de Alembic para la ejecución de las migraciones en la base de datos Postgre.

Aquí la Documentación oficial.

Las librerías necesarias para trabajar con FastAPI y Postgre son las siguientes:

alembic==1.4.3
click==7.1.2
fastapi==0.63.0
flake8==3.8.4
h11==0.11.0
kafka-python==2.0.2
Mako==1.1.3
MarkupSafe==1.1.1
mccabe==0.6.1
psycopg2==2.8.6
pycodestyle==2.6.0
pydantic==1.7.3
pydocstyle==5.1.1
pyflakes==2.2.0
pylama==7.7.1
python-dateutil==2.8.1
python-editor==1.0.4
six==1.15.0
snowballstemmer==2.0.0
SQLAlchemy==1.3.22
starlette==0.13.6
typing==3.7.4.3
uvicorn==0.13.2

NOTA: Puede que al instalar psycopg2 nos de un error, eso es porque falta instalar a nivel de sistema (no en el entorno virtual) libpq-dev. Para instalar:

sudo apt-get install libpq-dev

Estructura de un proyecto FastAPI

Tenemos la carpeta principal del proyecto y generamos una para todo lo relacionado con FastAPI (la llamamos fastapi_app), dentro de esa carpeta estarán los modelos etc:

carpeta_proyecto
	alembic
	fastapi_app
		__init__.py
		base.py
		crud.py
		database.py
		main.py
		models.py
		schemas.py
	env
	.gitignore
	README.md
	LICENSE.md
	requirements.txt
	alembic.ini

Detalle de cada fichero

A continuación describimos para qué sirve cada uno de los ficheros.

init.py

Este fichero estará vacío

base.py

Este fichero es necesario para ejecutar las migraciones automáticamente, en él solo tendremos que hacer un import de todos nuestros modelos:

from fastapi_app.models import *

Después, dentro de alembic indicaremos que este fichero es el que tiene los metadatos de los modelos.

database.py

Aquí estableceremos la conexión con la base de datos. Ejemplo de fichero:

"""coding=utf-8."""

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:pass@host:port/db_name"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Si usamos mysql, en el create_engine hay que añadir un parámetro más, sería así:

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)


models.py

Aquí pondremos nuestros modelos de base de datos. Este sería un ejemplo de este fichero:

"""coding=utf-8."""

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base


class User(Base):
    """User Class contains standard information for a User."""

    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

schemas.py

Aquí estarán los modelos de Pydantic. Digamos que estarán las clases con los atributos que serán necesarios para crear o leer cuando se haga un GET. También se indica qué se devolverá cuando se cree un nuevo registro.

Ojo que algunos de los métodos pueden (y deben) tener herencia de otros.

Ejemplo de fichero:

"""coding=utf-8."""

from typing import List, Optional
from pydantic import BaseModel

class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True

crud.py

Aquí estarán los métodos que serán visibles desde la API, usará las clases Pydantic definidas en el fichero schemas.py para saber qué atributos son necesarios dependiendo de qué método se invoque. Ejemplo:

"""coding=utf-8."""

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

main.py

Integrar las partes que hemos comentado anteriormente. Al inicio del fichero, se pone una instrucción que crea los modelos en la base de datos, pero una vez creados, si añadimos o modificamos algún campo, esta instrucción no actualiza la tabla, de ahí el uso de alembic.

Ejemplo:

"""coding=utf-8."""

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

Alembic

Usamos Alembic para ejecutar las migraciones (cambios) en la base de datos. Alembic también tiene una estructura de carpetas propia, y en la raíz deja un fichero llamado alembic.ini para la configuración. La estructura de carpetas se genera sola inicializando alembic con el comando (ejecutar con el entorno virtual activado):

alembic init alembic

Configuración de Alembic

De momento, lo único que nos interesa de la configuración de alembic es la cadena de conexión a la base de datos, para configurarla modificamos esta línea del fichero:

sqlalchemy.url = postgresql://pass@host:port/db_name

Migraciones automáticas

Existe una forma de meter los cambios uno a uno a mano en la base de datos, pero lo más cómodo es hacer que los cambios se detecten automáticamente, para ello, anteriormente generamos dentro de la app de FastAPI un fichero con un import de todos los modelos de la base de datos, el fichero base.py.

Para que alembic tenga en cuenta los modelos, tenemos que ir al fichero env.py en el directorio de alembic e indicárselo:

import os,sys,inspect

current_dir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
parent_dir = os.path.dirname(current_dir)
sys.path.insert(0, parent_dir) 

from atlas_app.base import Base

target_metadata = Base.metadata

Generar las migraciones

Para generar las migraciones, con el entorno virtual activo ejecutamos este comando:

alembic revision --autogenerate -m "Added user table"

una vez hecho esto, hacemos que las migraciones persistan en la base de datos con el siguiente comando:

alembic upgrade head

Ejecutar la aplicación

Una vez generados los modelos, configurado cadenas de conexión etc, nuestra aplicación está lista para ejecutarse. Con el entorno virtual activo ejecutamos:

uvicorn atlas_app.main:app --reload

En la terminal nos aparecerá la url de la aplicación que por defecto será:

http://127.0.0.1:8000

si queremos acceder al Swagger que se monta automáticamente donde se listan los métodos de la API y pueden probarse, basta con dirigirnos a la url:

http://127.0.0.1:8000/docs

Este es el repositorio de GitHub del side project que he comentado antes. Donde estoy usando las herramientas que quiero aprender, como Kafka, FastAPI etc.

Publicado por Fj Asensi

BigData & MachineLearning Developer | Senior Microsoft Dynamics 365 Business Central Developer

Responder

Introduce tus datos o haz clic en un icono para iniciar sesión:

Logo de WordPress.com

Estás comentando usando tu cuenta de WordPress.com. Cerrar sesión /  Cambiar )

Google photo

Estás comentando usando tu cuenta de Google. Cerrar sesión /  Cambiar )

Imagen de Twitter

Estás comentando usando tu cuenta de Twitter. Cerrar sesión /  Cambiar )

Foto de Facebook

Estás comentando usando tu cuenta de Facebook. Cerrar sesión /  Cambiar )

Conectando a %s

A <span>%d</span> blogueros les gusta esto: