curso de postgres de básico a avanzado

Ahora que ya hemos visto los conceptos básicos de FastAPI, vamos a crear un ejemplo sencillo sobre una API de películas. Lo primero que haremos es crear nuestra carpeta.

mkdir peliculas
cd peliculas

Ahora crearemos nuestro entorno virtual y activaremos nuestro entorno virtual.

python -m venv .venv
source .venv/bin/activate

Instalaremos fastapi.

pip install fastapi

Creamos nuestra carpeta app/ y dentro de ella nuestros archivos principales para nuestra api: main.py, init.py y schemas.py.

peliculas
├── .venv
└── app/
    ├── __init__.py
    ├── main.py
    └── schemas.py

Con esto ya tendríamos las bases para comenzar a crear nuestra API.

A partir de aquí, imaginemos que nuestras películas pueden guardar la siguiente información:

Título
Director
Año de lanzamiento
Genero
Rating

Creación de las Películas

Comenzaremos nuestra api creando el endpoint para registrar nuestras películas. Nos iremos al archivo schemas.py que habíamos creado anteriormente y vamos a definir le modelo de creación.

from pydantic import BaseModel
from typing import Optional

class CrearPelicula(BaseModel):
    titulo: str
    director: str
    anio_lanzamiento: int
    genero: str
    rating: Optional[float] = None # Rating es opcional.

Ahora sentaremos las bases de nuestra API creando un archivo main.py. Aquí crearemos nuestra primera ruta y una lista en la cual guardaremos nuestras películas.

from fastapi import FastAPI, HTTPException, Body, status
from typing import List, Optional, Annotated
from app.schemas import CrearPelicula

app = FastAPI()

peliculas_db = [] # Vacia por el momento.
current_id = 1

# Ruta para crear película
@app.post(
    "/peliculas/", 
    status_code=status.HTTP_201_CREATED,
    description="Crea una nueva película en la base de datos.",
    response_model=int,
)
async def crear_pelicula(
    pelicula: Annotated[CrearPelicula, Body()]
):
    pelicula_con_id = {
        "id": current_id,
        "titulo": pelicula.titulo,
        "director": pelicula.director,
        "anio_lanzamiento": pelicula.anio_lanzamiento,
        "genero": pelicula.genero,
        "rating": pelicula.rating,
    }

    peliculas_db.append(pelicula_con_id)
    current_id += 1

    return pelicula_con_id["id"]

Ejecutamos el servidor con el siguiente comando:

fastapi dev app/main.py

Y probamos en nuestra documentacion en http://127.0.0.1:8000/docs.

alt text

Como podemos ver, todo funciona y nuestras películas estan siendo creadas.

Leer las Películas.

Ahora crearemos el endpoint para leer las películas que hemos creado. Algo importante a destacar en este punto es que cada vez que el servidor sufra un cambio o se reinicie, todo el código se ejecuta nuevamente, por lo que el arreglo de películas se reinicia a un arreglo vacío [] como definimos en nuestro código.

A partir de aquí, agregaremos tres películas a nuestro arreglo para poder probar nuestras rutas sin tener que recrear las películas nuevamente cada vez que se borren.

# ...Resto del codigo...

peliculas_db = [
    {
        "id": 1,
        "titulo": "Pulp Fiction",
        "director": "Quentin Tarantino",
        "anio_lanzamiento": 1994,
        "genero": "Acción",
        "rating": 8.9
    },
    {
        "id": 2,
        "titulo": "The Godfather",
        "director": "Francis Ford Coppola",
        "anio_lanzamiento": 1972,
        "genero": "Drama",
        "rating": 9.2
    },
    {
        "id": 3,
        "titulo": "The Dark Knight",
        "director": "Christopher Nolan",
        "anio_lanzamiento": 2008,
        "genero": "Acción",
        "rating": 9.0
    }
]

current_id = 4 ## Actualizar para mantener la secuencia.

# ...Resto del codigo...

Habiendo hecho esto, procedemos a crear nuestras rutas para obtener nuestras películas. Para ello, en el archivo schema.py, vamos a crear un modelo para la información que queremos devolver.

Vamos a devolver todos los campos, excepto el id de la película y el director.

from pydantic import BaseModel
from typing import List, Optional

## ...Resto del codigo de nuestros modelos

class DevolverPeliculaRespuesta(BaseModel):
    titulo: str
    anio_lanzamiento: int
    genero: str
    rating: Optional[float] = None

Ahora vayamos a main.py y creemos nuestro endpoint con el método GET:


# Ruta para obtener todas las películas
@app.get(
    "/peliculas/", 
    response_model=list[DeolverPeliculaRespuesta],
    description="Obtiene una lista de todas las películas.",
)
async def obtener_peliculas():
    return peliculas_db

Probamos en nuestra documentacion y vemos que no hay error

alt text

Sin embargo, este endpoint es muy simple. Hagamos que sea más completo agregando un parámetro para filtrar por género. Vamos a nuestro archivo schemas.py y agregaremos un modelo para definir los filtros:

from pydantic import BaseModel
from typing import List, Optional

## ...Resto del codigo de nuestros modelos

class FiltroPeliculaRequest(BaseModel):
    genero: str | None = None

Como es un método get, o un endpoint de solo lectura, no podemos usar el body de la request disponible, por lo que debemos usar queryparams para obtener la información.

from fastapi import Query

# ...Resto del codigo...

# Ruta para obtener todas las películas
@app.get(
    "/peliculas/", 
    response_model=list[DeolverPeliculaRespuesta],
    description="Obtiene una lista de todas las películas.",
)
async def obtener_peliculas(
    filtros: Annotated[FiltroPeliculaRequest, Query()],
):
    peliculas_a_enviar = []

    for pelicula in peliculas_db:
        if filtros.genero:
            if pelicula["genero"] == filtros.genero:
                peliculas_a_enviar.append(pelicula)
        else:
            peliculas_a_enviar.append(pelicula)

    return peliculas_a_enviar

Ahora probamos en nuestra documentacion en http://127.0.0.1:8000/docs,

alt text

y vemos como todo funciona.

Eliminar Película

Ahora crearemos el endpoint para eliminar una película. Este lo haremos más simple, ya que solo necesitamos enviar el ID al backend para eliminarla. Sin embargo, sí agregaremos el HTTPException para manejar el error en caso de que la película no exista, devolviendo un error 404.

Ahora crearemos el endpoint para eliminar una pelicula. Este lo haremos mas simple, ya que solo necesitamos enviar el id al backend para eliminarla. Sin embargo, si agregaremos el HTTPException para manejar el error en caso de que la pelicula no exista, devolviendo 404.


# ...Resto del codigo.

# Ruta para eliminar una película
@app.delete("/peliculas/{pelicula_id}", status_code=status.HTTP_204_NO_CONTENT)
async def eliminar_pelicula(pelicula_id: int):

    for index, pelicula in enumerate(peliculas_db):
        if pelicula["id"] == pelicula_id:
            index_eliminado = peliculas_db.pop(index)
            return index_eliminado

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Película con ID {pelicula_id} no encontrada"
    )

Actualizar Película

Por último, crearemos el endpoint para actualizar una película. En este caso, haremos uso tanto del ID enviado en el endpoint como de un esquema de datos para indicar qué parámetros queremos actualizar.

Vamos a schemas.py y crearemos un esquema para la actualización de las películas.

from pydantic import BaseModel
from typing import List, Optional

## ...Resto del codigo de nuestros modelos

class ActualizarPeliculaRequest(BaseModel):
    titulo: str | None = None
    director: str | None = None
    anio_lanzamiento: int | None = None
    genero: str | None = None
    rating: Optional[float] = None

from app.schemas import ActualizarPeliculaRequest

# ...Resto del codigo...

# Ruta para actualizar película
@app.put("/peliculas/{pelicula_id}")
async def actualizar_pelicula(
    pelicula_id: int, 
    pelicula_actualizada: Annotated[ActualizarPeliculaRequest, Body()]
):

    for index, pelicula in enumerate(peliculas_db):
        if pelicula["id"] == pelicula_id:
            nueva_pelicula = pelicula

            if pelicula_actualizada.titulo:
                nueva_pelicula["titulo"] = pelicula_actualizada.titulo

            if pelicula_actualizada.director:
                nueva_pelicula["director"] = pelicula_actualizada.director

            if pelicula_actualizada.anio_lanzamiento:
                nueva_pelicula["anio_lanzamiento"] = pelicula_actualizada.anio_lanzamiento

            if pelicula_actualizada.genero:
                nueva_pelicula["genero"] = pelicula_actualizada.genero

            if pelicula_actualizada.rating:
                nueva_pelicula["rating"] = pelicula_actualizada.rating

            peliculas_db[index] = nueva_pelicula
            return nueva_pelicula

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Película con ID {pelicula_id} no encontrada"
    )

Probamos en nuestra documentacion en http://127.0.0.1:8000/docs,

alt text

y vemos como todo funciona.

Habiendo hecho esto, ya tendríamos las bases para crear nuestros primeros endpoints. Sin embargo, estamos omitiendo lo más importante: conectar nuestra API a una base de datos real para hacer persistentes nuestros datos.

Contenido

4. Ejemplo Completo, Api de Peliculas

Creación de las Películas

Leer las Películas.

Eliminar Película

Actualizar Película