Parte 5: Cómo crear un CRUD con FastAPI

👋 ¡Hola! Espero que estéis bien 😉. Llegamos a la quinta y penúltima parte de esta serie de tutoriales y esta semana vamos a ver cómo crear todas las acciones de un CRUD en nuestro proyecto con FastAPI para administrar nuestra to-do list.

Si queréis acceder al resto de tutoriales, os dejo los enlaces aquí 👇

• Parte 1: Cómo crear una API REST COMPLETA con FastAPI, instalación y estructura
• Parte 2: Conexiones a bases de datos y creación de modelos con FastAPI
• Parte 3: Creación de modelos de Pydantic y nuestro primer usuario con FastAPI
• Parte 4: Autenticación con JWT en FastAPI
• Parte 6: Tests en FastAPI

Descripción del CRUD

Para poder realizar las acciones del CRUD el usuario deberá estar autenticado, ya que las tareas estarán relacionadas con el usuario y un usuario solo podrá listar, editar y borrar sus tareas.

Modelo de Pydantic para los to-dos

Antes de empezar con las acciones de nuestro CRUD, necesitaremos un par de modelos de Pydantic para trabajar con las respuestas/peticiones del usuario así que creamos un archivo llamado todo_schema.py dentro de /app/v1/schema y añadimos el siguiente contenido:

# Python
from datetime import datetime

# Pydantic
from pydantic import BaseModel
from pydantic import Field


class TodoCreate(BaseModel):
    title: str = Field(
        ...,
        min_length=1,
        max_length=60,
        example="My first task"
    )


class Todo(TodoCreate):
    id: int = Field(...)
    is_done: bool = Field(default=False)
    created_at: datetime = Field(default=datetime.now())

Como podéis observar, tenemos dos modelos, uno llamado TodoCreate que únicamente tendrá el campo title, ya que será el único campo que necesitaremos para creación y un segundo modelo Todo que extiende de TodoCreate con el resto de campos. Este modelo lo usaremos cuando un usuario nos pida una tarea en concreto o un listado de tareas.

Crear tareas

La primera acción que vamos a realizar será la creación, ya que es la que más sentido tiene hacer primero (necesitamos datos para actualizar, leer y borrar).

Para ello vamos a la carpeta /app/v1/service y creamos un archivo llamado todo_service.py con el siguiente contenido:

from fastapi import HTTPException, status

from app.v1.schema import todo_schema
from app.v1.schema import user_schema
from app.v1.model.todo_model import Todo as TodoModel


def create_task(task: todo_schema.TodoCreate, user: user_schema.User):

    db_task = TodoModel(
        title=task.title,
        user_id=user.id
    )

    db_task.save()

    return todo_schema.Todo(
        id = db_task.id,
        title = db_task.title,
        is_done = db_task.is_done,
        created_at = db_task.created_at
    )

Los imports ya los conocéis de tutoriales anteriores así que lo importante es la función create_task. Esta recibirá por parámetro la tarea y el usuario, después creamos una instancia del modelo de Todo de peewee y guardamos los datos. Por último retornamos la información en un objeto Todo de Pydantic.

Ahora vamos a definir la ruta para poder insertar nuestras tareas así que vamos al directorio /app/v1/router y creamos un archivo que llamaremos todo_router.py.

from fastapi import APIRouter, Depends, Body
from fastapi import status

from app.v1.schema import todo_schema
from app.v1.service import todo_service
from app.v1.utils.db import get_db
from app.v1.schema.user_schema import User
from app.v1.service.auth_service import get_current_user


router = APIRouter(prefix="/api/v1/to-do")

@router.post(
    "/",
    tags=["to-do"],
    status_code=status.HTTP_201_CREATED,
    response_model=todo_schema.Todo,
    dependencies=[Depends(get_db)]
)
def create_task(
    todo: todo_schema.TodoCreate = Body(...),
    current_user: User = Depends(get_current_user)):
    return todo_service.create_task(todo, current_user)

Como en el router de usuarios, definimos una instancia de APIRouter añadiendo la url base y posteriormente generamos nuestro primer endpoint.

En el decorador router indicamos que será una petición de tipo post.

Como path solo escribimos /, ya que en el prefijo tenemos la ruta que queremos utilizar.

Para el status_code usamos el estado 201 que es de creación.

En response_model añadimos el modelo que generamos antes para indicar que información queremos retornar al usuario.

Como dependecies añadimos la base de datos, ya que la necesitamos activa para generar los datos.

La función create_task recibirá por parámetro una variable llamada todo que será de tipo TodoCreate y la obtendremos del body de la petición. También indicamos que es obligatoria con los tres puntos.

Como segundo parámetro tenemos current_user que será de tipo User y que dependerá del resultado de la función get_current_user, la cual creamos en el tutorial anterior y recibirá de forma automáticamente nuestro token de autenticación. Comprueba si es válido y si es así devuelve el usuario. Si no es válido o no enviamos ningún token, devolverá una excepción explicando el error.

Por último solo debemos llamar a la función create_task de todo_service y si todo ha ido bien retornará nuestra nueva tarea.

Antes de probar nuestro endpoint, debemos añadirlo en el archivo main.py como hicimos con el router de usuarios así que abrimos el archivo añadimos el nuevo router, debería quedar de esta forma:

from fastapi import FastAPI

from app.v1.router.user_router import router as user_router
from app.v1.router.todo_router import router as todo_router

app = FastAPI()

app.include_router(user_router)
app.include_router(todo_router)

Ahora sí, vamos a probar nuestro nuevo endpoint. Como siempre vamos a la dirección http://127.0.0.1:8000/docs y desplegamos la acción cara crear nuestra tarea. Si os fijáis bien, veréis un candado abierto a la derecha. Esto indica que nuestro endpoint necesitará una autenticación para ser utilizado. 

Si pulsáis en él, se abrirá un popup con el que podremos autenticarnos, de esta manera podremos usar el token de autenticación de automáticamente en esta página.

Recordad que podemos usar tanto el username como el email.

Una vez hecho esto, pulsamos en Authorize y listo, mientras no recarguemos la página, tendremos un token que podremos usar en los endpoints que necesiten autenticación. Ahora también aparecerá el candado como cerrado indicando que estamos autenticados.

Como en casos anteriores, ahora solo debemos pulsar en Try it out y crear nuestra primera tarea.

Os recomiendo que generéis unas cuantas tareas para que ahora podamos realizar unas cuantas pruebas.

Obtener tareas

Los siguientes endpoints que vamos a generar van de la mano y serán el de obtener un listado con todas nuestras tareas y también la opción de obtener una tarea en concreto.

Para ello, vamos a crear las funciones que se encargarán de traer los datos así que vamos al archivo todo_service.py y añadimos el siguiente código:

def get_tasks(user: user_schema.User, is_done: bool = None):

    if(is_done is None):
        tasks_by_user = TodoModel.filter(TodoModel.user_id == user.id).order_by(TodoModel.created_at.desc())
    else:
        tasks_by_user = TodoModel.filter((TodoModel.user_id == user.id) & (TodoModel.is_done == is_done)).order_by(TodoModel.created_at.desc())

    list_tasks = []
    for task in tasks_by_user:
        list_tasks.append(
            todo_schema.Todo(
                id = task.id,
                title = task.title,
                is_done = task.is_done,
                created_at = task.created_at
            )
        )

    return list_tasks

def get_task(task_id: int, user: user_schema.User):
    task = TodoModel.filter((TodoModel.id == task_id) & (TodoModel.user_id == user.id)).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    return todo_schema.Todo(
        id = task.id,
        title = task.title,
        is_done = task.is_done,
        created_at = task.created_at
    )

La función get_tasks recibe por parámetro el usuario autenticado y el parámetro is_done. Este será un campo opcional y si es enviado nos permitirá filtrar por tareas hechas o por hacer. Ejecutamos la consulta y devolvemos el listado de tareas que pertenezcan al usuario. Si is_done es distinto de None, añadimos un filtro adicional para obtener las tareas que cumplan esa condición.

La función get_task recibirá el id de tarea y el usuario autenticado. Hacemos la consulta y si existe la tarea y pertenece al usuario la retornará, si no, lanzará una excepción avisando al usuario de que no puede encontrar la tarea.

Ahora vamos al archivo todo_router.py y añadimos los siguientes imports y rutas al código existente:

.
.
.
from fastapi import Query
from fastapi import Path

from typing import List, Optional
. 
.
.
@router.get(
    "/",
    tags=["to-do"],
    status_code=status.HTTP_200_OK,
    response_model=List[todo_schema.Todo],
    dependencies=[Depends(get_db)]
)
def get_tasks(
    is_done: Optional[bool] = Query(None),
    current_user: User = Depends(get_current_user)
):
    return todo_service.get_tasks(current_user, is_done)


@router.get(
    "/{task_id}",
    tags=["to-do"],
    status_code=status.HTTP_200_OK,
    response_model=todo_schema.Todo,
    dependencies=[Depends(get_db)]
)
def get_task(
    task_id: int = Path(
        ...,
        gt=0
    ),
    current_user: User = Depends(get_current_user)
):
    return todo_service.get_task(task_id, current_user)

El primer import que hemos añadido es Query de FastAPI. Este lo usaremos cuando queramos capturar parámetros vía query en la url.

¿Y cuáles son los parámetros de tipo query?

Pues bien, en una url serían los que se envían después del carácter `?`. Ejemplo:

https://cosasdedevs.com/stats?limit=10&sort=desc

En este caso los parámetros serían limit y sort a los cuales se les asigna los valores 10 y desc.

El segundo import es Path de FastAPI. En este caso lo utilizaremos para capturar parámetros que estén dentro de la url del endpoint.

Un ejemplo de parámetro de tipo path podría ser un id dentro de la url como podéis ver a continuación:

https://cosasdedevs.com/post/1785

En este caso podríamos emplear este número para retornar el post con ese identificador y según el número enviado retornará el post asociado.

El tercer import es List y Optional para tipar nuestros parámetros.

El primer router que añadimos será de tipo get y como url tendrá en endpoint principal. Su respuesta será una lista que contendrá objetos de tipos Todo.

La función recibirá dos parámetros. El primero is_done será un parámetro que se enviará vía query por la url y como suele ser en este tipo de parámetros, será opcional.

El segundo parámetro es el usuario, ya que lo necesitaremos para obtener las tareas del usuario en cuestión.

Después solo tenemos que llamar a la función get_tasks de todo_service que creamos en pasos anteriores y esta se encargará de retornar una lista de tareas asociadas al usuario.

El segundo router también es tipo get y en este caso nos encargaremos de obtener una tarea en concreto. Si os fijáis en la url que hemos añadido, hemos añadido la palabra task_id entre llaves. Esto quiere decir que será un parámetro dinámico y en él podremos pasar la información que queramos. Ya en la función podremos capturarlo por el mismo nombre y definir su tipo.

La respuesta será un modelo de Pydantic de tipo Todo y también necesitaremos la dependencia para conectarnos a la base de datos.

Ya en la función get_task definimos como primer parámetro task_id que será de tipo entero y expresaremos que será igual a la función Path que ya explicamos antes. También será un campo obligatorio y deberá ser mayor a 0.

Como en el caso anterior, también obtenemos el usuario autenticado, ya que únicamente queremos retornar una tarea si pertenece al usuario.

Por último llamamos a la función get_task de todo_service que creamos anteriormente y si la tarea existe y pertenece al usuario esta será retornada. Si no devolverá un error 404 avisando al usuario de que la tarea no ha sido encontrada.

Como siempre, podéis testar estos dos nuevos endpoints dentro de la documentación de FastAPI: http://127.0.0.1:8000/docs

Actualizar el estado de las tareas

Los siguientes endpoints que vamos a crear se van a encargar de cambiar el estado de una tarea como realizada o no realizada.

Para ello primero vamos a crear la función que se encargará de realizar la actualización así que vamos al archivo todo_service.py y añadimos el siguiente código:

def update_status_task(is_done: bool, task_id: int, user: user_schema.User):
    task = TodoModel.filter((TodoModel.id == task_id) & (TodoModel.user_id == user.id)).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    task.is_done = is_done
    task.save()

    return todo_schema.Todo(
        id = task.id,
        title = task.title,
        is_done = task.is_done,
        created_at = task.created_at
    )

Como veis, esta función recibe el parámetro is_done que indicará el nuevo estado de la tarea, el id de la tarea y el usuario autenticado.

Después comprobamos que la tarea existe y que pertenece al usuario y si no es así lanzamos una excepción.

Si la tarea existe y pertenece al usuario, guardamos y retornamos el objeto Todo con los valores actualizados.

Una vez hecho esto, vamos a añadir las rutas para poder utilizar esta función así que vamos al archivo todo_router.py y añadimos el siguiente código:

@router.patch(
    "/{task_id}/mark_done",
    tags=["to-do"],
    status_code=status.HTTP_200_OK,
    response_model=todo_schema.Todo,
    dependencies=[Depends(get_db)]
)
def mark_task_done(
    task_id: int = Path(
        ...,
        gt=0
    ),
    current_user: User = Depends(get_current_user)
):
    return todo_service.update_status_task(True, task_id, current_user)

@router.patch(
    "/{task_id}/unmark_done",
    tags=["to-do"],
    status_code=status.HTTP_200_OK,
    response_model=todo_schema.Todo,
    dependencies=[Depends(get_db)]
)
def unmark_task_done(
    task_id: int = Path(
        ...,
        gt=0
    ),
    current_user: User = Depends(get_current_user)
):
    return todo_service.update_status_task(False, task_id, current_user)

Este caso lo he planteado de forma que tendremos dos endpoints que recibirán el id de la tarea y serán de tipo patch (Recordad que el tipo patch se utiliza para indicar que queremos realizar una actualización parcial de un elemento). El primero finaliza su ruta con "mark_done" y se encargará de actualizar el estado de la tarea como hecho y el segundo finaliza con "unmark_done" y volverá a cambiar el estado de la tarea como no hecha.

Si os fijáis, en ambos casos empleamos la función update_status_task y como primer parámetro definimos si queremos pasar su estado a realizada o no realizada con el valor True o False.

Ahora que ya tenemos estos endpoints, podéis hacer unas pruebas y probar el filtro que añadimos para el endpoint que lista todas nuestras tareas.

Eliminar tareas

Por último vamos a definir un endpoint para eliminar una tarea. Para ello vamos al archivo todo_service.py y añadimos la siguiente función la cual se encargará de ello:

def delete_task(task_id: int, user: user_schema.User):
    task = TodoModel.filter((TodoModel.id == task_id) & (TodoModel.user_id == user.id)).first()

    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Task not found"
        )

    task.delete_instance()

La función recibe el id de tarea y el usuario autenticado y si la tarea existe y pertenece al usuario la eliminamos.

Ahora volvemos al archivo todo_router.py y añadimos el siguiente endpoint:

@router.delete(
    "/{task_id}/",
    tags=["to-do"],
    status_code=status.HTTP_200_OK,
    dependencies=[Depends(get_db)]
)
def delete_task(
    task_id: int = Path(
        ...,
        gt=0
    ),
    current_user: User = Depends(get_current_user)
):
    todo_service.delete_task(task_id, current_user)

    return {
        'msg': 'Task has been deleted successfully'
    }

Como podéis observar en el router, el método será de tipo delete, recibirá el id de tarea como parámetro en el path y luego lo capturaremos en la función. Por último solo debemos llamar a la función que acabamos de crear en todo_service.py para eliminar la tarea y si todo ha ido bien retornaremos un simple JSON con un mensaje diciéndole al usuario que se ha eliminado correctamente.

Y listo, ya tenemos nuestro proyecto casi completado y ahora únicamente nos faltan crear unos cuantos tests para verificar que su funcionamiento es el correcto, pero eso ya lo veremos la próxima semana.

Como en tutoriales anteriores, os dejo repositorio por si necesitáis echarle un vistazo https://github.com/albertorc87/fastapi-todo-api/tree/tutorial-5-crud.

Espero que este post te ayude y como siempre, te recomiendo seguirme en Twitter para estar al tanto de los nuevo contenido. Ahora también puedes seguirme en Instagram donde estoy subiendo tips, tutoriales en vídeo e información sobre herramientas para developers.

Por último os dejo mi guía para aprender a trabajar con APIs donde explico todo el funcionamiento de una API, el protocolo HTTP y veremos como construir una API con arquitectura REST.

Nos leemos 👋.