es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Django programa/ejecuta tareas dinámicamente en un momento específico periódicamente a partir del pedido del usuario en la aplicación (sin celery)

Estoy creando una aplicación de Django en la cual el usuario puede programar algunas tareas para que ocurran en un momento específico.

Por ejemplo, en Google Calendar le decimos a Google lo que haremos mañana y luego, en el momento indicado, nos envía una notificación. De manera similar, aquí quiero que sea flexible, por ejemplo, que el usuario le indique a Django la hora y la función a ejecutar. Django esperará hasta la hora indicada para luego ejecutar la función.

Por ejemplo, si el usuario dice “apagar luces a las 12 pm”, entonces Django lo hará. O por ejemplo:

El usuario dice “recuérdame ir al gimnasio en 30 minutos”, y luego, después de 30 minutos, recibe una notificación.

En realidad, las tareas se agregan de forma dinámica, por lo que no podemos codificarlas de antemano.

Código:

skills.py # todas las funciones definidas por el usuario se importan aquí
# por ejemplo, toma esta tarea
def apagar_luz(habitación, id, *args, **kwargs):
    # apagar la luz usando python (tu lógica)
    print(f"¡La luz {id} ha sido apagada!")

Hay un modelo en Django conocido como Function en el cual esta función se almacena y el usuario puede acceder a ella fácilmente.

Quiero que los usuarios seleccionen esta función dentro de la aplicación y también proporcionen parámetros y hora, ¡y Django la ejecutará en ese momento!

En resumen, lo que necesito es que el usuario, desde dentro de la aplicación, pueda establecer la hora en la que se ejecutará una función (o tarea) en un momento específico (tal vez periódicamente o una vez o en una situación) y Django la ejecutará en ese momento determinado.

Nota: el usuario también proporcionará argumentos (args) y argumentos con nombre (kwargs) (me encargaré de eso). Lo que necesito es que Django ejecute la función con esos argumentos y argumentos con nombre.

(Solo se apreciará utilizar métodos de Django sin algo como Celery)

¡Gracias!

Tags:  , , ,

Answer

  1. Avatar for davy.ai

    Aplicación de Django para programar tareas

    Introducción

    En este proyecto, estamos creando una aplicación de Django en la cual los usuarios pueden programar tareas para que ocurran en un momento específico. El usuario puede seleccionar una función de una lista de funciones definidas por el usuario y también proporcionar argumentos y argumentos de palabras clave a esa función. Django esperará hasta el momento especificado y luego ejecutará la función con los argumentos y argumentos de palabras clave dados.

    Pasos

    Definir la función y almacenarla en un modelo de Django

    Definimos una función en un módulo separado de Python, digamos skills.py, que contiene todas las funciones a ser ejecutadas por Django. Por ejemplo, podemos definir una función en skills.py para apagar la luz.

    # skills.py
    
    def apagar_luz(habitacion, id, *args, **kwargs):
        # Apagamos la luz usando Python (aquí va tu lógica)
        print(f"¡La luz número {id} se ha apagado en {habitacion}!")
    

    Creamos un modelo de Django para almacenar estas funciones. El modelo se llama Funcion, donde name es el nombre de la función, module es el nombre del módulo de Python donde se define la función (en nuestro caso, skills), y args y kwargs son los argumentos por defecto y argumentos de palabras clave para la función, respectivamente.

    # models.py
    
    from django.db import models
    
    
    class Funcion(models.Model):
        name = models.CharField(max_length=100, unique=True)
        module = models.CharField(max_length=100, default="skills")
        args = models.JSONField(default=list)
        kwargs = models.JSONField(default=dict)
    
        def __str__(self):
            return self.name
    

    Crear vistas y formularios de Django

    Creamos un formulario de Django para permitir a los usuarios seleccionar una función, dar valores para los argumentos y argumentos de palabras clave, y seleccionar la fecha y hora en que la función debe ser ejecutada. El formulario se llama FormularioTarea y se define en forms.py.

    # forms.py
    
    from django import forms
    from django.forms.widgets import DateTimeInput
    from .models import Funcion
    
    
    class FormularioTarea(forms.Form):
        funcion = forms.ModelChoiceField(queryset=Funcion.objects.all())
        args = forms.JSONField(required=False)
        kwargs = forms.JSONField(required=False)
        hora_programada = forms.DateTimeField(widget=DateTimeInput(attrs={'type': 'datetime-local'}),
                                              input_formats=["%Y-%m-%dT%H:%M"])
    

    Luego, creamos una vista para manejar las peticiones GET para mostrar el formulario y las peticiones POST para agregar una nueva tarea. La vista se llama agregar_tarea y se define en views.py.

    # views.py
    
    from django.shortcuts import render
    from django.http import JsonResponse
    
    from .forms import FormularioTarea
    from .models import Funcion
    
    
    def agregar_tarea(request):
        if request.method == "POST":
            formulario = FormularioTarea(request.POST)
            if formulario.is_valid():
                funcion = formulario.cleaned_data['funcion']
                args = formulario.cleaned_data['args'] or funcion.args
                kwargs = formulario.cleaned_data['kwargs'] or funcion.kwargs
                hora_programada = formulario.cleaned_data['hora_programada']
    
                # programar la tarea
                # (escribe aquí tu lógica de programación)
    
                return JsonResponse({"estado": "ok"})
            else:
                return JsonResponse({"estado": "error", "error": formulario.errors})
        else:
            formulario = FormularioTarea()
            funciones = Funcion.objects.all()
            return render(request, "agregar_tarea.html", {"formulario": formulario, "funciones": funciones})
    

    La vista verifica si el método de la petición es POST. Si lo es, crea una instancia de FormularioTarea a partir de los datos de la petición. Si el formulario es válido, extrae la función seleccionada, los argumentos y los argumentos de palabras clave, y la hora programada de los datos del formulario, y programa la tarea (discutiremos la lógica de programación más adelante). Finalmente, devuelve una respuesta JSON con estado "ok" si la tarea fue programada correctamente, o una respuesta JSON con estado "error" y los errores del formulario si el formulario no era válido.

    Si el método de la petición es GET, la vista crea una instancia vacía de FormularioTarea y un queryset de todas las funciones disponibles, y renderiza la plantilla agregar_tarea.html con el formulario y el queryset.

    Crear plantillas HTML

    Creamos una plantilla HTML para el formulario FormularioTarea. La plantilla se llama agregar_tarea.html y se define de la siguiente manera.

    <!-- agregar_tarea.html -->
    {% extends "base.html" %}
    
    {% block content %}
    <h2>Agregar nueva tarea</h2>
    
    <form id="formulario-tarea" method="POST">
        {% csrf_token %}
        {{ formulario.as_p }}
        <input type="submit" value="Agregar tarea">
    </form>
    
    <script>
    $(function() {
        $('input[type=datetime-local]').flatpickr({
            enableTime: true,
            dateFormat: "Y-m-d H:i"
        });
    });
    </script>
    {% endblock %}
    

    La plantilla extiende una plantilla base (no mostrada aquí) y bloquea el content. Crea un formulario con una entrada datetime-local para la hora programada, y un botón de envío. El formulario usa el método POST para enviar datos a la vista agregar_tarea.

    También necesitamos crear una plantilla success.html y una plantilla error.html para mostrar un mensaje de éxito o de error después de que el usuario envíe el formulario.

    <!-- success.html -->
    {% extends "base.html" %}
    
    {% block content %}
    <h2>Tarea programada exitosamente</h2>
    Tu tarea ha sido programada exitosamente.
    <a href="{% url 'agregar_tarea' %}">Agregar otra tarea</a>
    {% endblock %}
    
    <!-- error.html -->
    {% extends "base.html" %}
    
    {% block content %}
    <h2>Error en la programación de la tarea</h2>
    <ul>
    {% for error in errores %}
      <li>{{ error }}</li>
    {% endfor %}
    </ul>
    <a href="{% url 'agregar_tarea' %}">Intentar nuevamente</a>
    {% endblock %}
    

    Programar la tarea

    Ahora llegamos a la parte central del proyecto: programar la tarea. En Django, podemos usar el paquete incorporado django_apscheduler, que es un envoltorio ligero alrededor del popular paquete APScheduler. django_apscheduler proporciona una forma sencilla de programar tareas en Django sin necesidad de una cola de tareas externa como Celery.

    Primero, instalamos el paquete django_apscheduler.

    pip install django_apscheduler
    

    Luego, agregamos apscheduler y django_apscheduler a la configuración INSTALLED_APPS en settings.py.

    # settings.py
    
    INSTALLED_APPS = [
        # ...
        'apscheduler',
        'django_apscheduler',
        # ...
    ]
    

    Después, agregamos la configuración de apscheduler a settings.py. Podemos configurar el planificador para usar una base de datos o una base de datos en memoria. Aquí, utilizamos la base de datos.

    # settings.py
    
    APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a"  # formato predeterminado
    APSCHEDULER_RUN_NOW = True  # ejecutar tareas perdidas al iniciar el servidor
    APSCHEDULER_TIMEZONE = "UTC"  # zona horaria para las tareas programadas
    APSCHEDULER_JOBSTORES = {
        "default": {"type": "django"},
    }
    APSCHEDULER_EXECUTORS = {
        "default": {"class": "apscheduler.executors.pool:ThreadPoolExecutor", "max_workers": 10},
    }
    APSCHEDULER_JOB_DEFAULTS = {
        "coalesce": True,  # ejecutar solo una instancia de la tarea a la vez
        "max_instances": 1,  # permitir solo una instancia de la tarea
    }
    

    También agregamos la configuración de URL de django_apscheduler a urls.py.

    # urls.py
    
    urlpatterns = [
        # ...
        path('django-apscheduler/', include('django_apscheduler.urls')),
        # ...
    ]
    

    Finalmente, modificamos la vista agregar_tarea para programar la tarea cuando se envía el formulario.

    “`python

    views.py

    import datetime
    from django_apscheduler.jobstores import DjangoJobStore
    from django_apscheduler.models import DjangoJobExecution

    def agregar_tarea(request):
    if request.method == “POST”:

Comments are closed.