FastAPI y ORM SQLAlchemy con PostgreSQL

Si estás usando el framework FasAPI para la creación de tus APIs, querrás conectarlas con una base de datos, para eso usamos el ORM SQLAlchemy, el más popular usado por la comunidad de desarrollo en Python.

Vamos a ver como instalarlo y usarlo para hacer peticiones a una base de datos relacionales más que conocida como es PostgreSQL.

Preparando el entorno

Lo primero es crear un entorno con todos los paquetes necesarios, así que vamos a preparar un entorno virtual e instalar PostgreSQL, FastAPI y SQLAlchemy, así ¡que manos a la obra!

Instalando PostgreSQL

La instalación es diferente según el sistema operativo que tengas:

• MacOS

brew install postgresql@15

Una vez instalada, ejecutas el servicio con el siguiente comando:

brew services start postgresql@15

• Linux

apt install postgresql

Una vez instalada, ejecuta el servicio con el siguiente comando:

systemctl start postgresql

• Windows

Puedes tener PostgreSQL usando el instalador, lo encontrarás en esta web: Instalar PostgreSQL en Windows.

¡Ya tienes instalada la base de datos, vamos a por lo siguiente!

Creando entorno virtual

Lo primero que vamos a hacer es configurar Python, es necesario que lo tengas instalado, en el caso de que no lo tengas, puedes ir a su página oficial y realizar la instalación que es sencilla para todos los sistemas operativos: https://www.python.org/downloads/

Una vez ya tienes instalado Python, activamos un entorno virtual para nuestro proyecto, y así los paquetes son locales a él y no se instalan de forma global.

Primero hay que crearlo indicando el nombre que queremos:

python3 -m venv medium-fastapi

Después lo activamos:

Windows:

medium-fastapi/bin/activate.bat

Mac/Linux:

source medium-fastapi/bin/activate

Puedes comprobar que el entorno virtual está correctamente activado, ya que en la parte del path de tu consola verás el nombre que le hayas dado entre paréntesis.

Instalando FastAPI

Ahora instalamos la librería FastAPI:

pip install fastapi

También vas a necesitar instalar uvicorn, este paquete es el servidor web que va a exponer la API creado con FastAPI.

pip install uvicorn

Creando una API básica

Para empezar con nuestra API ya tenemos todo lo que necesitamos.

Archivo: /main.py

from fastapi import FastAPI
app = FastAPI()

@app.get('/')
def index():
  return { "message": "Hey is workgin!" }

La puedes ejecutar con el siguiente comando:

uvicorn main:app --reload

Si todo está correcto verás la salida en la terminal, para parar la ejecución ejecuta: CTRL + C

Ya tenemos un entorno preparado para configurar el resto de componentes.

Instalar SQLAlchemy

Ahora que ya tenemos la base de datos y la API, vamos a por el ORM, el elemento que los va a conectar y así poder comunicarlos.

La instalación se realiza desde el mismo directorio que hemos usado para FastAPI, estando en él ejecuta:

pip install sqlalchemy

Una vez instalado vamos a definir los elementos necesarios, configuración e inicio de la base de datos, modelo y esquema.

Conexión con la base de datos

Primero vamos a crear la configuración de la base de datos, esto lo voy a hacer en un fichero diferente a app.py, así podemos tener clara cada parte.

Fichero: /src/db/database.py

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

SQLALCHEMY_DATABASE_URL = 'postgresql://postgres:postgres@localhost/medium_sqlalchemy'

engine = create_engine(SQLALCHEMY_DATABASE_URL)

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

Base = declarative_base()

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

En las importaciones puedes ver que tenemos varias funciones de diferentes subpaquetes de sqlalchemy, create_engine, declarative_base y sessionmaker.

• Primero declaramos una constante con la URI para conectar con postgres, poniendo el usuario (postgres), el password (postgres), y después del caracter @, el nombre o IP del host y el nombre de la base de datos a la que queremos conectar.
• Creamos el motor de conexión con create_engine(), pasandole como argumento la constante con la URI.
• Después queremos obtener una sesión de donde sacar el objeto de base de datos, para eso, usamos sessionmaker() con los argumentos, autocommit y autoflush en False, así que tendremos que hacerlo esplicitamente en el momento de realizar las operaciones que lo necesiten, y así salvar los datos en la base de datos. También enlazamos bind, con el engine que hemos creado anteriormente.
• Declaramos la clase Base, de la que heredarán el resto de clases de los modelos que creemos. Estas clases mapean las tablas en forma de objeto.
• Por último creamos una función get_db(), desde la que devolveremos el objeto para gestionar la base de datos (db).

Creando el modelo

Ahora vamos a crear las clases que mapean la tabla de nuestra base de datos.

Fichero: /src/users/model.py

from db.database import Base
from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"

    id = Column(Integer,primary_key=True,nullable=False,autoincrement=True)
    name = Column(String,nullable=False)
    surname = Column(String,nullable=False)
    email = Column(String,nullable=False)
    password = Column(String,nullable=False)

Primero importamos la clase Base del fichero database, indicando el módulo db (el directorio que hemos creado). Desde sqlalchemy importamos lo que necesitamos para definir la tabla User: Column, Integer y String.

Definimos la case User() e indicamos que hereda de Base, pasándosela como argumento.

Indicamos el nombre de la tabla "users", y después indicamos los campos (columnas) que tendrá nuestra tabla.

id es un número entero (Integer), es una clave primaria, por lo que será un campo con valores únicos y por lo que será nuestra búsqueda principal, no podemos dejar el valor en blanco (null), y se incrementará automáticamente en 1 cada vez que introduzcamos un registro en la tabla.

El resto de campos, son todos iguales, todos son de tipo String y ninguno puede estar en blanco (null).

Creando los schemas

Una vez tenemos esto podemos crear los schemas y definir la estructura de los datos para las requests y las response.

Fichero: /src/users/user_schemas.py

from pydantic import BaseModel

class UserBase(BaseModel):
  name: str
  surname: str
  email: str
  password: str

  class Config:
    orm_mode = True

class UserRequest(UserBase):
  class Config:
    orm_mode = True

class UserResponse(UserBase):
  id: int

  class Config:
    orm_mode = True

FastAPI trabaja muy estrechamente con la librería pydantic, la cual ofrece la clase de la que heredan todas las clases que definen las estructuras de datos que recibimos y enviamos.

Importamos BaseModel, y definimos una serie de clases, UserBase, es la clase mas genérica con los datos necesarios del objeto (tabla) User. En ella definimos las propiedades (campos/columnas): name, surname, email y password, todas ellas del tipo string (str).

La clase UserRequest es la que define la estructura de los datos que nos llegan desde el cliente, en este caso es igual que la clase base. En cambio la clase UserResponse, añade una propiedad más id, ya que es un dato necesario para identificar al usuario en futuras peticiones desde el cliente.

En todas incluimos una clase Config que inicia el modo orm a verdadero.

Si te gusta el contenido que te comparto, puedes apoyarme haciendo una aportación! :) https://buymeacoffee.com/raulalhena

Creando los end points

Ya tenemos todas las estructuras necesarias para enviar y recibir datos y la base de datos lista para comunicarse con nuestra API, así que vamos a crear el fichero app.py donde se ejecutará toda la lógica necesaria para crear usuarios y consultar todos los usuario existentes en la base de datos.

Fichero: /src/app.py

from fastapi import FastAPI, Depends, status
from typing import List
from db.database import engine, get_db
from users import user_model
from users.user_schemas import UserRequest, UserResponse
from users.user_model import User
from sqlalchemy.orm import Session
from datetime import datetime

app = FastAPI()

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

@app.get('/')
def index():
    return { 'message': 'Server alive!', 'time': datetime.now() }

@app.get('/users', status_code=status.HTTP_200_OK, response_model=List[UserResponse])
def get_all_users(db: Session = Depends(get_db)):
    all_users = db.query(User).all()
    return all_users

@app.post('/users', status_code=status.HTTP_201_CREATED, response_model=UserResponse)
def create_user(post_user: UserRequest, db: Session = Depends(get_db)):
    new_user = User(**post_user.model_dump())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user.__dict__

Imports

Importamos los elementos necesarios de FastAPI, la instancia principal de la aplicación, el objeto status para indicar el código de estado de la respuesta y Depends para gestionar las dependencias de nuestros end points.

Importamos el tipo List de la librería Typing, que nos sirve para tipar de forma estática los tipos de List, Dict, Tuples, que Python no dispone de forma nativa, ya que sólo se pueden tipar variables y funciones.

Después importamos las funciones de configuración de la base de datos, el modelo y las clases que representan el objeto Request y Response de User.

Por último la sesión desde SQLAlchemy y datetime para mostrar que la API está activa.

Instanciamos la clase FastAPI, y con user_model.Base.metadata.create_all(bind=engine) se crea la tabla en la base de datos. Esta función tiene un argumento iniciado en True, que comprueba si la tabla existe, en el caso que sea así, no la crea de nuevo.

End points

Después ya encontramos los end points. Puedes ver el '/', que es la raíz del dominio, que únicamente devuelve un mensaje con la fecha y la hora actual.

GET /users

El siguiente end point '/users', es de tipo GET, y sirve para obtener todos los registros de la tabla users. Le indicamos el código de estado 200 cuando la petición es correcta, y el modelo de respuesta, en este caso, una lista (List) del tipo UserResponse.

Este end point ejecuta la función get_all_users() que como argumento recibe la conexión de la base de datos, que guarda en la variable db, tipada como Session, y almacena el valor devuelto por get_db a través de la función Depends() que facilita FastAPI.

Esta función ejecuta la petición que obtiene todos los registros de la tabla users: db.query(User).all() que ofrece SQLAlchemy.

Estos registros se almacenan en la variable all_users y es lo que devolvemos al cliente.

POST /users

El otro end point '/users', es de tipo POST, y nos permite crear un nuevo usuario en la tabla users de la base de datos. También devolvemos el código de estado si la petición es correcta, 201 es el de creación exitosa de un recurso, y la respuesta es del mismo tipo que el anterior, pero sólo un objeto UserResponse.

La función es create_user() que recibe como argumento, a parte de la instancia de la conexión con la base de datos (db), un objeto post_user, donde están todas las propiedades que se envían desde el cliente.

Los datos que recibe el end point y que se almacenan en post_user:

{
  "name": "user1",
  "surname": "user1surname",
  "email": "user1@email.com",
  "password": "1234"
}

Esta estructura corresponde con el schema de UserBase que hemos creado y corresponde a UserRequest.

En la función primero se crea un diccionario del objeto post_user usando model_dump(), los dos asteriscos se usan para "desempacar" las key y values del diccionario, el resultado se guarda en new_user.

Después se registra una transacción con db.add(), pero esta transacción no se comunica a la base de datos, eso lo hace la función .commit() y por último .refresh(new_user), hace un update de la variable new_user.

En el momento de hacer un return, se transforma el objeto de la base de datos en un diccionario con la propiedad __dict__.

Conclusiones

FastAPI es un framework de fácil uso, y junto con SQLAlchemy tienes un potente y rápido combo para crear APIs con conexión a una base de datos con todas las herramientas necesarias para recibir, enviar y comprobar los tipos de datos que se ajustan a las necesidades de tu aplicación.

Como siempre te dejo mi repositorio de Github donde encontrarás todo el código del ejemplo: FastAPI, ORM SQLAlchemy con PostgreSQL.

¡Felicidades, ya sabes como crear una API que puede conectarse a una base de datos usando SQLAlchemy!