FastAPI WebSockets: Guía Completa con Autenticación, REST API y Vue.js

▶

¿Qué es un WebSocket?

Lo primero que debemos entender es que los WebSockets no son una tecnología exclusiva de FastAPI o Python; son un estándar global. Al igual que el protocolo HTTP, los WebSockets existen para solucionar un problema específico de comunicación, pero operan de una manera completamente distinta.

Un WebSocket es un protocolo de comunicación que proporciona un canal de comunicación full-duplex sobre una única conexión TCP. A diferencia del ciclo de solicitud-respuesta de HTTP, que es stateless (sin estado), una conexión WebSocket permanece abierta, permitiendo que tanto el cliente como el servidor envíen mensajes en cualquier momento.

• HTTP (Stateless): El cliente envía una petición, el servidor responde y la conexión se cierra. Cada petición es independiente y no tiene conocimiento de las anteriores. Para simular interactividad, se usan técnicas como el polling, que es ineficiente.
• WebSockets (Stateful, Full-duplex): Se establece una conexión persistente. El servidor puede "empujar" (push) datos al cliente sin que este los solicite explícitamente, y viceversa. Esto reduce la latencia y el overhead.

Mientras que aplicaciones tradicionales (FastAPI, Django, Flask, Laravel) utilizan HTTP para servir contenido, los WebSockets son un protocolo que permite una conexión bidireccional y persistente.

Toda esta implementación la puedes adaptar para una arquitectura Limpia en FastAPI

El problema del modelo Cliente-Servidor tradicional (HTTP)

En el esquema clásico de HTTP, la comunicación funciona así:

1. El Cliente (Navegador): Hace una petición cuando "le sale del forro" (al recargar o navegar).
2. El Servidor: Recibe la petición, procesa y devuelve una respuesta.
3. Fin de la comunicación: Una vez entregada la respuesta, la conexión muere.

Si el servidor tiene información nueva, es imposible que se la envíe al cliente a menos que este la solicite primero. En un diseño tradicional, si no hay una acción del usuario, el servidor no puede comunicarse contigo.

La solución: WebSockets y la comunicación Full-Duplex

¿Qué pasa si necesitas una aplicación de chat en tiempo real (como WhatsApp o Telegram)? Si usáramos HTTP, el cliente tendría que preguntar cada 2 o 3 segundos: "¿Hay mensajes nuevos? ¿Y ahora? ¿Y ahora?". Esto es extremadamente ineficiente y consume recursos innecesarios.

Los WebSockets resuelven esto creando un canal de comunicación persistente. Es como entrar en una habitación:

• Conexión persistente: Una vez que el servidor acepta al cliente, el canal queda abierto (como una llamada telefónica activa).
• Full-Duplex: El servidor puede enviarte actualizaciones en el momento que quiera, y tú puedes enviarle datos a él simultáneamente sin cerrar la conexión.

Estados y Funcionamiento Técnico: El ciclo de vida de un WebSocket

A diferencia de HTTP, que es un protocolo "sin estado" (stateless), los WebSockets introducen estados adicionales que debemos gestionar:

• Handshake (Apretón de manos): El cliente pide unirse y el servidor decide si lo acepta.
• Conexión activa: El estado donde se intercambian mensajes.
• Intercambio de mensajes: Los datos se envían en "frames" que pueden ser de texto, binarios o JSON.
• Desconexión: Ya sea porque el cliente cierra el navegador o el servidor decide "botar" al usuario.

En herramientas como FastAPI o Django Channels, esto se traduce en métodos específicos para conectar, recibir mensajes y desconectar.

Cómo identificar un WebSocket en el navegador

Cuando inspeccionas una página web normal en el apartado de Network (Red), verás que las peticiones empiezan por http:// o https://. Sin embargo, los WebSockets utilizan un esquema distinto:

• ws:// (WebSocket simple)
• wss:// (WebSocket seguro)

En el código de FastAPI, esto lo verás reflejado no con los típicos decoradores @app.get o @app.post, sino con @app.websocket.

Casos de uso comunes

• Chats en vivo: Comunicación instantánea entre usuarios.
• Notificaciones Push: Avisos que llegan al teléfono o navegador al instante.
• Colaboración en vivo: Como Google Docs, donde ves lo que otros escriben en tiempo real.
• Mercados financieros: Actualización de precios de acciones o criptomonedas segundo a segundo.

Ahora que tenemos clara la teoría y la diferencia entre un canal sin estado (HTTP) y uno persistente (WebSocket), ¡estamos listos para empezar con la práctica en FastAPI!

Creación de una aplicación mínima con WebSockets en FastAPI

Nuestro propósito en este apartado es construir una aplicación básica para conectarnos a un WebSocket. Para ello, utilizaremos Jinja2 para renderizar la plantilla HTML desde la cual consumiremos el servicio.

La estructura mínima de nuestra aplicación quedaría así:

from fastapi import FastAPI, APIRouter, WebSocket, Request
from fastapi.templating import Jinja2Templates
app = FastAPI()
router = APIRouter()
templates = Jinja2Templates(directory="templates")

Implementación del WebSocket en el Servidor

A diferencia de las rutas tradicionales (GET, POST, etc.) que utilizan el protocolo HTTP, definiremos un endpoint específico para WebSockets:

@router.websocket("/ws")

Y creamos un WS tipo Hola Mundo, es decir, es lo más sencillo posible:

api.py

@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
   await websocket.accept() # Aceptamos la conexión
   while True:
       data = await websocket.receive_text() # Esperamos mensajes
       await websocket.send_text(f"Mensaje recibido: {data}") # Respondemos

Como comentamos antes, al ser los WS una conexión con estado y permanente entre el cliente y el servidor, existen un paso preivo antes de enviar/recibir mensajes que es aceptar la conexión:

websocket.accept()

Lógicamente, puedes implementar lógica antes de aceptar la conexión dependiendo del propósito de la aplicación, por ejemplo, si fuera un servicio de pago, puedes verificar el registro del pago del usuario ANTES de aceptar la petición.

El resto de la función lo que haces es esperar un mensaje del usuario (que enviaremos mediante un formulario que implementamos en el siguiente apartado):

data = await websocket.receive_text() # Esperamos mensajes

Y luego enviar el mensaje de vuelta con un envoltorio adicional de “Mensaje recibido: ”:

await websocket.send_text(f"Mensaje recibido: {data}") # Respondemos

¿Por qué usamos async?

La función debe ser asíncrona (async def). Esto es vital porque la conexión de un WebSocket es como una "habitación" permanente. Si usáramos un esquema síncrono (como en versiones antiguas de Django), el hilo del servidor se quedaría "pegado" esperando la respuesta del usuario.

Con async, si el usuario no está interactuando, el servidor puede liberar ese hilo para procesar otras peticiones, haciendo que FastAPI sea extremadamente eficiente.

El método anterior totalmente eficiente aunque veas un bucle infinito con el while True, en el cual, al tener un async dentro de el, el hilo asíncrono NO queda a la espera, lo cual sucedería si la función NO fuera asíncrona, si no, el hilo puede realizar cualquier tarea MIENTRAS espera las respuestas del resto de las funciones asíncronas que espera:

data = await websocket.receive_text() # Esperamos mensajes
await websocket.send_text(f"Mensaje recibido: {data}") # Respondemos

El Cliente: Conexión desde el Navegador

Para interactuar con nuestra API, crearemos un formulario en HTML y utilizaremos JavaScript Vanilla. Es importante recordar que JavaScript es independiente del framework del backend; la conexión será la misma ya sea que uses FastAPI, Django o Node.js:

templates\ws\chat.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Document</title>
</head>
<body>
    <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
</body>
</html>

El Protocolo ws://

Lo primero que notarás en el script es que la URL de conexión no empieza con http, sino con ws:// (o wss:// si es seguro), ya que es un protocolo estándar diferente.

var ws = new WebSocket("ws://localhost:8000/ws");

JavaScript

Conexión al servidor: 

const socket = new WebSocket("ws://localhost:8000/ws");

Escuchar mensajes del servidor:

socket.onmessage = function(event) {
   const messages = document.getElementById('messages');
   const message = document.createElement('li');
   message.textContent = event.data;
   messages.appendChild(message);
};

Por el resto del código anterior, lo que haces es obtener referencia al HTML, al UL, crear un LI y establecerlo, que corresponde al mensaje de vuelta del servidor.

Enviar mensajes al servidor:

function sendMessage(event) {
   const input = document.getElementById("messageText");
   socket.send(input.value);
   input.value = '';
   event.preventDefault();
}

En el código anterior, al enviar el formulario, se llama la función de sendMessage(), obtiene la referencia al input, su valor y lo envía al servidor.

Verificación en las Herramientas de Desarrollador

Al ejecutar la aplicación con uvicorn, podemos abrir la consola del navegador (F12) y dirigirnos a la pestaña Network.

• Handshake: Verás una petición con estado 101 Switching Protocols. Esto indica que el servidor aceptó "cambiar" de HTTP a WebSocket.
• Frames: Dentro de la conexión, podrás ver las flechas hacia arriba (mensajes enviados por ti) y las flechas hacia abajo (respuestas del servidor).
• Si cierras la pestaña o detienes el servidor, se disparará el evento onclose, cerrando la "habitación" de comunicación. Este flujo de Aceptar -> Escuchar/Enviar -> Cerrar es la base de cualquier sistema en tiempo real.

Figura 14--1: Conexión y envío de mensajes por el WS

Envío y Recepción de Datos

El objeto websocket ofrece varios métodos para intercambiar datos, dependiendo del formato:

• Recepción: receive_text(), receive_json(), receive_bytes().
• Envío: send_text(), send_json(), send_bytes().

Es importante usar await con todos ellos, ya que son operaciones de red asíncronas.

Excepciones, desconectar y conexiones simple

En la implementación anterior, vimos un WebSocket en su mínima expresión; lo que podríamos llamar el "Hola Mundo" de esta tecnología. En resumidas cuentas, nos permitía enviar un mensaje desde nuestra aplicación y recibirlo de vuelta.

Es importante notar que aquí no estamos empleando HTTP, sino el protocolo WS. Hasta ahora hemos visto tres métodos principales:

• Aceptar la conexión: El apretón de manos (handshake) inicial.
• Recibir texto: Escuchar lo que el cliente envía.
• Enviar texto: La respuesta del servidor.

A diferencia del esquema tradicional de HTTP (cliente-servidor de una sola dirección), los WebSockets funcionan como una habitación donde los mensajes fluyen de manera Full-Duplex. Esto significa que tanto el cliente como el servidor pueden hablar al mismo tiempo sin esperar a que el otro termine.

Comparativa y Ciclo de Vida: FastAPI vs Django Channels

Para que entiendas mejor la estructura, podemos compararlo con una implementación en Django Channels. Aunque la sintaxis varía según la tecnología, el concepto es idéntico. En Django, solemos tener métodos más organizados como connect, disconnect y receive:

import json
from channels.generic.websocket import WebsocketConsumer
class ChatConsumer(WebsocketConsumer):
    def connect(self):
        print('Connect')
        self.accept()
    def disconnect(self, code):
        print("Disconnet")
    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        print(message)
        self.send(text_data=json.dumps({ 'message':message }))

En FastAPI, aunque parezca que todo está más "mezclado" dentro de un bucle while True, podemos completar el esquema manejando la desconexión mediante excepciones.

Manejo de la desconexión con WebSocketDisconnect

Es fundamental implementar el método de desconexión para liberar recursos. Si estás creando un chat o una aplicación que requiere un conteo de tiempo de conexión, debes saber cuándo el cliente se retira.

Para esto, envolvemos nuestra lógica en un bloque try...except...finally:

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            print('conection open')
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        # --- EVENTO: AL DESCONECTARSE (Cierre limpio) ---
        print("El cliente cerró la conexión")
        
    except Exception as e:
        # Captura otros errores inesperados
        print(f"Error inesperado: {e}")
        
    finally:
        # --- LÓGICA FINAL ---
        # Este bloque se ejecuta siempre, ideal para limpieza de recursos
        print("Limpieza de conexión finalizada")

• except WebSocketDisconnect: Se ejecuta cuando el cliente cierra la pestaña o pierde la conexión. Aquí puedes ejecutar lógica de limpieza.
• finally: Un bloque que siempre se ejecutará, independientemente de si la conexión terminó bien o con un error. Es el lugar ideal para cerrar procesos internos.

Conexiones simple

Para cerrar este apartado, debemos identificar una limitación crítica en nuestra implementación actual. Si abres dos pestañas del navegador (dos instancias del cliente) y escribes en una, verás que el mensaje no aparece en la otra.

¿Por qué sucede esto? Actualmente, cada conexión es tratada como una habitación privada donde solo están el cliente y el servidor. Cada vez que nos conectamos, FastAPI crea un espacio distinto y estos espacios no se comunican entre sí.

Si quisiéramos crear una aplicación de chat grupal o una herramienta colaborativa, este esquema es inútil porque necesitamos que el servidor sea capaz de retransmitir (broadcast) el mensaje a todos los clientes conectados que veremos en el siguiente apartado.

Gestión de Múltiples Conexiones (Connection Manager)

Hasta ahora, nuestra comunicación vía WebSockets era bastante limitada, ya que no podíamos interactuar con otros usuarios conectados al mismo servidor. Cada conexión era una "isla" aislada. Para solucionar esto y permitir la creación de habitaciones o grupos donde los mensajes se compartan entre varios participantes, vamos a implementar una lógica de gestión de conexiones.

En una aplicación real, no solo chateamos con nosotros mismos. Necesitamos una forma de gestionar múltiples clientes conectados y enviarles mensajes.

Como cada conexión WebSocket es manejada por una instancia separada de nuestra función de endpoint, no podemos simplemente usar una lista local para rastrear las conexiones. Necesitamos una clase singleton o un objeto global que gestione el estado de todas las conexiones activas.

Creación de una clase ConnectionManager

Para organizar mejor el código, hemos creado una clase llamada ConnectionManager. Esta clase se encarga de rastrear manualmente quién está conectado, ya que en FastAPI debemos gestionar este listado nosotros mismos.

Esta clase centralizará la lógica para conectar, desconectar y enviar mensajes a los clientes.

api_multiple.py

from fastapi import FastAPI, APIRouter, Request, WebSocket, WebSocketDisconnect
from fastapi.templating import Jinja2Templates
templates = Jinja2Templates(directory="templates/")
# uvicorn api:app --reload
# 
app = FastAPI()
router = APIRouter()
@app.get('/')
def form(request: Request):
    return templates.TemplateResponse(request=request, name='ws/chat.html')
from typing import List
class ConnectionManager:
    def __init__(self):
        # Lista para almacenar las conexiones activas
        self.active_connections: List[WebSocket] = []
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    # async def send_personal_message(self, message: str, websocket: WebSocket):
    #     await websocket.send_text(message)
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Cliente #{client_id} dice: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Cliente #{client_id} se ha desconectado")

Constructor y manejo de conexiones

En el método constructor lo que vamos a hacer es manejar varias conexiones.

En el caso de FastAPI, tenemos que agregar esas conexiones de manera manual. Es decir, cuando ocurran las conexiones las vamos a agregar a una lista.

Por aquí fíjate que declaramos un array de tipo WebSocket, porque son las conexiones, y lo inicializamos por defecto como una lista vacía. Cuando inicia el servidor, esto se carga así, vacío, esperando conexiones.

Método connect

Ahora tenemos un método llamado connect, en el cual vamos a recibir la conexión del cliente. Esto viene siendo justamente la conexión WebSocket.

Lo primero que hacemos es aceptar la conexión, es decir, esperamos a que el proceso se complete exitosamente.
Aquí podrías agregar cualquier lógica de negocio: verificar autenticación, validar permisos, comprobar algún estado, etc.

Una vez aceptada la conexión, la agregamos al array que declaramos antes. Por lo tanto, ya tenemos una conexión almacenada, o varias conexiones, que en nuestro ejemplo serán dos o más cuando lo probemos.

Método disconnect

El método disconnect lo que hace es remover la conexión del listado.

No hay ninguna lógica adicional aquí, porque la conexión ya existe: simplemente la quitamos de la lista cuando el cliente se desconecta.

Manejo de la desconexión

Cuando el cliente se desconecta, aprovechamos el mismo esquema: llamamos a disconnect y removemos la conexión del listado.

Este método no necesita ser asíncrono, porque la conexión ya no existe, simplemente la quitamos de la lista.

Envío de mensajes y broadcast

Luego tenemos los métodos para enviar mensajes.

El mensaje es un tema interesante porque puede variar el formato: texto plano, JSON, etc., pero eso lo veremos más adelante.

Aquí siempre utilizamos el WebSocket para enviar los mensajes.

El método broadcast es asíncrono, como todas las operaciones que trabajan con WebSockets.

Si tenemos conexiones activas, enviamos el mensaje a cada una de ellas. Esto no ocurre de forma mágica: lo hacemos manualmente, iterando sobre la lista de conexiones y enviando el mensaje una por una.

Finalmente, esta sería toda la definición de la clase.

Creando la ruta WebSocket

Al igual que antes, ahora creamos una ruta que nos permita consumir este WebSocket.

@app.websocket("/ws/{client_id}")

Fíjate que ahora estamos recibiendo un parámetro llamado client_id. Este parámetro identifica la habitación.

Todos los clientes que tengan el mismo client_id se van a conectar al mismo WebSocket y podrán comunicarse entre ellos.

Creamos el método, recibimos la conexión y el client_id, utilizamos la instancia del manager y llamamos al método connect. Luego entramos en el while true, esperamos recibir datos y llamamos al método broadcast, que envía el mensaje a todos los clientes conectados a esa habitación.

En el template, coloca de manera demostrativa un cliente fijo:

templates\ws\chat.html

var ws = new WebSocket("ws://localhost:8000/ws/5");

$ uvicorn api_multiple:app --reload

18.4 Manejo de Errores y Desconexiones

Una conexión WebSocket puede terminar abruptamente. Nuestro servidor debe ser robusto y manejar estas situaciones con elegancia.

Excepción WebSocketDisconnect

Cuando un cliente se desconecta (cierra la pestaña, pierde conexión a internet), la llamada a receive_text() o receive_json() lanzará una excepción WebSocketDisconnect. Debemos capturarla para limpiar la conexión de nuestra lista y evitar fugas de memoria.

El ejemplo anterior ya incluye un bloque try...except para este propósito. Es fundamental para mantener la integridad de nuestra lista de conexiones.

Bloques Try/Except para mensajes malformados

Si esperas recibir JSON (receive_json()) y el cliente envía texto plano, la aplicación crasheará. Es una buena práctica envolver la recepción de datos en un bloque try...except más general para manejar datos inválidos sin que el servidor se caiga.

18.5 Seguridad en WebSockets

Proteger los endpoints de WebSocket es tan crucial como proteger las rutas HTTP, pero presenta desafíos únicos.

Autenticación

El sistema de inyección de dependencias de FastAPI (Depends) no funciona directamente con WebSockets de la misma manera que con las rutas HTTP. Esto se debe a que el handshake de WebSocket no es una petición HTTP estándar.

La solución más común es pasar el token de autenticación (como un JWT) a través de parámetros de consulta (query params) en la URL de conexión.

# Cliente se conecta a: ws://example.com/ws?token=eyJ...
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str):
    user = get_user_from_token(token) # Función para validar el token
    if user is None:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    
    await manager.connect(websocket)
    # ... resto de la lógica

Dentro de la función, validamos el token manualmente. Si no es válido, cerramos la conexión con un código de estado apropiado.

Limitación de Conexiones

Para prevenir ataques de denegación de servicio (DoS) donde un atacante abre miles de conexiones para agotar los recursos del servidor, se pueden implementar estrategias de limitación de velocidad (rate limiting) o restringir el número de conexiones por IP o por usuario.

18.6 Cliente de Pruebas (Frontend Simple)

Para probar nuestro backend, podemos crear un cliente simple con HTML y JavaScript nativo.

<!DOCTYPE html>
<html>
    <head>
        <title>Chat con FastAPI</title>
    </head>
    <body>
        <h1>Chat en Tiempo Real</h1>
        <ul id='messages'></ul>
        <input type="text" id="messageText" autocomplete="off"/>
        <button onclick="sendMessage()">Enviar</button>
        <script>
            // Conectarse al endpoint de WebSocket
            var ws = new WebSocket("ws://localhost:8000/ws/1");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages');
                var message = document.createElement('li');
                var content = document.createTextNode(event.data);
                message.appendChild(content);
                messages.appendChild(message);
            };
            function sendMessage() {
                var input = document.getElementById("messageText");
                ws.send(input.value);
                input.value = '';
            }
        </script>
    </body>
</html>

Este código se conecta al servidor, escucha los mensajes entrantes (onmessage) para agregarlos a una lista en el DOM, y envía el contenido de un campo de texto al servidor cuando se hace clic en un botón (ws.send()).

18.7 Integración con otros módulos

Los WebSockets se vuelven aún más poderosos cuando se combinan con otras partes de tu aplicación.

WebSockets + Base de Datos

Podemos guardar el historial de un chat en la base de datos antes de emitirlo a los demás usuarios. Esto asegura que los mensajes persistan.

# Dentro del bucle while del WebSocket
data = await websocket.receive_text()
# Guardar en la base de datos (usando una librería async como databases o Tortoise ORM)
await database.execute(query=messages.insert().values(content=data, user_id=client_id))
# Emitir a todos los usuarios
await manager.broadcast(f"Cliente #{client_id} dice: {data}")

WebSockets + Background Tasks

Imagina que un usuario solicita la generación de un reporte pesado a través de una ruta HTTP. El proceso puede tardar varios minutos. En lugar de hacer esperar al usuario, podemos iniciar una tarea en segundo plano y notificarle a través de un WebSocket cuando el reporte esté listo.

from fastapi import BackgroundTasks
# Función que se ejecuta en segundo plano
def generate_report(user_id: str, report_id: str):
    # ... lógica pesada para generar el reporte ...
    message = f"Tu reporte {report_id} está listo."
    # Usamos el manager para notificar al usuario específico
    # Nota: esto requiere que el manager sea accesible y que la función pueda ser async
    # o que se use un helper para correr la corutina en el event loop.
    import asyncio
    asyncio.run(manager.send_personal_message_to_user(user_id, message))
@app.post("/generate-report")
async def create_report(background_tasks: BackgroundTasks, current_user: User = Depends(get_current_user)):
    report_id = "some_unique_id"
    background_tasks.add_task(generate_report, current_user.id, report_id)
    return {"message": "La generación del reporte ha comenzado. Serás notificado cuando esté listo."}

Configurar los CORS en FastAPI para consumir desde Vue

Como te comenté al inicio de esta sección, mi idea es comparar lo que hicimos en el curso de Django. Si no has tomado esa formación, te la recomiendo para que tengas la visión de otro excelente framework (personalmente, Django es mi favorito).

Aunque no es obligatorio, ya que aquí explicaré todo desde cero, en aquel curso creamos un pequeño proyecto en Vue para consumir una API REST.

Adaptaremos ese proyecto, que originalmente consumía Django Channels, para conectarlo ahora con FastAPI mediante WebSockets. Esto nos permitirá identificar algunos problemas comunes en la comunicación entre aplicaciones:

https://github.com/libredesarrollo/curso-libro-django-vue-channels

¿Qué son los CORS?

Por si no lo mencionamos antes, CORS significa Cross-Origin Resource Sharing (Intercambio de Recursos de Origen Cruzado). Es un mecanismo de seguridad que implementan los navegadores para controlar cómo se comparten recursos entre distintos dominios. En nuestro caso, al ser proyectos separados, FastAPI bloquea el acceso por seguridad hasta que nosotros le indiquemos explícitamente qué aplicaciones tienen permiso para conectarse.

Aplicaciones de Terceros y Conectividad

Algo interesante que notarás es que, hasta ahora, hemos consumido los WebSockets directamente desde el mismo proyecto en FastAPI. Es como si creáramos una API para consumirla nosotros mismos; tiene sentido para algunas pruebas, pero el verdadero potencial de una aplicación surge cuando se interconecta con aplicaciones de terceros, como este proyecto en Vue.

Al intentar conectar una aplicación externa, nos topamos con el primer gran obstáculo: los benditos CORS (Cross-Origin Resource Sharing).

Cómo probar el proyecto en Vue:

• Asegúrate de tener Node.js instalado.
• Clona el repositorio con git clone.
• Instala las dependencias con npm install.

El código que debes de empezar a adaptar queda como:

src\App.vue

<template>
  <div class="container">
    <MessageComponent />
  </div>
</template>
<script>
import MessageComponent from '@/components/MessageComponent.vue'
export default {
  mounted() {
    this.tokenAuth = this.$cookies.get('token') ?? ''
  },
  data() {
    return {
      tokenAuth: '',
      roomIdSelected: ''
    }
  },
  computed: {
    tokenAuthRest: function () {
      const res = this.tokenAuth.split('_')
      return res[0] + ' ' + res[1]
    }
  },
  name: 'App',
  components: {
    MessageComponent
  }
}
</script>

Que es el componente en Vue que se usa para conectarse al WS que hicimos antes desde FastAPI, pero en esta oportunidad lo hacemos desde un proyecto en Vue.

Ejecuta el proyecto con:

$ npm run dev

En el componente Messages.vue, verás que la lógica para conectar el WebSocket es casi idéntica a la que usamos antes, solo cambia el "envoltorio": definimos la dirección (ws://...) y el método se ejecuta automáticamente cuando el componente se monta (mounted):

src\components\MessageComponent.vue

<template>
  <div class="chat-container">
    <div class="messages-area" ref="messagesArea">
      <!-- <AlertsComponent ref="alertsComponent" @loaded="scrollToBottom" /> -->
      <p v-if="connecting">Connecting to chat...</p>
    </div>
    <div class="input-area">
      <textarea
        v-model="message"
        @keydown.enter.exact.prevent="send"
        placeholder="Type a message..."
        class="form-control"
        :disabled="connecting"
      ></textarea>
      <button @click="send" class="btn btn-primary" :disabled="connecting || message.trim() === ''">
        Send
      </button>
    </div>
  </div>
</template>
<script>
export default {
  components: {
  },
  data() {
    return {
      alertSocket: null,
      connecting: true
    }
  },
  mounted() {
    // Using setTimeout to ensure that root properties are available.
    // Consider a more robust state management solution (like Vuex or Pinia) in a larger app.
    setTimeout(this.websocketInit, 1000)
  },
  methods: {
    websocketInit() {
      const wsUrl = `ws://127.0.0.1:8000/ws/5`;
      this.alertSocket = new WebSocket(wsUrl);
      this.connecting = true;
      this.alertSocket.onopen = () => {
        console.log('WebSocket connection established.');
        this.connecting = false;
      };
      this.alertSocket.onmessage = (event) => {
        console.log('Message received:', event.data);
      };
      this.alertSocket.onclose = () => {
        console.log('WebSocket connection closed.');
        this.connecting = true; // Or handle reconnection logic
      };
      this.alertSocket.onerror = (error) => {
        console.error('WebSocket error:', error);
        this.connecting = false;
      };
    },
    send() {
      if (this.message.trim() !== '' && this.alertSocket && this.alertSocket.readyState === WebSocket.OPEN) {
        this.alertSocket.send(JSON.stringify({ message: this.message }));
        this.message = '';
        // We update getMessage to show our own message in the alerts component
        // this.$refs.alertsComponent.getAlerts();
      }
    },
  beforeDestroy() {
    if (this.alertSocket) {
      this.alertSocket.close();
    }
  }
}
</script>
<style scoped>
.chat-container {
  display: flex;
  flex-direction: column;
  height: 80vh; /* Example height, adjust as needed */
  border: 1px solid #ccc;
  border-radius: 8px;
  overflow: hidden;
}
.messages-area {
  flex-grow: 1;
  overflow-y: auto;
  padding: 15px;
  background-color: #f9f9f9;
}
.input-area {
  display: flex;
  padding: 10px;
  border-top: 1px solid #ccc;
  background-color: #fff;
  flex-shrink: 0;
}
.input-area textarea {
  flex-grow: 1;
  margin-right: 10px;
  resize: none; /* Prevents manual resizing which can break layout */
}
.input-area button {
  flex-shrink: 0;
}
p {
    text-align: center;
}
</style>

El Problema de los CORS

Si intentas conectar Vue con FastAPI sin configurar nada, verás en la consola del navegador un error característico de CORS. Esto ocurre porque el navegador bloquea las peticiones cuando el dominio (o el puerto) es distinto al del servidor:

INFO:     127.0.0.1:55956 - "WebSocket /ws/alert/room/" 403
INFO:     connection rejected (403 Forbidden)
INFO:     127.0.0.1:64104 - "OPTIONS /api/alerts HTTP/1.1" 404 Not Found
INFO:     connection closed

En mi caso, tuve que cambiar el puerto por defecto de Vue (5173) para las pruebas, ya que a veces los navegadores guardan una especie de "caché" de los permisos CORS y permiten la conexión incluso si los comentas en el código. Al cambiar de puerto, forzamos al navegador a validar la seguridad nuevamente.

Habilitando CORS en FastAPI

Para corregir esto y permitir que aplicaciones externas consuman nuestros recursos, debemos configurar un Middleware en FastAPI:

• Importar el paquete: Necesitamos CORSMiddleware.
• Definir los orígenes: Creamos una lista llamada origins con las URL permitidas (por ejemplo, la de nuestro proyecto en Vue).
• Agregar el Middleware: Lo inyectamos en la instancia de la aplicación.

api_multiple.py

from fastapi.middleware.cors import CORSMiddleware
***
# Define los orígenes permitidos (puedes usar ["*"] para permitir todo, 
# pero no es recomendable en producción)
origins = [
    "http://localhost:3000",
    "http://127.0.0.1:3000",
    "http://localhost:5173", # Puerto común de Vue/Vite
]
app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Al reiniciar el servidor con esta configuración, la conexión se establecerá correctamente. Podrás ver en la pestaña de Network del navegador cómo FastAPI acepta la conexión y empieza a recibir mensajes del servidor.

Implementación del Sistema de Autenticación (Login)

Vamos a implementar un sencillo sistema de autenticación. Para ello, crearemos un nuevo método que luego inyectaremos como una dependencia, tal como hemos visto a lo largo del curso. De momento, no utilizaremos bases de datos; manejaremos todo de forma local:

api_multiple.py

async def get_token(token: str = Query(...)): # REQUERIDO
    # Validación simple de token
    if token != "token-secreto":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Token inválido"
        )
    return token
# para conectar al WebSocket, el cliente 
# deberá incluir el parámetro token: ws://localhost:8000/ws/123?token=token-secreto
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int, token: str = Depends(get_token)):

El método recibirá un token. Recordemos que, al estar definido con "tres puntos" en nuestra lógica, es obligatorio y viene por parámetros en la URL. Aquí validamos el token; en un escenario real, este sería el lugar para consultar la base de datos, algo que ya hemos practicado y que puedes adaptar tú mismo. Por ahora, y para fines prácticos, el usuario deberá suministrar el valor "token-secreto".

Esto es netamente una demostración. Si el token es distinto al esperado, devolvemos una excepción de tipo 403 Forbidden indicando que el acceso está prohibido. Si es correcto, retornamos el token.

Pruebas de Conexión y Protección del WebSocket

Si regresamos a la aplicación en Vue, deberíamos recibir un error 403. Al revisar la pestaña de Network, verás que la conexión fue rechazada porque no estamos enviando el token todavía. Esta es, precisamente, la forma de proteger nuestro WebSocket.

Para solucionar esto, debemos pasar el parámetro vía GET (Query Parameter) al momento de conectarnos:

src/components/MessageComponent.vue

const wsUrl = `ws://localhost:8000/ws?token=token-secreto`;

Si el token es correcto, verás en el panel de Network que la conexión se establece exitosamente. Si colocas cualquier otra cosa, recibirás el error 403.

Aquí tienes el texto mejorado, organizado con títulos y corregido para una lectura más fluida, manteniendo tu estilo explicativo y técnico.

Implementación del Endpoint de Login

Vamos a implementar un endpoint sencillo para realizar el login. Inicialmente, la validación será local (hardcoded), aunque lo tradicional sería consultar una base de datos.

Para esto, utilizaremos un modelo de Pydantic que reciba username y password. La lógica de validación comparará si ambos campos son iguales a "admin". Si los datos son correctos, retornaremos el "super token secreto" que utilizaremos para las comprobaciones posteriores. En caso contrario, devolveremos un error 401 (Unauthorized) con el mensaje “Credenciales inválidas”:

src\components\LoginComponent.vue

<template>
    <div>
        <div class="row">
            <div class="col-md-6 offset-3 mt-3">
                <div class="card">
                    <div class="card-header">
                        Login
                    </div>
                    <div class="card-body">
                        <input type="text" v-model="username" class="form-control" placeholder="User">
                        <input type="password" v-model="password" class="form-control mt-3" placeholder="Password">
                        <button class="btn btn-success btn-sm mt-3" @click="send">Send</button>
                    </div>
                </div>
            </div>
        </div>
        
    </div>
</template>
<script>
export default {
    data() {
        return {
            username:'admin',
            password:'12345'
        }
    },
    methods: {
        send:function(){
            const data = {
                username:this.username.trim(),
                password:this.password.trim(),
            }
            this.$axios.post('http://127.0.0.1:8000/api/login',data).then(
                (res) => {
                    console.log(res.data)
                    this.$root.tokenAuth=res.data.token
                    this.$cookies.set('token', this.$root.tokenAuth)
                }).catch((error) =>{
                    console.error(error)
                    // this.$cookies.set('token', "token-secreto")
                })
        }
    },
}
</script>

Configuración del Componente Logout en Vue

Antes de probar el login, debemos habilitar el componente de Logout en nuestra aplicación. Es un proceso sencillo:

Contamos con un botón que dispara la función logout:

src\components\LogoutComponent.vue

<template>
    <div class="d-flex justify-content-end my-2 me-3">
        <button class="btn btn-danger" @click="logout">Logout</button>
    </div>
</template>
<script>
export default {
    methods: {
        logout() {
            this.$axios.post('http://127.0.0.1:8000/api/logout', {
                'token': this.$root.tokenAuth
            }).then(
                (res) => {
                    this.$root.tokenAuth = ''
                    this.$cookies.set('token', '')
                }).catch((error) => {
                    console.error(error)
                    this.$root.tokenAuth = ''
                    this.$cookies.set('token', '')
                })
        }
    },
}
</script>

Esta función:

this.$cookies.set('token', '')

Se encarga de limpiar el token, tanto en la memoria de ejecución (estado local) como en las cookies.

Es importante manejar los errores: si el servidor falla o el recurso no existe, de igual manera eliminamos el token localmente. Esto garantiza que, desde la perspectiva del usuario, la sesión se cierre sin importar el estado del servidor.

Configurar el Logout al momento del login

Con el componente de Logout listo, vamos a cargarlo también al momento de hacer el login.

Para organizar los componentes de Login y Logout en la interfaz, utilizaremos la etiqueta <template> en lugar de un <div>:

src\App.vue

<template>
  <div class="container">
    
    <LoginComponent v-if="tokenAuth == ''" />
    <template v-else >
      <MessageComponent />
      <LogoutComponent />
    </template>
  </div>
</template>

¿Por qué? A diferencia de un <div>, la etiqueta <template> no se renderiza en el HTML final, lo que mantiene el DOM más limpio y elegante.

Pruebas de Conexión y Depuración

En el componente de Login, realizamos la petición mediante Axios:

this.$axios.post('http://127.0.0.1:8000/api/login',data).then(***)

Si ingresamos credenciales incorrectas, la consola nos mostrará el error 401. Una vez ingresamos "admin/admin", el servidor nos devuelve el token correctamente.

Conclusión

Al verificar la pestaña de Network, vemos que la conexión se establece y el token se almacena, permitiendo una sesión persistente. Lo más importante de esta clase no es la base de datos en sí, sino ver cómo se interconectan las aplicaciones: un frontend en Vue comunicándose con un backend en FastAPI. Como puedes ver, no es nada del otro mundo, es seguir la misma lógica de integración.

Creación de Modelos para el Chat con WebSockets

Ya que hemos practicado con WebSockets y simulado comportamientos de usuarios mediante el Client ID (que en realidad representa el Room o habitación), es momento de formalizar la estructura. Vamos a crear los modelos de base de datos:

models.py

from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Table
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database import Base
import secrets
# Tabla intermedia para la relación ManyToMany entre Room y User
room_users = Table('room_users', Base.metadata,
    Column('room_id', Integer, ForeignKey('rooms.id')),
    Column('user_id', Integer, ForeignKey('users.id'))
)
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    password = Column(String)
    
    alerts = relationship("Alert", back_populates="user")
    rooms_joined = relationship("Room", secondary=room_users, back_populates="users")
    auth_token = relationship("Token", back_populates="user", uselist=False)
class Alert(Base):
    __tablename__ = "alerts"
    id = Column(Integer, primary_key=True, index=True)
    content = Column(String(200))
    user_id = Column(Integer, ForeignKey("users.id"))
    room_id = Column(Integer, ForeignKey("rooms.id"))
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    
    user = relationship("User", back_populates="alerts")
    room = relationship("Room", back_populates="alerts")
class Room(Base):
    __tablename__ = "rooms"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(60), unique=True)
    
    alerts = relationship("Alert", back_populates="room")
    users = relationship("User", secondary=room_users, back_populates="rooms_joined")
class Token(Base):
    __tablename__ = "tokens"
    # Generamos una key similar a DRF
    key = Column(String, primary_key=True, index=True, default=lambda: secrets.token_hex(20))
    user_id = Column(Integer, ForeignKey("users.id"))
    
    user = relationship("User", back_populates="auth_token")

Los esquemas:

schemas.py

from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
class UserBase(BaseModel):
    username: str
class User(UserBase):
    id: int
    class Config:
        from_attributes = True
class AlertBase(BaseModel):
    content: str
class Alert(AlertBase):
    id: int
    created_at: datetime
    user_id: int
    class Config:
        from_attributes = True
class RoomBase(BaseModel):
    name: str
class Room(RoomBase):
    id: int
    users: List[User] = []
    class Config:
        from_attributes = True
# Schemas para Request Body
class LoginRequest(BaseModel):
    username: str
    password: str
class LogoutRequest(BaseModel):
    token: str

Recuerda instalar SQL:

$ pip install SQLAlchemy

En Resumen:

1. La Tabla Pivot (Muchos a Muchos): room_users 

Comencemos con la relación entre usuarios y habitaciones. Una habitación puede tener múltiples usuarios y un usuario puede pertenecer a varias habitaciones.

Para manejar esta relación Many-to-Many, necesitamos una tabla intermedia (Pivot). El identificador que antes llamábamos Client ID ahora será formalmente una Room.

2. Modelo de Usuario e Identidad

El modelo de usuario (User) tendrá los campos básicos: id, username y password.

Relaciones: El usuario se relaciona con sus "Alertas" (que en este contexto son los mensajes del chat).

Tokens: También incluimos una relación One-to-One con una tabla de tokens para manejar la autenticación. He configurado uselist=False en SQLAlchemy para asegurar que cada token devuelva un único objeto y no una lista.

3. Gestión de Tokens

La tabla de tokens generará automáticamente una clave única mediante una función lambda utilizando secrets.token_hex. Esto es vital para que el servidor pueda validar quién intenta conectar a la habitación antes de darle permiso.

4. Habitaciones y Mensajes (Alertas)

Finalmente, definimos los modelos de Room y Message (llamado Alert en el código):

Habitaciones: Contienen la lista de usuarios (usando la tabla secundaria pivot) y la lista de mensajes asociados.

Mensajes: Guardan el contenido, la fecha de creación y la relación con el autor.

Nota de corrección: He corregido un detalle importante. Inicialmente, los mensajes no estaban asociados directamente a la habitación. He añadido el campo room_id y su respectiva Foreign Key (FK) para que, al cargar una habitación, el sistema pueda filtrar y mostrar solo los mensajes que pertenecen a ese grupo.

Implementación del API Router en FastAPI

Para mantener el orden, vamos a crear los recursos REST necesarios para las habitaciones y los mensajes. He configurado un APIRouter para no saturar el archivo principal de la aplicación.

1. Conexión a la Base de Datos

Utilizamos el esquema estándar de FastAPI para la inyección de dependencias. Implementamos una función con yield para obtener la sesión de la base de datos de forma eficiente; esto garantiza que la conexión se cierre automáticamente cuando la petición HTTP finalice como hicimos antes:

rest_api.py

from fastapi import APIRouter, Depends, HTTPException, status, Header
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from typing import List, Optional
import models
import schemas
from database import SessionLocal, engine
# Crear tablas en la BD (para propósitos de este ejemplo)
models.Base.metadata.create_all(bind=engine)
router = APIRouter()
# Dependencia de Base de Datos
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

2. Endpoints de Alertas y Habitaciones

Se definen dos rutas principales:

• Alertas (/alerts): Este método devuelve un listado de mensajes. Actualmente, realiza un query simple ordenado por fecha de creación. Más adelante, añadiremos un filtro por room_id para que solo devuelva los mensajes de una habitación específica.
• Habitaciones (/rooms): Devuelve el "pool" o listado de habitaciones disponibles. En el frontend, estas serán clickeables para permitir al usuario entrar en cada una.

rest_api.py

@router.get("/alerts", response_model=List[schemas.Alert])
def alerts(user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
    alerts = db.query(models.Alert).order_by(models.Alert.created_at).all()
    return alerts
@router.get("/rooms", response_model=List[schemas.Room])
def rooms(db: Session = Depends(get_db)):
    return db.query(models.Room).all()

3. Login

Lo primero es crear la ruta de tipo POST. Para el login, emplearemos el esquema LoginRequest de Pydantic, donde simplemente solicitaremos el username y el password.

Pasos de la implementación:

• Búsqueda: Inyectamos la base de datos y buscamos al usuario por su username.
• Validación de existencia: Si el usuario no existe, devolvemos un error.
• Nota de seguridad: Yo suelo devolver un mensaje genérico como "Usuario o contraseña incorrectos". Esto evita dar pistas a un posible atacante sobre si un nombre de usuario es válido o no, aunque al ser una demo puedes personalizarlo como prefieras.
• Verificación de contraseña: Si el usuario existe, verificamos la contraseña usando verify_password. mediante el paquete de bcrypt que instalamos al final de este apartado y que nos permite crear un hash de password (recordemos que un hash consiste en pocas palabras en convertir un conjunto de datos, un texto -password en este ejemplo- en una cadena alfanumérica única y que no se puede revertir para obtener el valor original, lo cual, es ideal para las contraseñas.

rest_api.py

import bcrypt
***
def verify_password(plain_password: str, hashed_password: str) -> bool:
    password_byte_enc = plain_password.encode('utf-8')
    hashed_password_enc = hashed_password.encode('utf-8')
    return bcrypt.checkpw(password_byte_enc, hashed_password_enc)
@router.post("/login")
def login(request: schemas.LoginRequest, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.username == request.username).first()
    
    if not user:
        return JSONResponse("User invalid", status_code=status.HTTP_401_UNAUTHORIZED)
    
    if not verify_password(request.password, user.password):
        return JSONResponse("Password invalid", status_code=status.HTTP_401_UNAUTHORIZED)
    
    # Get or Create Token
    token = db.query(models.Token).filter(models.Token.user_id == user.id).first()
    if not token:
        token = models.Token(user_id=user.id)
        db.add(token)
        db.commit()
        db.refresh(token)
        
    return f"Token_{token.key}"

Generación del Token

Una vez que las credenciales son válidas, procedemos a gestionar el token:

1. Buscamos en la tabla tokens si ya existe uno para ese usuario (manteniendo nuestra relación 1 a 1).
2. Si no existe, lo creamos, lo agregamos a la base de datos y hacemos el commit.
3. Finalmente, retornamos la clave (key) del token al cliente.

rest_api.py

@router.post("/login")
def login(request: schemas.LoginRequest, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.username == request.username).first()
    
    if not user:
        return JSONResponse("User invalid", status_code=status.HTTP_401_UNAUTHORIZED)
    
    if not verify_password(request.password, user.password):
        return JSONResponse("Password invalid", status_code=status.HTTP_401_UNAUTHORIZED)
    
    # Get or Create Token
    token = db.query(models.Token).filter(models.Token.user_id == user.id).first()
    if not token:
        token = models.Token(user_id=user.id)
        db.add(token)
        db.commit()
        db.refresh(token)
        
    return {"token": f"Token_{token.key}"}

schemas.py

class LoginRequest(BaseModel):
    username: str
    password: str

Recuerda instalar:

$ pip install bcrypt

4. Logout

El objetivo del Logout es eliminar el token de la base de datos para invalidar la sesión.

A diferencia del login, aquí el esquema no pide usuario y contraseña, sino el token directamente. Generalmente, el token llega en un formato específico (tipo Bearer), por lo que realizamos un split para separar la palabra "Bearer" de la clave real.

Lógica del proceso:

• Verificamos que el formato sea correcto (que el split nos devuelva exactamente dos partes).
• Filtramos en la base de datos para encontrar ese token.
• Eliminación: Si el token es válido y existe, lo eliminamos de la tabla y hacemos el commit.
• Respuesta: Retornamos un mensaje de "OK". Si el token no existía, también retornamos "OK", ya que, de cualquier forma, el usuario ya no está autenticado en el sistema.

rest_api.py

@router.post("/logout")
def logout(request: schemas.LogoutRequest, db: Session = Depends(get_db)):
    token_str = request.token
    if len(token_str.split('_')) == 2:
        token_name, token_key = token_str.split('_')
        if token_name == 'Token':
            token = db.query(models.Token).filter(models.Token.key == token_key).first()
            if token:
                db.delete(token)
                db.commit()
    return "ok"

schemas.py

class LogoutRequest(BaseModel):
    token: str

Implementación del Endpoint de Registro

Para probar los procesos de Login y Logout, necesitamos contar con usuarios en nuestro sistema. Por ello, vamos a crear un endpoint de registro (/register).

1. Lógica en el Backend

Utilizaremos una ruta de tipo POST que devolverá un estatus 201 Created. El esquema de datos será muy similar al de login (username y password).

El flujo de registro sigue estos pasos:

• Verificación: Buscamos en la base de datos si el username ya existe. Si es así, lanzamos una excepción indicando que el nombre de usuario ya ha sido tomado.
• Hashing de contraseña: Si el usuario es nuevo, convertimos la contraseña en un hash seguro antes de guardarla.
• Persistencia: Creamos la entidad, la guardamos en la base de datos y retornamos un mensaje de éxito.

rest_api.py

import bcrypt
def get_password_hash(password: str) -> str:
    # Convertimos la contraseña a bytes
    pwd_bytes = password.encode('utf-8')
    # Generamos el salt y el hash
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(pwd_bytes, salt)
    # Retornamos como string para guardar en la BD
    return hashed_password.decode('utf-8')
    
@router.post("/register", status_code=status.HTTP_201_CREATED)
def register(request: schemas.RegisterRequest, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.username == request.username).first()
    if user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered")
    
    hashed_password = get_password_hash(request.password)
    new_user = models.User(username=request.username, password=hashed_password)
    db.add(new_user)
    db.commit()
    return {"message": "User created successfully"}

schemas.py

class RegisterRequest(BaseModel):
    username: str
    password: str

Interfaz en Vue.js

En el frontend, he adaptado el componente de Login para crear uno de Registro. Para mantener el ejercicio sencillo y no complicarlo con vue-router, he implementado una lógica de intercambio de componentes basada en una variable booleana (showRegister).

• Componente Padre: Maneja el estado y decide si mostrar el formulario de acceso o el de registro.
• Componente Hijo (Registro): Emite un evento personalizado al padre tras un registro exitoso para que la interfaz oculte el formulario de registro y vuelva al login automáticamente.

src/components/RegisterComponent.vue

<template>
    <div>
        <div class="row">
            <div class="col-md-6 offset-3 mt-3">
                <div class="card">
                    <div class="card-header">
                        Register
                    </div>
                    <div class="card-body">
                        <input type="text" v-model="username" class="form-control" placeholder="User">
                        <input type="password" v-model="password" class="form-control mt-3" placeholder="Password">
                        <button class="btn btn-success btn-sm mt-3" @click="send">Register</button>
                        <button class="btn btn-link btn-sm mt-3" @click="$emit('show-login')">Login</button>
                    </div>
                </div>
            </div>
        </div>
        
    </div>
</template>
<script>
export default {
    data() {
        return {
            username:'',
            password:''
        }
    },
    methods: {
        send:function(){
            const data = {
                username:this.username.trim(),
                password:this.password.trim(),
            }
            this.$axios.post('http://127.0.0.1:8000/api/register',data).then(
                (res) => {
                    console.log(res.data)
                    this.$emit('show-login')
                }).catch((error) =>{
                    console.error(error)
                    // this.$cookies.set('token', "token-secreto")
                })
        }
    },
}
</script>

src/components/LoginComponent.vue

<template>
    <div>
        <div class="row" v-if="!showRegister">
            *** Login Form
        </div>
        <RegisterComponent v-else @show-login="showRegister = false" />
    </div>
</template>
<script>
import RegisterComponent from './RegisterComponent.vue'
export default {
    components: {
        RegisterComponent
    },
    data() {
        return {
            ***
            showRegister: false
        }
    },
    ***
}
</script>

Pruebas de Registro, Login y Logout: FastAPI + Vue

Vamos a revisar los puntos más importantes de los módulos de Login y Logout. Estos son los pilares de nuestro desarrollo y es fundamental que verifiques que tu configuración coincida para poder usar consumir la app en FastAPI desde Vue.

1. Configuración de la API y Rutas

Estamos empleando un esquema de API múltiple. Es importante notar que, aunque tenemos configurado un Websocket, este permite una sola conexión; es decir, por ahora no podemos conectar a varios usuarios en una misma habitación.

Respecto a las rutas:

Estamos utilizando el prefijo /api para ser consistentes con la configuración que tenemos en el frontend (Vue.js).

El enrutador principal (API router) gestiona los endpoints de Login y Logout.

Para acceder a ellos, las rutas finales son: api/login y api/logout:

app.include_router(api_router, prefix="/api")

2. Gestión de Usuarios

Creas un usuario mediante el registro y con esos datos intentas autentncarte mediante el componente de login.

3. Seguridad

Al generar el token, anexamos la palabra token antes del string generado:

return {"token": f"Token_{token.key}"}

Esta es una convención similar a la que usan frameworks como Laravel o Django (que usan los prefijos Bearer o Token).

He implementado esto como una medida de protección básica. Al recibir la petición de Logout, el sistema verifica que el string tenga esta estructura antes de procesarlo. Si no la tiene, ni siquiera intentamos la operación, evitando así posibles errores o problemas de procesamiento.

Ejecución de las Pruebas

• Iniciamos la aplicación (puede tardar un poco por la conexión inicial del Websocket).
• Ingresamos las credenciales del usuario.
• Al hacer clic en Sign In, vemos en la consola que los datos se pasan correctamente.
• El token se genera (o se recupera si ya existía uno válido) y la cookie queda configurada en el navegador.

Prueba de Logout

El flujo de cierre de sesión es el siguiente:

• Hacemos clic en Logout.
• La aplicación elimina el token local y nos redirige automáticamente a la pantalla de Login.
• Si intentamos recargar o navegar sin el token, el sistema nos deniega el acceso. El flujo funciona perfectamente.

Revisa en la base de datos cuando generas el token y cuando se elimina mediante el logout y vea que todo funciona a la perfección.

Siguiente paso, aprende a crear tu propio AutoCRUD para tus modelos en FastAPI.

Código fuente:

https://github.com/libredesarrollo/curso-libro-django-vue-channels

https://github.com/libredesarrollo/fastapi-websockets