Header Ads Widget

Ticker

6/recent/ticker-posts

Cree una API de aplicación de chat con Python y GraphQL


En un artículo anterior , cubrimos cómo crear una API básica usando GraphQL y Python. En este mini-proyecto a continuación, usaremos una característica de GraphQL llamada suscripciones para crear una aplicación de chat funcional.

¿Qué es la suscripción GraphQL?

Las suscripciones es una funcionalidad de GraphQL que permite al servidor GraphQL enviar datos al cliente cada vez que ocurre un evento específico. Aquí, las suscripciones permanecen en el estado conectado con el cliente, eliminando el ciclo de solicitud-respuesta de la API.

Las suscripciones a GraphQL le permiten crear aplicaciones en tiempo real utilizando API GraphQL que devuelven solo los datos que el cliente desea. Es algo similar a los websockets . Las suscripciones se basan en eventos , lo que significa que si hay actualizaciones en el lado del servidor, el cliente recibe una notificación automática sobre la actualización.

Ahora que tenemos un conocimiento fundamental de las suscripciones, centrémonos en nuestro mini-proyecto:

Prerrequisitos:

  • Pitón
  • Conocimientos básicos de GraphQL

Paso –1: Configuración del entorno

Abra su terminal y cree una nueva carpeta usando el siguiente comando:

mkdir chatql

Luego abre la carpeta:

cd chatql

Ahora, una vez que esté en la carpeta, inicialicemos un entorno virtual.

virtualenv .

Ahora actívala:

source bin/activate

Ahora instalemos algunas dependencias. Entonces, ejecute el siguiente comando:

pip3 install ariadne "uvicorn[standard]"

Ahora hemos terminado con la configuración del entorno. Procedamos con los siguientes pasos.

Paso –2: definir mutaciones

Cree un nombre de archivo mutations.pyy pegue el siguiente código:


from ariadne import ObjectType, convert_kwargs_to_snake_case

from store import users, messages, queues

mutation = ObjectType("Mutation")


@mutation.field("createUser")
@convert_kwargs_to_snake_case
async def resolve_create_user(obj, info, username):
    try:
        if not users.get(username):
            user = {
                "user_id": len(users) + 1,
                "username": username
            }
            users[username] = user
            return {
                "success": True,
                "user": user
            }
        return {
            "success": False,
            "errors": ["Username is taken"]
        }

    except Exception as error:
        return {
            "success": False,
            "errors": [str(error)]
        }

@mutation.field("createMessage")
@convert_kwargs_to_snake_case
async def resolve_create_message(obj, info, content, sender_id, recipient_id):
    try:
        message = {
            "content": content,
            "sender_id": sender_id,
            "recipient_id": recipient_id
        }
        messages.append(message)
        for queue in queues:
            await queue.put(message)
        return {
            "success": True,
            "message": message
        }
    except Exception as error:
        return {
            "success": False,
            "errors": [str(error)]
        }

Explicación del código:

Hemos definido dos mutaciones: createUsercreateMessageComo sugieren los nombres, createUsercrea usuarios mientras createMessagecrea mensajes. La createUsertoma un parámetro, el nombre de usuario, y devuelve el identificador del usuario. Si el usuario ya existe, devuelve un error. La createMessagemutación toma el mensaje sender_id, y recipient_id.

Paso –3: definir el esquema

Cree un archivo con nombre schema.graphqly pegue el siguiente código:


type User {
    username: String
    userId: String
}

type Message {
    content: String
    senderId: String
    recipientId: String
}

type createUserResult {
    user: User
    success: Boolean!
    errors: [String]
}

type createMessageResult {
    message: Message
    success: Boolean!
    errors: [String]
}

type messagesResult {
    messages: [Message]
    success: Boolean!
    errors: [String]
}

type Query {
    hello: String!
    messages(userId: String!): messagesResult
    userId(username: String!): String
}

type Mutation {
    createUser(username: String!): createUserResult
    createMessage(senderId: String, recipientId: String, content: String): createMessageResult
}


type Subscription {
    messages(userId: String): Message
}

schema {
  query: Query
  mutation: Mutation
  subscription: Subscription
}

Explicación del código:

Hemos definido el tipo UserMessagecreateUserResultcreateMessageResult, y messagesResultEl Usertiene tipo de datos definido usernameuserIdcomo cadena. El Messagetiene tipos de datos definidos como contentsenderIdrecipientIdcomo cadena.

También tenemos una consulta, mutación y suscripción, que se utilizarán para resolver la consulta y habilitar la suscripción.

Paso –4: Definición de suscripciones

Cree un nuevo archivo subscriptions.pyy pegue el siguiente código:

import asyncio
from ariadne import convert_kwargs_to_snake_case, SubscriptionType

from store import queues

subscription = SubscriptionType()


@subscription.source("messages")
@convert_kwargs_to_snake_case
async def messages_source(obj, info, user_id):
    queue = asyncio.Queue()
    queues.append(queue)
    try:
        while True:
            print('listen')
            message = await queue.get()
            queue.task_done()
            if message["recipient_id"] == user_id:
                yield message
    except asyncio.CancelledError:
        queues.remove(queue)
        raise

@subscription.field("messages")
@convert_kwargs_to_snake_case
async def messages_resolver(message, info, user_id):
    return message

Explicación del código:

El sistema de mensajería aquí seguirá el primero en entrar, primero en salir (FIFO). Esto significa que el primer mensaje que llegue a la cola se enviará al cliente. Hay un bucle infinito en ejecución, que buscará mensajes nuevos infinitamente await queue.get().

Paso –5: Definición de consultas

Cree un nuevo archivo queries.pyy pegue el siguiente código:

from ariadne import ObjectType, convert_kwargs_to_snake_case

from store import messages, users

query = ObjectType("Query")


@query.field("messages")
@convert_kwargs_to_snake_case
async def resolve_messages(obj, info, user_id):
    def filter_by_userid(message):
        return message["sender_id"] == user_id or \
               message["recipient_id"] == user_id

    user_messages = filter(filter_by_userid, messages)
    return {
        "success": True,
        "messages": user_messages
    }

@query.field("userId")
@convert_kwargs_to_snake_case
async def resolve_user_id(obj, info, username):
    user = users.get(username)
    if user:
        return user["user_id"]

Explicación del código:

Estamos importando ariadneaquí, que usaremos para facilitar la implementación de GraphQL. Luego, hemos definido dos campos de consulta, messagesuserId, que buscarán los mensajes para una identificación de usuario y detalles de usuario como nombre de usuario, respectivamente.

También importamos un storemódulo, que se usa para almacenar mensajes. Hablaremos de este módulo en los próximos pasos.

Paso –6: Definición del módulo store.pyapp.py:

Este es el último paso de la configuración. Así que crea un archivo store.pyy pega el siguiente código:

users = dict()
messages = []
queues = []

Explicación del código:

Hemos creado un diccionario de usuarios, una lista de mensajes y colas.

Ahora cree otro archivo, app.pyy pegue el siguiente código:

from ariadne import make_executable_schema, load_schema_from_path, \
    snake_case_fallback_resolvers
from ariadne.asgi import GraphQL
from mutations import mutation
from queries import query
from subscriptions import subscription


type_defs = load_schema_from_path("schema.graphql")

schema = make_executable_schema(type_defs, query, mutation, subscription,
                                snake_case_fallback_resolvers)
app = GraphQL(schema, debug=True)

Explicación del código:

Estamos importando todos los módulos que creamos, es decir, mutaciones, consultas, suscripciones y también ariadneEl código se está cargando schema.graphqly almacenando type_defsLuego también estamos llamando make_executable_schema()y pasando mutaciones, consultas, suscripciones, resolutores y type_defsEsta función se utiliza para crear un esquema ejecutable.

Ahora es el momento de probar la API.

Paso –7: prueba de la API

Para ejecutar el servidor, use el siguiente comando:

uvicorn app:app --reload

Ahora abra su navegador web y diríjase a URL: y pegue el siguiente código en el panel izquierdo de la página para crear un nuevo usuario:http://127.0.0.1:8000/

mutation {
createUser(username:"YOUR_NAME_1_HERE") {
success
user {
userId
username
}
}
}

Estamos creando un usuario aquí. Simplemente reemplace el marcador de posición YOUR_NAME_1_HEREcon un nombre de usuario único. Asegúrate de crear dos usuarios porque tienes que enviar un mensaje a alguien, ¿verdad? Una vez que ejecute la consulta anterior, devolverá algo como esto:

{
"data": {
"createUser": {
"success": true,
"user": {
"userId": "1",
"username": "YOUR_NAME_1_HERE"
}
}
}
}

Asegúrese de anotar los datos userIdde ambos usuarios; necesitaremos estos ID para enviar mensajes. Ahora abra una pestaña más en su navegador e ir a URL: Verá la misma página, ahora pegue el siguiente código para suscribirse a los mensajes:http://localhost:8000/

subscription {
messages(userId: "YOUR SECOND USER’s ID HERE") {
content
senderId
recipientId
}
}

Si todo está bien, debería ver algo como esto:

¿Puedes ver el texto Escuchando " en la parte inferior de la pantalla? Significa que se ha suscrito a los mensajes.

Ahora regrese a la pestaña anterior donde abrió la URL: y en el panel izquierdo reemplace el código existente con el siguiente código:http://127.0.0.1:8000/

mutation {
createMessage(
senderId: "YOUR FIRST USER’s ID HERE",
recipientId: "YOUR SECOND USER’s ID HERE",
content:"YOUR MESSAGE HERE"
) {
success
message {
content
recipientId
senderId
}
}
}

Simplemente reemplace los marcadores de posición con las ID correctas y el mensaje que desea enviar, y presione el botón de reproducción. En la segunda pestaña, ahora debería poder recibir los mensajes. ¡Eso es! Tu API de chat está lista ahora.

Ultimas palabras:

Hemos cubierto una API de aplicación de chat básica en este artículo usando Python y GraphQL. Puede crear una interfaz de usuario usando un marco JS y hacerlo más interactivo, mientras que también puede usar JWT para hacerlo más seguro.

Publicar un comentario

0 Comentarios