Cómo implementar una API que ejecute un script de Python con los datos de la solicitud POST.

Quiero ejecutar un script de Python que básicamente monitorea cualquier cambio que ocurra en un directorio específico (el directorio a monitorear se pasa como parte de la solicitud POST). Cada vez que se llama a la API (estoy usando FastAPI), se debe iniciar una nueva instancia del script para monitorear ese directorio en particular y enviar un mensaje de “éxito” como respuesta si se inició el script correctamente. Además, estoy planeando agregar otro punto final de API que detendrá el script que se está ejecutando para vigilar un directorio.

¿Se pueden utilizar colas de mensajes como RQ o Celery para lograr esto? Tenga en cuenta que quiero que se inicien nuevos scripts cada vez que se llama a la API, por lo que varias instancias del script deberían ejecutarse al mismo tiempo. Estoy usando el módulo watchdog para monitorear el sistema de archivos.

No sé cómo hacer esto de la manera correcta, pero esto es lo que he ideado hasta ahora donde se crea un nuevo hilo para cada llamada a la API:

from fastapi import FastAPI from schemas import Data # modelo de esquema pydantic para la API from threading import Thread import filewatcher # el script que se debe ejecutar @app.post('/register/event') def register_watchdog(data: Data): th = Thread(target=filewatcher.create_watchdog, args=(data)) th.start() return {"status": "éxito"}

¿Cuál es la mejor manera de lograr esto? Una pregunta más es, ¿puedo implementar mi script como un servicio de Linux que se pueda ejecutar en segundo plano?

davy.ai

May 28, 2023 at 1:20 pm

Sí, se pueden utilizar sistemas de cola de mensajes como RQ y Celery para lograr esto. Permiten la ejecución de tareas asíncronas y pueden manejar varias instancias del script ejecutándose al mismo tiempo.

Para implementar esto usando Celery, es necesario definir una tarea de Celery que llame al script filewatcher y monitoree el directorio especificado. Cuando se llame a la API, se deberá agregar una nueva tarea a la cola de Celery para comenzar a monitorear el directorio. Para detener el script, se podría definir otra tarea de Celery que detenga la tarea especificada utilizando su ID.

Aquí hay una implementación de ejemplo utilizando Celery:

from fastapi import FastAPI
from schemas import Data
from celery import Celery
import filewatcher

app = FastAPI()
celery = Celery('tasks', broker='redis://localhost:6379') # usando Redis como broker

@celery.task
def start_watchdog(data: Data):
    filewatcher.create_watchdog(data)

@celery.task
def stop_watchdog(task_id):
    celery.control.revoke(task_id)

@app.post('/register/event')
def register_watchdog(data: Data):
    task = start_watchdog.delay(data)
    return {"status": "success", "task_id": task.id}

@app.post('/stop/event/{task_id}')
def stop_watchdog(task_id):
    stop_watchdog.delay(task_id)
    return {"status": "success"}

En esta implementación, se crea una nueva tarea de Celery para comenzar a monitorear el directorio especificado cada vez que se llama a la API. La tarea start_watchdog se llama de forma asíncrona y su ID se devuelve como parte de la respuesta de la API. Para detener la tarea, se puede llamar a la tarea stop_watchdog con el ID de la tarea que se desea detener.

Con respecto a la implementación del script como un servicio de Linux, se puede utilizar un sistema de control de procesos como systemd para ejecutar el script en segundo plano como un servicio. Es necesario definir un archivo de servicio de systemd que especifique el comando para ejecutar el script, así como cualquier dependencia o límite de recursos. Una vez que se crea y se habilita el archivo de servicio, el script se ejecutará como un servicio en segundo plano cada vez que el servidor se inicie.

Answer

davy.ai

May 28, 2023 at 1:20 pm
Sí, se pueden utilizar sistemas de cola de mensajes como RQ y Celery para lograr esto. Permiten la ejecución de tareas asíncronas y pueden manejar varias instancias del script ejecutándose al mismo tiempo.

Para implementar esto usando Celery, es necesario definir una tarea de Celery que llame al script filewatcher y monitoree el directorio especificado. Cuando se llame a la API, se deberá agregar una nueva tarea a la cola de Celery para comenzar a monitorear el directorio. Para detener el script, se podría definir otra tarea de Celery que detenga la tarea especificada utilizando su ID.

Aquí hay una implementación de ejemplo utilizando Celery:
```
from fastapi import FastAPI
from schemas import Data
from celery import Celery
import filewatcher

app = FastAPI()
celery = Celery('tasks', broker='redis://localhost:6379') # usando Redis como broker

@celery.task
def start_watchdog(data: Data):
    filewatcher.create_watchdog(data)

@celery.task
def stop_watchdog(task_id):
    celery.control.revoke(task_id)

@app.post('/register/event')
def register_watchdog(data: Data):
    task = start_watchdog.delay(data)
    return {"status": "success", "task_id": task.id}

@app.post('/stop/event/{task_id}')
def stop_watchdog(task_id):
    stop_watchdog.delay(task_id)
    return {"status": "success"}
```
En esta implementación, se crea una nueva tarea de Celery para comenzar a monitorear el directorio especificado cada vez que se llama a la API. La tarea start_watchdog se llama de forma asíncrona y su ID se devuelve como parte de la respuesta de la API. Para detener la tarea, se puede llamar a la tarea stop_watchdog con el ID de la tarea que se desea detener.

Con respecto a la implementación del script como un servicio de Linux, se puede utilizar un sistema de control de procesos como systemd para ejecutar el script en segundo plano como un servicio. Es necesario definir un archivo de servicio de systemd que especifique el comando para ejecutar el script, así como cualquier dependencia o límite de recursos. Una vez que se crea y se habilita el archivo de servicio, el script se ejecutará como un servicio en segundo plano cada vez que el servidor se inicie.

Comments are closed.

es.davy.ai

¿Tienes una pregunta?

Cómo implementar una API que ejecute un script de Python con los datos de la solicitud POST.

Answer

davy.ai